learningcircuit--local-deep-research/tests/security/test_data_sanitizer_coverage.py

"""
Comprehensive coverage tests for data_sanitizer.py.

Targets gaps not covered by existing test suites:
- Tuple and set inputs (non-list iterables)
- sanitize/redact with dict values that are lists of lists
- redact on deeply nested structures with custom keys + custom text combined
- filter_research_metadata with truthy/falsy edge values (0, empty string, empty list)
- strip_settings_snapshot preserving complex nested values
- Ensuring bool coercion in filter_research_metadata for various truthy types
- sanitize/redact on dicts where values are themselves sensitive-key dicts
- Mixed list containing nested lists and dicts
- redact with empty string redaction_text
- Large flat dict with only sensitive keys
- Key collision: same key different cases in one dict
- Convenience functions with all optional params
- strip_settings_snapshot with empty string input
- filter_research_metadata with empty string input
- Numeric and boolean JSON strings for metadata functions
"""

import json

import pytest

from local_deep_research.security.data_sanitizer import (
    DataSanitizer,
    filter_research_metadata,
    redact_data,
    sanitize_data,
    strip_settings_snapshot,
)


# ---------------------------------------------------------------------------
# sanitize() -- additional coverage
# ---------------------------------------------------------------------------


class TestSanitizeTupleAndUnusualTypes:
    """Sanitize behavior with tuple, set, and other non-list/dict types."""

    def test_tuple_returned_as_is(self):
        """Tuples are primitives from sanitize's perspective -- returned unchanged."""
        data = (1, 2, 3)
        result = DataSanitizer.sanitize(data)
        assert result == (1, 2, 3)

    def test_set_returned_as_is(self):
        """Sets are not traversed -- returned unchanged."""
        data = {1, 2, 3}
        result = DataSanitizer.sanitize(data)
        assert result == {1, 2, 3}

    def test_float_returned_as_is(self):
        """Float primitives pass through."""
        assert DataSanitizer.sanitize(3.14) == 3.14

    def test_bytes_returned_as_is(self):
        """Bytes are treated as a primitive."""
        data = b"hello"
        assert DataSanitizer.sanitize(data) == b"hello"

    def test_zero_returned_as_is(self):
        """Zero (falsy int) passes through unchanged."""
        assert DataSanitizer.sanitize(0) == 0

    def test_empty_string_returned_as_is(self):
        """Empty string (falsy) passes through unchanged."""
        assert DataSanitizer.sanitize("") == ""

    def test_false_returned_as_is(self):
        """False (falsy bool) passes through unchanged."""
        assert DataSanitizer.sanitize(False) is False


class TestSanitizeNestedListsOfLists:
    """Sanitize handles nested list structures."""

    def test_list_of_lists_with_dicts(self):
        """Nested list of lists containing dicts are sanitized."""
        data = [[{"api_key": "k1", "val": 1}], [{"password": "p1", "val": 2}]]
        result = DataSanitizer.sanitize(data)
        assert result == [[{"val": 1}], [{"val": 2}]]

    def test_list_of_mixed_nested(self):
        """Mixed list containing dicts, lists, and primitives."""
        data = [
            {"secret": "s", "ok": True},
            [{"access_token": "t"}, 42],
            "plain",
        ]
        result = DataSanitizer.sanitize(data)
        assert result[0] == {"ok": True}
        assert result[1] == [{}, 42]
        assert result[2] == "plain"


