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

"""Comprehensive coverage tests for data_sanitizer and log_sanitizer modules.

Module 1 -- data_sanitizer:
    DataSanitizer.sanitize, DataSanitizer.redact, sanitize_data, redact_data,
    filter_research_metadata, strip_settings_snapshot

Module 2 -- log_sanitizer:
    strip_control_chars, sanitize_for_log
"""

import json

import pytest

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


# ===================================================================
# DataSanitizer.sanitize
# ===================================================================


class TestSanitizeNestedDicts:
    """sanitize removes sensitive keys inside deeply nested dicts."""

    def test_two_levels_deep(self):
        data = {"outer": {"api_key": "abc", "safe": 1}}
        result = DataSanitizer.sanitize(data)
        assert result == {"outer": {"safe": 1}}

    def test_three_levels_deep(self):
        data = {"a": {"b": {"password": "pw", "x": 9}}}
        result = DataSanitizer.sanitize(data)
        assert result == {"a": {"b": {"x": 9}}}

    def test_multiple_sensitive_keys_at_different_depths(self):
        data = {
            "secret": "top",
            "level1": {
                "refresh_token": "tok",
                "level2": {"private_key": "pk", "ok": True},
            },
        }
        result = DataSanitizer.sanitize(data)
        assert result == {"level1": {"level2": {"ok": True}}}


class TestSanitizeLists:
    """sanitize recurses into lists and sanitizes dicts inside them."""

    def test_list_of_dicts(self):
        data = [{"api_key": "k", "id": 1}, {"password": "p", "id": 2}]
        result = DataSanitizer.sanitize(data)
        assert result == [{"id": 1}, {"id": 2}]

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

    def test_list_of_primitives_unchanged(self):
        data = [1, "hello", None, True]
        assert DataSanitizer.sanitize(data) == [1, "hello", None, True]


class TestSanitizeSensitiveKeyRemoval:
    """Verify all default sensitive keys are removed."""

    @pytest.mark.parametrize(
        "key", sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS)
    )
    def test_each_default_key_removed(self, key):
        data = {key: "value", "keep_me": 42}
        result = DataSanitizer.sanitize(data)
        assert key not in result
        assert result["keep_me"] == 42


