diff --git a/tests/test_research_dedup.py b/tests/test_research_dedup.py new file mode 100644 index 0000000..4fc3d6b --- /dev/null +++ b/tests/test_research_dedup.py @@ -0,0 +1,310 @@ +# retoor + +import unittest + +from typosaurus_sandbox.research.envelopes import SearchResult +from typosaurus_sandbox.research.frontier import ( + MAX_QUERY_LENGTH, + MIN_QUERY_LENGTH, + DedupStats, + QueryFrontier, + fingerprint_text, + normalize_url, + query_variants_from_result, +) + + +def _round_halts(frontier: QueryFrontier, before: DedupStats) -> bool: + after = frontier.snapshot() + new_urls = after.urls_seen - before.urls_seen + new_queries = after.queries_enqueued - before.queries_enqueued + return new_urls == 0 and new_queries == 0 + + +class TestNormalizeUrl(unittest.TestCase): + + def test_lowercases_scheme_and_host_and_strips_default_port(self) -> None: + self.assertEqual( + normalize_url("HTTPS://Example.COM:443/Path//To//Page/"), + "https://example.com/Path/To/Page", + ) + + def test_strips_userinfo_and_fragment_keeps_query(self) -> None: + self.assertEqual( + normalize_url("https://user:pass@example.com:8443/a?x=1#frag"), + "https://example.com:8443/a?x=1", + ) + + def test_fragment_dropped_with_default_port(self) -> None: + self.assertEqual(normalize_url("https://example.com/a?x=1#sec"), "https://example.com/a?x=1") + + def test_non_default_port_preserved(self) -> None: + self.assertEqual(normalize_url("https://example.com:80/x"), "https://example.com:80/x") + + def test_idna_encodes_non_ascii_host(self) -> None: + self.assertEqual(normalize_url("https://MÜNCHEN.example/"), "https://xn--mnchen-3ya.example/") + + def test_http_and_https_remain_distinct(self) -> None: + self.assertNotEqual(normalize_url("http://example.com/x"), normalize_url("https://example.com/x")) + + def test_non_http_scheme_returned_cleaned(self) -> None: + self.assertEqual(normalize_url("not a url"), "not a url") + + def test_blank_url_normalizes_to_empty(self) -> None: + self.assertEqual(normalize_url(" "), "") + + +class TestFingerprintText(unittest.TestCase): + + def test_whitespace_variants_produce_identical_fingerprint(self) -> None: + self.assertEqual(fingerprint_text("identical body\n\n"), fingerprint_text("identical body")) + + def test_distinct_text_produces_distinct_fingerprint(self) -> None: + self.assertNotEqual(fingerprint_text("first text"), fingerprint_text("second text")) + + def test_fingerprint_is_sha256_hex(self) -> None: + digest = fingerprint_text("sample") + self.assertEqual(len(digest), 64) + int(digest, 16) + + +class TestQueryVariantsFromResult(unittest.TestCase): + + def test_title_description_and_string_extra_become_variants(self) -> None: + result = SearchResult( + title="Deep research", + description="Survey of deep research systems", + url="https://a.example", + extra={"tag": "research methods", "rank": 3}, + ) + self.assertEqual( + query_variants_from_result(result), + [ + ("Deep research", "title"), + ("Survey of deep research systems", "description"), + ("research methods", "extra"), + ], + ) + + def test_non_string_extra_values_ignored(self) -> None: + result = SearchResult(title="t", url="https://a.example", extra={"rank": 3, "ok": True}) + self.assertEqual(query_variants_from_result(result), [("t", "title")]) + + def test_empty_fields_produce_no_variants(self) -> None: + result = SearchResult(url="https://a.example") + self.assertEqual(query_variants_from_result(result), []) + + +class TestUrlDeduplication(unittest.IsolatedAsyncioTestCase): + + async def test_first_registration_records_url_once(self) -> None: + frontier = QueryFrontier() + self.assertTrue(frontier.register_url("https://example.com/page")) + stats = frontier.snapshot() + self.assertEqual(stats.urls_seen, 1) + self.assertEqual(stats.urls_duplicates_skipped, 0) + + async def test_same_url_registered_twice_rejects_second(self) -> None: + frontier = QueryFrontier() + self.assertTrue(frontier.register_url("https://example.com/page")) + self.assertFalse(frontier.register_url("https://example.com/page")) + stats = frontier.snapshot() + self.assertEqual(stats.urls_seen, 1) + self.assertEqual(stats.urls_duplicates_skipped, 1) + + async def test_normalized_variants_of_same_url_rejected(self) -> None: + frontier = QueryFrontier() + self.assertTrue(frontier.register_url("HTTPS://Example.COM:443/a//b/")) + self.assertFalse(frontier.register_url("https://example.com/a/b")) + stats = frontier.snapshot() + self.assertEqual(stats.urls_seen, 1) + self.assertEqual(stats.urls_duplicates_skipped, 1) + + async def test_duplicate_urls_across_responses_recorded_once(self) -> None: + frontier = QueryFrontier() + first = SearchResult(url="https://example.com/page", title="first title", description="first description") + second = SearchResult(url="https://example.com/page", title="second title", description="second description") + self.assertTrue(frontier.register_result(first)) + self.assertFalse(frontier.register_result(second)) + stats = frontier.snapshot() + self.assertEqual(stats.urls_seen, 1) + self.assertEqual(stats.urls_duplicates_skipped, 1) + + async def test_empty_url_rejected(self) -> None: + frontier = QueryFrontier() + self.assertFalse(frontier.register_url("")) + self.assertEqual(frontier.snapshot().urls_seen, 0) + + async def test_whitespace_url_normalized_and_deduplicated(self) -> None: + frontier = QueryFrontier() + self.assertTrue(frontier.register_url(" ")) + self.assertFalse(frontier.register_url(" ")) + stats = frontier.snapshot() + self.assertEqual(stats.urls_seen, 1) + self.assertEqual(stats.urls_duplicates_skipped, 1) + + +class TestContentDeduplication(unittest.IsolatedAsyncioTestCase): + + async def test_identical_content_different_urls_rejects_second_occurrence(self) -> None: + frontier = QueryFrontier() + first = SearchResult(url="https://a.example/1", title="t1", description="d1", content="identical body") + second = SearchResult(url="https://b.example/2", title="t2", description="d2", content="identical body") + self.assertTrue(frontier.register_result(first)) + self.assertTrue(frontier.register_result(second)) + stats = frontier.snapshot() + self.assertEqual(stats.urls_seen, 2) + self.assertEqual(stats.content_seen, 1) + self.assertEqual(stats.content_duplicates_skipped, 1) + self.assertFalse(frontier.register_content("identical body")) + + async def test_near_identical_whitespace_content_rejected(self) -> None: + frontier = QueryFrontier() + self.assertTrue(frontier.register_content(" Deep research system \n")) + self.assertFalse(frontier.register_content("Deep research system")) + stats = frontier.snapshot() + self.assertEqual(stats.content_seen, 1) + self.assertEqual(stats.content_duplicates_skipped, 1) + + async def test_blank_content_rejected(self) -> None: + frontier = QueryFrontier() + self.assertFalse(frontier.register_content("")) + self.assertFalse(frontier.register_content(" \n ")) + self.assertEqual(frontier.snapshot().content_seen, 0) + + async def test_result_without_content_registers_url_only(self) -> None: + frontier = QueryFrontier() + result = SearchResult(url="https://a.example", title="t", description="d") + self.assertTrue(frontier.register_result(result)) + stats = frontier.snapshot() + self.assertEqual(stats.urls_seen, 1) + self.assertEqual(stats.content_seen, 0) + + +class TestQueryDeduplication(unittest.IsolatedAsyncioTestCase): + + async def test_duplicate_query_rejected(self) -> None: + frontier = QueryFrontier() + self.assertTrue(frontier.push_query("deep research", "manual")) + self.assertFalse(frontier.push_query("deep research", "manual")) + stats = frontier.snapshot() + self.assertEqual(stats.queries_enqueued, 1) + self.assertEqual(stats.queries_duplicates_skipped, 1) + + async def test_query_dedup_ignores_case_and_whitespace(self) -> None: + frontier = QueryFrontier() + self.assertTrue(frontier.push_query(" Deep RESEARCH ")) + self.assertFalse(frontier.push_query("deep research")) + self.assertEqual(frontier.snapshot().queries_enqueued, 1) + + async def test_variants_from_result_deduplicated_across_fields(self) -> None: + frontier = QueryFrontier() + result = SearchResult( + title="Python asyncio", + description="python asyncio", + url="https://a.example", + extra={"tag": " PYTHON ASYNCIO "}, + ) + self.assertEqual(frontier.push_variants_from_result(result), 1) + stats = frontier.snapshot() + self.assertEqual(stats.queries_enqueued, 1) + self.assertEqual(stats.queries_duplicates_skipped, 2) + + async def test_duplicate_query_never_issued_twice(self) -> None: + frontier = QueryFrontier("asyncio python") + self.assertEqual(frontier.pop_query(), "asyncio python") + self.assertFalse(frontier.push_query("ASYNCIO python")) + self.assertIsNone(frontier.pop_query()) + self.assertEqual(frontier.snapshot().queries_issued, 1) + + async def test_query_length_window_enforced(self) -> None: + frontier = QueryFrontier() + self.assertFalse(frontier.push_query("a" * (MIN_QUERY_LENGTH - 1))) + self.assertTrue(frontier.push_query("a" * MIN_QUERY_LENGTH)) + self.assertTrue(frontier.push_query("b" * MAX_QUERY_LENGTH)) + self.assertFalse(frontier.push_query("c" * (MAX_QUERY_LENGTH + 1))) + self.assertEqual(frontier.snapshot().queries_enqueued, 2) + + async def test_reseed_same_subject_enqueues_once(self) -> None: + frontier = QueryFrontier("subject alpha") + frontier.seed("SUBJECT ALPHA") + stats = frontier.snapshot() + self.assertEqual(stats.queries_enqueued, 1) + self.assertEqual(stats.queries_duplicates_skipped, 1) + + +class TestClosureDecision(unittest.IsolatedAsyncioTestCase): + + async def test_round_with_no_new_urls_and_no_new_queries_halts(self) -> None: + frontier = QueryFrontier("subject alpha") + frontier.pop_query() + discovered = SearchResult( + url="https://a.example/page", title="alpha discovery", description="alpha details", content="body text" + ) + self.assertTrue(frontier.register_result(discovered)) + before = frontier.snapshot() + self.assertFalse(frontier.register_result(discovered)) + self.assertFalse(frontier.push_query("SUBJECT ALPHA")) + after = frontier.snapshot() + self.assertEqual(after.urls_seen - before.urls_seen, 0) + self.assertEqual(after.queries_enqueued - before.queries_enqueued, 0) + self.assertTrue(_round_halts(frontier, before)) + + async def test_round_adding_new_url_continues(self) -> None: + frontier = QueryFrontier("subject beta") + frontier.pop_query() + before = frontier.snapshot() + self.assertTrue(frontier.register_url("https://new.example/x")) + after = frontier.snapshot() + self.assertEqual(after.urls_seen - before.urls_seen, 1) + self.assertFalse(_round_halts(frontier, before)) + + async def test_round_adding_new_query_continues(self) -> None: + frontier = QueryFrontier("subject gamma") + frontier.pop_query() + before = frontier.snapshot() + result = SearchResult(url="https://a.example", title="gamma subtopic", description="") + self.assertEqual(frontier.push_variants_from_result(result), 1) + after = frontier.snapshot() + self.assertEqual(after.queries_enqueued - before.queries_enqueued, 1) + self.assertFalse(_round_halts(frontier, before)) + + async def test_round_adding_url_and_query_continues(self) -> None: + frontier = QueryFrontier("subject delta") + frontier.pop_query() + before = frontier.snapshot() + result = SearchResult(url="https://b.example/page", title="delta subtopic", description="delta details") + self.assertTrue(frontier.register_result(result)) + self.assertEqual(frontier.push_variants_from_result(result), 2) + after = frontier.snapshot() + self.assertEqual(after.urls_seen - before.urls_seen, 1) + self.assertEqual(after.queries_enqueued - before.queries_enqueued, 2) + self.assertFalse(_round_halts(frontier, before)) + + async def test_empty_result_round_halts_and_exhausts_pending(self) -> None: + frontier = QueryFrontier("lonely topic") + before = frontier.snapshot() + self.assertEqual(frontier.pop_query(), "lonely topic") + self.assertFalse(frontier.has_pending()) + self.assertTrue(_round_halts(frontier, before)) + + async def test_snapshot_reports_round_deltas_for_closure_accounting(self) -> None: + frontier = QueryFrontier("subject epsilon") + before = frontier.snapshot() + self.assertEqual(frontier.pop_query(), "subject epsilon") + result = SearchResult( + url="https://c.example/page", title="epsilon topic", description="epsilon details", content="epsilon body" + ) + self.assertTrue(frontier.register_result(result)) + after = frontier.snapshot() + self.assertEqual(after.urls_seen - before.urls_seen, 1) + self.assertEqual(after.content_seen - before.content_seen, 1) + self.assertEqual(after.queries_issued - before.queries_issued, 1) + self.assertEqual(len(after.to_dict()), 8) + + +if __name__ == "__main__": + unittest.main() + + +