class TestSanitizeValueTypes:
    """Sanitize removes keys regardless of the value type."""

    def test_sensitive_key_with_list_value(self):
        """A sensitive key whose value is a list is still removed entirely."""
        data = {"api_key": [1, 2, 3], "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert "api_key" not in result
        assert result == {"name": "ok"}

    def test_sensitive_key_with_dict_value(self):
        """A sensitive key whose value is a dict is still removed entirely."""
        data = {"password": {"nested": "data"}, "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert "password" not in result
        assert result == {"name": "ok"}

    def test_sensitive_key_with_none_value(self):
        """A sensitive key with None value is still removed."""
        data = {"secret": None, "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert "secret" not in result

    def test_sensitive_key_with_bool_value(self):
        """A sensitive key with boolean value is still removed."""
        data = {"csrf_token": True, "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert "csrf_token" not in result

    def test_sensitive_key_with_numeric_value(self):
        """A sensitive key with numeric value is still removed."""
        data = {"auth_token": 12345, "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert "auth_token" not in result


class TestSanitizeAllKeySensitive:
    """When every key in a dict is sensitive."""

    def test_all_keys_removed_returns_empty_dict(self):
        data = {
            "api_key": "1",
            "password": "2",
            "secret": "3",
        }
        result = DataSanitizer.sanitize(data)
        assert result == {}

    def test_large_dict_all_sensitive(self):
        """All 10 default sensitive keys in one dict."""
        data = {
            k: f"val_{i}"
            for i, k in enumerate(DataSanitizer.DEFAULT_SENSITIVE_KEYS)
        }
        result = DataSanitizer.sanitize(data)
        assert result == {}


class TestSanitizeReturnTypeSame:
    """Sanitize returns the same container type."""

    def test_dict_returns_dict(self):
        result = DataSanitizer.sanitize({"a": 1})
        assert isinstance(result, dict)

    def test_list_returns_list(self):
        result = DataSanitizer.sanitize([1, 2])
        assert isinstance(result, list)


# ---------------------------------------------------------------------------
# redact() -- additional coverage
# ---------------------------------------------------------------------------


class TestRedactTupleAndUnusualTypes:
    """Redact behavior with non-list/dict types."""

    def test_tuple_returned_as_is(self):
        data = (1, 2, 3)
        assert DataSanitizer.redact(data) == (1, 2, 3)

    def test_set_returned_as_is(self):
        data = {1, 2, 3}
        assert DataSanitizer.redact(data) == {1, 2, 3}

    def test_bytes_returned_as_is(self):
        assert DataSanitizer.redact(b"data") == b"data"

    def test_zero_returned_as_is(self):
        assert DataSanitizer.redact(0) == 0

    def test_empty_string_returned_as_is(self):
        assert DataSanitizer.redact("") == ""


class TestRedactNestedListsOfLists:
    """Redact handles nested list structures."""

    def test_list_of_lists_with_dicts(self):
        data = [[{"api_key": "k1", "val": 1}], [{"password": "p1", "val": 2}]]
        result = DataSanitizer.redact(data)
        assert result == [
            [{"api_key": "[REDACTED]", "val": 1}],
            [{"password": "[REDACTED]", "val": 2}],
        ]


class TestRedactCombinedCustomKeysAndText:
    """Redact with both custom keys and custom redaction text."""

    def test_custom_keys_and_custom_text(self):
        data = {"my_secret": "val", "api_key": "keep", "name": "ok"}
        result = DataSanitizer.redact(
            data, sensitive_keys={"my_secret"}, redaction_text="<REMOVED>"
        )
        assert result["my_secret"] == "<REMOVED>"
        assert result["api_key"] == "keep"
        assert result["name"] == "ok"

    def test_custom_keys_and_empty_text(self):
        data = {"password": "keep_default", "custom": "hide"}
        result = DataSanitizer.redact(
            data, sensitive_keys={"custom"}, redaction_text=""
        )
        assert result["custom"] == ""
        assert result["password"] == "keep_default"


class TestRedactAllKeysSensitive:
    """When every key in a dict is sensitive, all values are redacted."""

    def test_all_keys_redacted(self):
        data = {"api_key": "1", "password": "2", "secret": "3"}
        result = DataSanitizer.redact(data)
        assert all(v == "[REDACTED]" for v in result.values())
        assert len(result) == 3

    def test_large_dict_all_redacted(self):
        data = {
            k: f"val_{i}"
            for i, k in enumerate(DataSanitizer.DEFAULT_SENSITIVE_KEYS)
        }
        result = DataSanitizer.redact(data)
        assert all(v == "[REDACTED]" for v in result.values())
        assert len(result) == len(DataSanitizer.DEFAULT_SENSITIVE_KEYS)


class TestRedactDeeplyNestedCustom:
    """Redact with custom keys in deeply nested structures."""

    def test_deeply_nested_custom_key(self):
        data = {"a": {"b": {"c": {"my_field": "secret", "val": 1}}}}
        result = DataSanitizer.redact(data, sensitive_keys={"my_field"})
        assert result["a"]["b"]["c"]["my_field"] == "[REDACTED]"
        assert result["a"]["b"]["c"]["val"] == 1

    def test_list_nested_in_dict_custom_key(self):
        data = {"items": [{"token": "t1", "id": 1}, {"token": "t2", "id": 2}]}
        result = DataSanitizer.redact(data, sensitive_keys={"token"})
        assert result["items"][0]["token"] == "[REDACTED]"
        assert result["items"][0]["id"] == 1
        assert result["items"][1]["token"] == "[REDACTED]"


class TestRedactDoesNotMutateNested:
    """Redact does not mutate nested original data."""

    def test_nested_dict_original_unchanged(self):
        inner = {"password": "secret", "val": 1}
        outer = {"config": inner}
        DataSanitizer.redact(outer)
        assert inner["password"] == "secret"

    def test_list_original_unchanged(self):
        item = {"api_key": "k", "id": 1}
        data = [item]
        DataSanitizer.redact(data)
        assert item["api_key"] == "k"


# ---------------------------------------------------------------------------
# Convenience functions -- additional coverage
# ---------------------------------------------------------------------------


class TestConvenienceFunctionsExtended:
    """Additional convenience function tests."""

    def test_sanitize_data_with_none_keys(self):
        """sanitize_data with explicit None uses defaults."""
        data = {"api_key": "k", "name": "n"}
        result = sanitize_data(data, sensitive_keys=None)
        assert "api_key" not in result
        assert result == {"name": "n"}

    def test_redact_data_with_none_keys(self):
        """redact_data with explicit None uses defaults."""
        data = {"password": "p", "name": "n"}
        result = redact_data(data, sensitive_keys=None)
        assert result["password"] == "[REDACTED]"
        assert result["name"] == "n"

    def test_redact_data_custom_keys_and_text(self):
        """redact_data with both custom keys and custom text."""
        data = {"my_field": "val", "api_key": "keep"}
        result = redact_data(
            data, sensitive_keys={"my_field"}, redaction_text="XXX"
        )
        assert result["my_field"] == "XXX"
        assert result["api_key"] == "keep"

    def test_sanitize_data_list_input(self):
        """sanitize_data works with list input."""
        data = [{"api_key": "k", "id": 1}]
        result = sanitize_data(data)
        assert result == [{"id": 1}]

    def test_redact_data_list_input(self):
        """redact_data works with list input."""
        data = [{"password": "p", "id": 1}]
        result = redact_data(data)
        assert result == [{"password": "[REDACTED]", "id": 1}]

    def test_sanitize_data_primitive(self):
        """sanitize_data passes through primitives."""
        assert sanitize_data("hello") == "hello"
        assert sanitize_data(42) == 42
        assert sanitize_data(None) is None

    def test_redact_data_primitive(self):
        """redact_data passes through primitives."""
        assert redact_data("hello") == "hello"
        assert redact_data(42) == 42
        assert redact_data(None) is None


# ---------------------------------------------------------------------------
# filter_research_metadata() -- additional coverage
# ---------------------------------------------------------------------------


class TestFilterResearchMetadataExtended:
    """Additional edge cases for filter_research_metadata."""

    def test_is_news_search_zero_coerced_to_false(self):
        """0 is falsy, coerced to False."""
        result = filter_research_metadata({"is_news_search": 0})
        assert result == {"is_news_search": False}
        assert type(result["is_news_search"]) is bool

    def test_is_news_search_empty_string_coerced_to_false(self):
        """Empty string is falsy, coerced to False."""
        result = filter_research_metadata({"is_news_search": ""})
        assert result == {"is_news_search": False}
        assert type(result["is_news_search"]) is bool

    def test_is_news_search_empty_list_coerced_to_false(self):
        """Empty list is falsy, coerced to False."""
        result = filter_research_metadata({"is_news_search": []})
        assert result == {"is_news_search": False}

    def test_is_news_search_nonempty_list_coerced_to_true(self):
        """Non-empty list is truthy, coerced to True."""
        result = filter_research_metadata({"is_news_search": [1]})
        assert result == {"is_news_search": True}

    def test_is_news_search_none_coerced_to_false(self):
        """None is falsy, coerced to False."""
        result = filter_research_metadata({"is_news_search": None})
        assert result == {"is_news_search": False}

    def test_empty_string_input(self):
        """Empty string input (falsy) returns default."""
        result = filter_research_metadata("")
        assert result == {"is_news_search": False}

    def test_json_string_with_extra_fields_stripped(self):
        """JSON string with extra fields returns only is_news_search."""
        meta_json = json.dumps(
            {
                "is_news_search": True,
                "settings_snapshot": {"api_key": "secret"},
                "phase": "complete",
            }
        )
        result = filter_research_metadata(meta_json)
        assert result == {"is_news_search": True}
        assert "settings_snapshot" not in result
        assert "phase" not in result

    def test_dict_with_only_settings_snapshot(self):
        """Dict with only settings_snapshot but no is_news_search."""
        result = filter_research_metadata({"settings_snapshot": {"key": "val"}})
        assert result == {"is_news_search": False}

    def test_boolean_input(self):
        """Boolean input (not a dict/str) returns default."""
        result = filter_research_metadata(True)
        assert result == {"is_news_search": False}

    def test_float_input(self):
        """Float input returns default."""
        result = filter_research_metadata(3.14)
        assert result == {"is_news_search": False}

    def test_list_input(self):
        """List input (non-dict) returns default."""
        result = filter_research_metadata([{"is_news_search": True}])
        assert result == {"is_news_search": False}

    def test_json_number_string(self):
        """JSON string that parses to a number returns default."""
        result = filter_research_metadata("42")
        assert result == {"is_news_search": False}

    def test_json_boolean_string(self):
        """JSON string that parses to a boolean returns default."""
        result = filter_research_metadata("true")
        assert result == {"is_news_search": False}

    def test_json_null_string(self):
        """JSON string 'null' parses to None, then `meta or {}` gives {}."""
        result = filter_research_metadata("null")
        assert result == {"is_news_search": False}


# ---------------------------------------------------------------------------
# strip_settings_snapshot() -- additional coverage
# ---------------------------------------------------------------------------


class TestStripSettingsSnapshotExtended:
    """Additional edge cases for strip_settings_snapshot."""

    def test_empty_string_input(self):
        """Empty string (falsy) returns empty dict."""
        result = strip_settings_snapshot("")
        assert result == {}

    def test_preserves_all_non_snapshot_keys(self):
        """All keys except settings_snapshot are preserved."""
        meta = {
            "phase": "complete",
            "error_type": None,
            "processed_query": "test query",
            "mode": "quick",
            "duration": 12.5,
            "is_news_search": True,
            "settings_snapshot": {"api_key": "secret"},
        }
        result = strip_settings_snapshot(meta)
        assert "settings_snapshot" not in result
        assert result["phase"] == "complete"
        assert result["error_type"] is None
        assert result["processed_query"] == "test query"
        assert result["mode"] == "quick"
        assert result["duration"] == 12.5
        assert result["is_news_search"] is True

    def test_preserves_nested_complex_values(self):
        """Nested complex values (lists, dicts) are preserved."""
        meta = {
            "results": [{"title": "a"}, {"title": "b"}],
            "config": {"depth": 3},
            "settings_snapshot": {},
        }
        result = strip_settings_snapshot(meta)
        assert result["results"] == [{"title": "a"}, {"title": "b"}]
        assert result["config"] == {"depth": 3}

    def test_boolean_input(self):
        """Boolean input returns empty dict."""
        assert strip_settings_snapshot(True) == {}
        assert strip_settings_snapshot(False) == {}

    def test_float_input(self):
        """Float input returns empty dict."""
        assert strip_settings_snapshot(3.14) == {}

    def test_list_input(self):
        """List input returns empty dict."""
        assert strip_settings_snapshot([1, 2, 3]) == {}

    def test_json_number_string(self):
        """JSON number string returns empty dict (not a dict)."""
        assert strip_settings_snapshot("42") == {}

    def test_json_null_string(self):
        """JSON 'null' string returns empty dict."""
        assert strip_settings_snapshot("null") == {}

    def test_json_boolean_string(self):
        """JSON 'true' string returns empty dict."""
        assert strip_settings_snapshot("true") == {}

    def test_json_array_string(self):
        """JSON array string returns empty dict."""
        assert strip_settings_snapshot("[1, 2]") == {}

    def test_does_not_strip_similar_key_names(self):
        """Keys like 'settings_snapshot_v2' are NOT stripped."""
        meta = {"settings_snapshot_v2": "data", "settings_snapshot": "secret"}
        result = strip_settings_snapshot(meta)
        assert "settings_snapshot_v2" in result
        assert "settings_snapshot" not in result

    def test_settings_snapshot_with_large_value(self):
        """Large settings_snapshot value is stripped cleanly."""
        snapshot = {f"key_{i}": f"value_{i}" for i in range(100)}
        meta = {"phase": "done", "settings_snapshot": snapshot}
        result = strip_settings_snapshot(meta)
        assert result == {"phase": "done"}

    def test_json_string_preserves_value_types(self):
        """JSON string input preserves value types after parsing."""
        meta_json = json.dumps(
            {
                "phase": "done",
                "count": 42,
                "active": True,
                "settings_snapshot": {},
            }
        )
        result = strip_settings_snapshot(meta_json)
        assert result == {"phase": "done", "count": 42, "active": True}

    def test_original_dict_not_mutated(self):
        """strip_settings_snapshot does not mutate the original dict."""
        meta = {"phase": "done", "settings_snapshot": {"api_key": "secret"}}
        strip_settings_snapshot(meta)
        assert "settings_snapshot" in meta


# ---------------------------------------------------------------------------
# DEFAULT_SENSITIVE_KEYS -- exhaustive membership
# ---------------------------------------------------------------------------


class TestDefaultSensitiveKeysComplete:
    """Verify exact membership of DEFAULT_SENSITIVE_KEYS."""

    def test_exact_set(self):
        expected = {
            "api_key",
            "apikey",
            "password",
            "secret",
            "access_token",
            "refresh_token",
            "private_key",
            "auth_token",
            "session_token",
            "csrf_token",
            "client_secret",
            "secret_key",
            "bearer_token",
            "api_secret",
            "app_secret",
        }
        assert DataSanitizer.DEFAULT_SENSITIVE_KEYS == expected

    def test_count(self):
        assert len(DataSanitizer.DEFAULT_SENSITIVE_KEYS) == 15

    def test_all_lowercase(self):
        """All default keys are stored in lowercase."""
        for key in DataSanitizer.DEFAULT_SENSITIVE_KEYS:
            assert key == key.lower(), f"Key {key!r} is not lowercase"


# ---------------------------------------------------------------------------
# Cross-method consistency
# ---------------------------------------------------------------------------


class TestSanitizeRedactConsistency:
    """Sanitize and redact should agree on which keys are sensitive."""

    @pytest.mark.parametrize(
        "key", sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS)
    )
    def test_sanitize_removes_what_redact_replaces(self, key):
        """For each default key, sanitize removes it and redact replaces it."""
        data = {key: "value", "safe": "keep"}
        sanitized = DataSanitizer.sanitize(data)
        redacted = DataSanitizer.redact(data)
        assert key not in sanitized
        assert redacted[key] == "[REDACTED]"
        assert sanitized["safe"] == "keep"
        assert redacted["safe"] == "keep"

    def test_sanitize_and_redact_same_custom_keys(self):
        """Both methods respect the same custom keys set."""
        custom = {"my_token", "my_secret"}
        data = {"my_token": "t", "my_secret": "s", "name": "n", "api_key": "k"}
        sanitized = DataSanitizer.sanitize(data, sensitive_keys=custom)
        redacted = DataSanitizer.redact(data, sensitive_keys=custom)

        assert "my_token" not in sanitized
        assert "my_secret" not in sanitized
        assert "api_key" in sanitized  # not in custom set

        assert redacted["my_token"] == "[REDACTED]"
        assert redacted["my_secret"] == "[REDACTED]"
        assert redacted["api_key"] == "k"  # not in custom set


# ---------------------------------------------------------------------------
# Sanitize/redact with dict containing duplicate-ish case keys
# ---------------------------------------------------------------------------


class TestCaseVariantKeys:
    """Dict keys that differ only in case."""

    def test_sanitize_removes_all_case_variants(self):
        """All case variants of a sensitive key are removed."""
        data = {"Api_Key": "v1", "API_KEY": "v2", "api_key": "v3", "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert result == {"name": "ok"}

    def test_redact_replaces_all_case_variants(self):
        """All case variants of a sensitive key are redacted."""
        data = {
            "Password": "p1",
            "PASSWORD": "p2",
            "password": "p3",
            "name": "ok",
        }
        result = DataSanitizer.redact(data)
        assert result["Password"] == "[REDACTED]"
        assert result["PASSWORD"] == "[REDACTED]"
        assert result["password"] == "[REDACTED]"
        assert result["name"] == "ok"