class TestSanitizeCaseInsensitive:
    """sanitize matches keys case-insensitively."""

    def test_uppercase_key(self):
        data = {"API_KEY": "secret", "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert "API_KEY" not in result
        assert result == {"name": "ok"}

    def test_mixed_case_key(self):
        data = {"PaSsWoRd": "secret", "name": "ok"}
        result = DataSanitizer.sanitize(data)
        assert "PaSsWoRd" not in result

    def test_title_case_key(self):
        data = {"Access_Token": "tok", "status": "active"}
        result = DataSanitizer.sanitize(data)
        assert "Access_Token" not in result
        assert result == {"status": "active"}


class TestSanitizeJsonStringInput:
    """sanitize treats a JSON string as a primitive -- it is NOT parsed."""

    def test_json_string_returned_as_is(self):
        json_str = json.dumps({"api_key": "secret"})
        result = DataSanitizer.sanitize(json_str)
        assert result == json_str

    def test_json_array_string_returned_as_is(self):
        json_str = json.dumps([{"password": "pw"}])
        assert DataSanitizer.sanitize(json_str) == json_str


class TestSanitizeCustomKeys:
    """sanitize accepts a custom set of sensitive keys."""

    def test_custom_keys_only(self):
        data = {"my_token": "t", "api_key": "k", "name": "n"}
        result = DataSanitizer.sanitize(data, sensitive_keys={"my_token"})
        assert "my_token" not in result
        assert "api_key" in result  # not in custom set

    def test_empty_custom_keys_removes_nothing(self):
        data = {"api_key": "k", "password": "p"}
        result = DataSanitizer.sanitize(data, sensitive_keys=set())
        assert result == data


# ===================================================================
# DataSanitizer.redact
# ===================================================================


class TestRedactBasic:
    """redact replaces sensitive values with placeholder text."""

    def test_default_redaction_text(self):
        data = {"api_key": "secret123", "name": "ok"}
        result = DataSanitizer.redact(data)
        assert result["api_key"] == "[REDACTED]"
        assert result["name"] == "ok"

    def test_key_preserved_value_replaced(self):
        data = {"password": "mypass"}
        result = DataSanitizer.redact(data)
        assert "password" in result
        assert result["password"] == "[REDACTED]"


class TestRedactCustomText:
    """redact accepts custom redaction placeholder text."""

    def test_custom_redaction_text(self):
        data = {"api_key": "secret", "name": "ok"}
        result = DataSanitizer.redact(data, redaction_text="***")
        assert result["api_key"] == "***"
        assert result["name"] == "ok"

    def test_empty_string_redaction_text(self):
        data = {"secret": "val"}
        result = DataSanitizer.redact(data, redaction_text="")
        assert result["secret"] == ""

    def test_long_redaction_text(self):
        text = "SENSITIVE DATA REMOVED BY SECURITY FILTER"
        data = {"password": "pw"}
        result = DataSanitizer.redact(data, redaction_text=text)
        assert result["password"] == text


class TestRedactNestedStructures:
    """redact recurses into nested dicts and lists."""

    def test_nested_dict(self):
        data = {"config": {"auth_token": "tok", "host": "localhost"}}
        result = DataSanitizer.redact(data)
        assert result["config"]["auth_token"] == "[REDACTED]"
        assert result["config"]["host"] == "localhost"

    def test_list_of_dicts(self):
        data = [{"session_token": "s1"}, {"csrf_token": "c1"}]
        result = DataSanitizer.redact(data)
        assert result[0]["session_token"] == "[REDACTED]"
        assert result[1]["csrf_token"] == "[REDACTED]"

    def test_dict_in_list_in_dict(self):
        data = {"items": [{"private_key": "pk", "id": 1}]}
        result = DataSanitizer.redact(data)
        assert result["items"][0]["private_key"] == "[REDACTED]"
        assert result["items"][0]["id"] == 1

    def test_original_data_not_mutated(self):
        data = {"api_key": "original_value", "name": "ok"}
        DataSanitizer.redact(data)
        assert data["api_key"] == "original_value"


class TestRedactCaseInsensitive:
    """redact matches keys case-insensitively."""

    def test_uppercase_key_redacted(self):
        data = {"SECRET": "val"}
        result = DataSanitizer.redact(data)
        assert result["SECRET"] == "[REDACTED]"

    def test_mixed_case_key_redacted(self):
        data = {"Apikey": "val", "name": "ok"}
        result = DataSanitizer.redact(data)
        assert result["Apikey"] == "[REDACTED]"
        assert result["name"] == "ok"


# ===================================================================
# sanitize_data wrapper
# ===================================================================


class TestSanitizeDataWrapper:
    """sanitize_data convenience function delegates to DataSanitizer.sanitize."""

    def test_removes_default_sensitive_keys(self):
        data = {"api_key": "k", "name": "n"}
        result = sanitize_data(data)
        assert result == {"name": "n"}

    def test_accepts_custom_sensitive_keys(self):
        data = {"custom_field": "v", "api_key": "k"}
        result = sanitize_data(data, sensitive_keys={"custom_field"})
        assert "custom_field" not in result
        assert "api_key" in result

    def test_primitive_passthrough(self):
        assert sanitize_data(99) == 99
        assert sanitize_data(None) is None


# ===================================================================
# redact_data wrapper
# ===================================================================


class TestRedactDataWrapper:
    """redact_data convenience function delegates to DataSanitizer.redact."""

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

    def test_accepts_custom_keys_and_text(self):
        data = {"my_key": "v", "password": "pw"}
        result = redact_data(
            data, sensitive_keys={"my_key"}, redaction_text="XXX"
        )
        assert result["my_key"] == "XXX"
        assert result["password"] == "pw"

    def test_primitive_passthrough(self):
        assert redact_data("hello") == "hello"
        assert redact_data(None) is None


# ===================================================================
# filter_research_metadata
# ===================================================================


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

    def test_extracts_is_news_search_true(self):
        result = filter_research_metadata(
            {"is_news_search": True, "extra": "x"}
        )
        assert result == {"is_news_search": True}
        assert "extra" not in result

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

    def test_missing_is_news_search_defaults_false(self):
        result = filter_research_metadata({"some_other_key": 123})
        assert result == {"is_news_search": False}

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

    def test_json_string_input_dict(self):
        meta_json = json.dumps(
            {"is_news_search": True, "settings_snapshot": {}}
        )
        result = filter_research_metadata(meta_json)
        assert result == {"is_news_search": True}
        assert "settings_snapshot" not in result

    def test_json_string_input_false(self):
        meta_json = json.dumps({"is_news_search": False})
        result = filter_research_metadata(meta_json)
        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_integer_input_returns_default(self):
        result = filter_research_metadata(42)
        assert result == {"is_news_search": False}

    def test_is_news_search_truthy_int_coerced(self):
        """Non-zero int is truthy, coerced to True via bool()."""
        result = filter_research_metadata({"is_news_search": 1})
        assert result["is_news_search"] is True
        assert type(result["is_news_search"]) is bool

    def test_settings_snapshot_never_leaks(self):
        meta = {
            "is_news_search": False,
            "settings_snapshot": {"api_key": "SUPER_SECRET"},
        }
        result = filter_research_metadata(meta)
        assert "settings_snapshot" not in result


# ===================================================================
# strip_settings_snapshot
# ===================================================================


class TestStripSettingsSnapshot:
    """strip_settings_snapshot removes settings_snapshot, preserves the rest."""

    def test_removes_settings_snapshot(self):
        meta = {"phase": "done", "settings_snapshot": {"api_key": "k"}}
        result = strip_settings_snapshot(meta)
        assert "settings_snapshot" not in result
        assert result["phase"] == "done"

    def test_preserves_all_other_keys(self):
        meta = {
            "mode": "quick",
            "duration": 5.2,
            "is_news_search": True,
            "settings_snapshot": {},
        }
        result = strip_settings_snapshot(meta)
        assert result == {
            "mode": "quick",
            "duration": 5.2,
            "is_news_search": True,
        }

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

    def test_string_metadata_json(self):
        meta_json = json.dumps(
            {"phase": "complete", "settings_snapshot": {"k": "v"}}
        )
        result = strip_settings_snapshot(meta_json)
        assert result == {"phase": "complete"}
        assert "settings_snapshot" not in result

    def test_no_settings_snapshot_key_noop(self):
        meta = {"phase": "done", "duration": 3.0}
        result = strip_settings_snapshot(meta)
        assert result == meta

    def test_invalid_json_string_returns_empty(self):
        result = strip_settings_snapshot("not valid json {{{")
        assert result == {}

    def test_non_dict_type_returns_empty(self):
        assert strip_settings_snapshot(123) == {}
        assert strip_settings_snapshot([1, 2]) == {}


# ===================================================================
# strip_control_chars (log_sanitizer)
# ===================================================================


class TestStripControlCharsComprehensive:
    """strip_control_chars removes C0/C1 control, zero-width, and format chars."""

    def test_c0_null_and_bell(self):
        assert strip_control_chars("a\x00b\x07c") == "abc"

    def test_c0_tab_newline_carriage_return(self):
        assert strip_control_chars("a\tb\nc\rd") == "abcd"

    def test_c1_range(self):
        """C1 control characters (0x80-0x9F) are stripped."""
        assert strip_control_chars("a\x80b\x8fc\x9fd") == "abcd"

    def test_del_character(self):
        """DEL (0x7F) is stripped."""
        assert strip_control_chars("hello\x7fworld") == "helloworld"

    def test_zero_width_space(self):
        assert strip_control_chars("hello\u200bworld") == "helloworld"

    def test_zero_width_non_joiner(self):
        assert strip_control_chars("a\u200cb") == "ab"

    def test_zero_width_joiner(self):
        assert strip_control_chars("a\u200db") == "ab"

    def test_ltr_rtl_marks(self):
        assert strip_control_chars("a\u200eb\u200fc") == "abc"

    def test_unicode_format_chars_rlo(self):
        """Right-to-left override (RLO) is stripped."""
        assert strip_control_chars("hello\u202eworld") == "helloworld"

    def test_unicode_embedding_chars(self):
        """LRE, RLE, PDF, LRO are stripped."""
        assert strip_control_chars("\u202a\u202b\u202c\u202dtext") == "text"

    def test_bom_stripped(self):
        assert strip_control_chars("\ufeffhello") == "hello"

    def test_word_joiner_stripped(self):
        assert strip_control_chars("a\u2060b") == "ab"

    def test_arabic_letter_mark_stripped(self):
        assert strip_control_chars("a\u061cb") == "ab"

    def test_isolate_chars_stripped(self):
        """LRI, RLI, FSI, PDI are stripped."""
        assert strip_control_chars("\u2066\u2067\u2068\u2069text") == "text"

    def test_preserves_valid_unicode(self):
        assert (
            strip_control_chars("cafe\u0301") == "cafe\u0301"
        )  # e-acute combining

    def test_preserves_emoji(self):
        result = strip_control_chars("hello world")
        assert result == "hello world"

    def test_preserves_cjk(self):
        assert strip_control_chars("test") == "test"

    def test_empty_string(self):
        assert strip_control_chars("") == ""

    def test_all_control_chars_yields_empty(self):
        assert strip_control_chars("\x00\x01\x02\x7f\x80") == ""


# ===================================================================
# sanitize_for_log (log_sanitizer)
# ===================================================================


class TestSanitizeForLogTruncation:
    """sanitize_for_log truncates with '...' suffix when exceeding max_length."""

    def test_long_string_truncated_with_ellipsis(self):
        result = sanitize_for_log("a" * 100, max_length=50)
        assert result == "a" * 47 + "..."
        assert len(result) == 50

    def test_default_max_length_50(self):
        result = sanitize_for_log("b" * 200)
        assert len(result) == 50
        assert result.endswith("...")

    def test_exact_max_length_no_truncation(self):
        result = sanitize_for_log("c" * 50, max_length=50)
        assert result == "c" * 50
        assert "..." not in result

    def test_shorter_than_max_no_truncation(self):
        result = sanitize_for_log("short", max_length=50)
        assert result == "short"


class TestSanitizeForLogMaxLengthEdgeCases:
    """Edge-case max_length values: 0, 1, 3, 50."""

    def test_max_length_zero(self):
        """max_length=0 returns empty string (no room for anything)."""
        result = sanitize_for_log("hello", max_length=0)
        assert result == ""

    def test_max_length_one(self):
        """max_length=1 -- too short for '...' so just truncates."""
        result = sanitize_for_log("hello", max_length=1)
        assert len(result) <= 1
        assert result == "h"

    def test_max_length_three(self):
        """max_length=3 -- exactly the length of '...', uses raw truncation."""
        result = sanitize_for_log("hello", max_length=3)
        assert len(result) <= 3
        assert result == "hel"

    def test_max_length_four_uses_ellipsis(self):
        """max_length=4 -- enough room for 1 char + '...'."""
        result = sanitize_for_log("hello world", max_length=4)
        assert result == "h..."
        assert len(result) == 4

    def test_max_length_50_default(self):
        result = sanitize_for_log("x" * 60, max_length=50)
        assert len(result) == 50
        assert result == "x" * 47 + "..."


class TestSanitizeForLogEmptyString:
    """sanitize_for_log on empty input."""

    def test_empty_string_returns_empty(self):
        assert sanitize_for_log("") == ""

    def test_empty_string_with_explicit_max_length(self):
        assert sanitize_for_log("", max_length=10) == ""


class TestSanitizeForLogControlCharsAndTruncation:
    """Control chars are stripped BEFORE truncation is applied."""

    def test_control_chars_removed_then_truncated(self):
        # 10 visible chars + control chars = still only 10 visible after strip
        raw = "abcde\x00\x01\x02fghij"
        result = sanitize_for_log(raw, max_length=50)
        assert result == "abcdefghij"

    def test_control_chars_reduce_length_below_max(self):
        """After stripping, string may become shorter than max_length."""
        raw = "\x00\x01\x02abc"
        result = sanitize_for_log(raw, max_length=50)
        assert result == "abc"

    def test_control_chars_stripped_still_needs_truncation(self):
        """Even after stripping, result may still exceed max_length."""
        raw = "a" * 40 + "\x00" * 5 + "b" * 40
        result = sanitize_for_log(raw, max_length=50)
        # After strip: 80 visible chars, needs truncation to 50
        assert len(result) == 50
        assert result.endswith("...")

    def test_all_control_chars_yields_empty(self):
        result = sanitize_for_log("\x00\x01\x7f", max_length=50)
        assert result == ""

    def test_unicode_preserved_controls_stripped(self):
        raw = "cafe\u0301\x00\x07 world"
        result = sanitize_for_log(raw, max_length=50)
        assert result == "cafe\u0301 world"