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

"""High-value tests for security/data_sanitizer.py - Data Sanitization."""

import json


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


# ---------------------------------------------------------------------------
# DataSanitizer.sanitize()
# ---------------------------------------------------------------------------


class TestSanitizeBasic:
    """Basic sanitize behavior."""

    def test_removes_api_key(self):
        result = DataSanitizer.sanitize(
            {"username": "bob", "api_key": "secret"}
        )
        assert result == {"username": "bob"}

    def test_removes_password(self):
        result = DataSanitizer.sanitize({"password": "pass123", "name": "x"})
        assert result == {"name": "x"}

    def test_removes_multiple_sensitive_keys(self):
        data = {"api_key": "k", "secret": "s", "safe": "v"}
        result = DataSanitizer.sanitize(data)
        assert result == {"safe": "v"}

    def test_case_insensitive_removal(self):
        data = {"API_KEY": "k", "Password": "p", "name": "n"}
        result = DataSanitizer.sanitize(data)
        assert result == {"name": "n"}

    def test_primitive_passthrough_string(self):
        assert DataSanitizer.sanitize("hello") == "hello"

    def test_primitive_passthrough_int(self):
        assert DataSanitizer.sanitize(42) == 42

    def test_primitive_passthrough_none(self):
        assert DataSanitizer.sanitize(None) is None

    def test_empty_dict(self):
        assert DataSanitizer.sanitize({}) == {}

    def test_empty_list(self):
        assert DataSanitizer.sanitize([]) == []


class TestSanitizeRecursive:
    """Recursive sanitization through nested structures."""

    def test_nested_dict(self):
        data = {"outer": {"api_key": "secret", "value": 1}}
        result = DataSanitizer.sanitize(data)
        assert result == {"outer": {"value": 1}}

    def test_deeply_nested(self):
        data = {"a": {"b": {"c": {"password": "p", "data": "d"}}}}
        result = DataSanitizer.sanitize(data)
        assert result == {"a": {"b": {"c": {"data": "d"}}}}

    def test_list_of_dicts(self):
        data = [{"api_key": "k1", "name": "a"}, {"api_key": "k2", "name": "b"}]
        result = DataSanitizer.sanitize(data)
        assert result == [{"name": "a"}, {"name": "b"}]

    def test_dict_with_list_of_dicts(self):
        data = {"items": [{"secret": "s", "id": 1}]}
        result = DataSanitizer.sanitize(data)
        assert result == {"items": [{"id": 1}]}

    def test_list_of_primitives_unchanged(self):
        assert DataSanitizer.sanitize([1, 2, 3]) == [1, 2, 3]


class TestSanitizeCustomKeys:
    """Custom sensitive_keys parameter."""

    def test_custom_keys_only(self):
        data = {"api_key": "keep", "custom_secret": "remove", "name": "n"}
        result = DataSanitizer.sanitize(data, sensitive_keys={"custom_secret"})
        assert result == {"api_key": "keep", "name": "n"}

    def test_custom_keys_case_insensitive(self):
        data = {"MY_TOKEN": "t", "name": "n"}
        result = DataSanitizer.sanitize(data, sensitive_keys={"my_token"})
        assert result == {"name": "n"}


# ---------------------------------------------------------------------------
# DataSanitizer.redact()
# ---------------------------------------------------------------------------


class TestRedactBasic:
    """Basic redact behavior."""

    def test_replaces_with_redacted(self):
        data = {"api_key": "secret", "name": "bob"}
        result = DataSanitizer.redact(data)
        assert result == {"api_key": "[REDACTED]", "name": "bob"}

    def test_case_insensitive(self):
        result = DataSanitizer.redact({"PASSWORD": "p", "x": 1})
        assert result == {"PASSWORD": "[REDACTED]", "x": 1}

    def test_custom_redaction_text(self):
        result = DataSanitizer.redact({"api_key": "k"}, redaction_text="***")
        assert result == {"api_key": "***"}

    def test_primitive_passthrough(self):
        assert DataSanitizer.redact("hello") == "hello"
        assert DataSanitizer.redact(42) == 42
        assert DataSanitizer.redact(None) is None


class TestRedactRecursive:
    """Recursive redaction."""

    def test_nested_dict_redaction(self):
        data = {"outer": {"password": "secret", "value": 1}}
        result = DataSanitizer.redact(data)
        assert result == {"outer": {"password": "[REDACTED]", "value": 1}}

    def test_list_of_dicts_redaction(self):
        data = [{"secret": "s1", "id": 1}, {"secret": "s2", "id": 2}]
        result = DataSanitizer.redact(data)
        assert result == [
            {"secret": "[REDACTED]", "id": 1},
            {"secret": "[REDACTED]", "id": 2},
        ]

    def test_list_passthrough(self):
        assert DataSanitizer.redact([1, "two", 3]) == [1, "two", 3]


