typosaurus-sandbox/tests/test_research_dedup.py

# retoor <retoor@molodetz.nl>

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()