""" 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="" ) assert result["my_secret"] == "" 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"