class TestRedactNonPrimitiveValue:
    """Redact replaces the value wholesale, even if it's a nested structure."""

    def test_nested_dict_value_replaced_not_recursed(self):
        data = {"api_key": {"nested": "data"}}
        result = DataSanitizer.redact(data)
        assert result == {"api_key": "[REDACTED]"}


class TestRedactCustomKeys:
    """Custom sensitive_keys for redact."""

    def test_custom_keys(self):
        data = {"api_key": "keep", "my_field": "redact"}
        result = DataSanitizer.redact(data, sensitive_keys={"my_field"})
        assert result == {"api_key": "keep", "my_field": "[REDACTED]"}


# ---------------------------------------------------------------------------
# Convenience functions
# ---------------------------------------------------------------------------


class TestConvenienceFunctions:
    def test_sanitize_data_delegates(self):
        data = {"api_key": "k", "name": "n"}
        assert sanitize_data(data) == {"name": "n"}

    def test_redact_data_delegates(self):
        data = {"password": "p", "name": "n"}
        result = redact_data(data)
        assert result == {"password": "[REDACTED]", "name": "n"}

    def test_redact_data_custom_text(self):
        result = redact_data({"api_key": "k"}, redaction_text="HIDDEN")
        assert result == {"api_key": "HIDDEN"}


# ---------------------------------------------------------------------------
# filter_research_metadata()
# ---------------------------------------------------------------------------


class TestFilterResearchMetadata:
    """filter_research_metadata extracts only safe fields."""

    def test_dict_with_is_news_search_true(self):
        result = filter_research_metadata(
            {"is_news_search": True, "settings_snapshot": {"api_key": "x"}}
        )
        assert result == {"is_news_search": True}

    def test_dict_with_is_news_search_false(self):
        result = filter_research_metadata({"is_news_search": False})
        assert result == {"is_news_search": False}

    def test_dict_missing_is_news_search(self):
        result = filter_research_metadata({"other": "stuff"})
        assert result == {"is_news_search": False}

    def test_json_string_input(self):
        result = filter_research_metadata('{"is_news_search": true}')
        assert result == {"is_news_search": True}

    def test_none_input(self):
        result = filter_research_metadata(None)
        assert result == {"is_news_search": False}

    def test_non_dict_input(self):
        result = filter_research_metadata([1, 2, 3])
        assert result == {"is_news_search": False}

    def test_invalid_json_string(self):
        result = filter_research_metadata("{bad json")
        assert result == {"is_news_search": False}

    def test_empty_dict(self):
        result = filter_research_metadata({})
        assert result == {"is_news_search": False}

    def test_int_truthy_coerced_to_bool(self):
        result = filter_research_metadata({"is_news_search": 1})
        assert result == {"is_news_search": True}
        assert type(result["is_news_search"]) is bool

    def test_string_truthy_coerced_to_bool(self):
        result = filter_research_metadata({"is_news_search": "yes"})
        assert result == {"is_news_search": True}

    def test_json_string_non_dict_returns_default(self):
        """JSON string that parses to a list exercises the not-isinstance(meta, dict) branch."""
        result = filter_research_metadata("[1, 2]")
        assert result == {"is_news_search": False}


# ---------------------------------------------------------------------------
# strip_settings_snapshot()
# ---------------------------------------------------------------------------


class TestStripSettingsSnapshot:
    """strip_settings_snapshot removes settings_snapshot key."""

    def test_removes_settings_snapshot(self):
        data = {
            "phase": "done",
            "settings_snapshot": {"api_key": "x"},
            "duration": 5,
        }
        result = strip_settings_snapshot(data)
        assert result == {"phase": "done", "duration": 5}

    def test_preserves_other_keys(self):
        data = {"phase": "done", "error_type": None, "mode": "quick"}
        result = strip_settings_snapshot(data)
        assert result == data

    def test_json_string_input(self):
        data = json.dumps({"settings_snapshot": {}, "phase": "done"})
        result = strip_settings_snapshot(data)
        assert result == {"phase": "done"}

    def test_none_input(self):
        assert strip_settings_snapshot(None) == {}

    def test_non_dict_input(self):
        assert strip_settings_snapshot(42) == {}

    def test_invalid_json(self):
        assert strip_settings_snapshot("{invalid") == {}

    def test_nested_settings_snapshot_survives(self):
        data = {
            "results": {"settings_snapshot": {"key": "val"}},
            "phase": "done",
        }
        result = strip_settings_snapshot(data)
        assert result == {
            "results": {"settings_snapshot": {"key": "val"}},
            "phase": "done",
        }

    def test_json_string_non_dict_returns_empty(self):
        """JSON string that parses to a list exercises the not-isinstance branch."""
        assert strip_settings_snapshot("[1, 2]") == {}

    def test_empty_dict(self):
        assert strip_settings_snapshot({}) == {}