test(tests): update suite for sync‑twenty migration and drop obsolete gmail/notion sync tests

tests/test_twenty_tools.py

@@ -0,0 +1,172 @@
+"""Guards for the deterministic cores of the Twenty CRM tooling.
+
+tools/twenty_enrich.py owns the enrichment plan (salary ladder, company field
+fills, post-link name cleanup) and tools/gather_recent_context.py owns the
+email-thread -> opportunity matching. Both are pure given their inputs, so
+these tests pin the contracts the /sync-twenty and /gather-recent-context
+commands rely on:
+
+- source priority: a stated posting salary always beats the levels.fyi
+  estimate, and an already-filled amount is never touched;
+- company fills only ever ADD missing fields and reject the jobs.db
+  stringified-NaN domain artifact;
+- name cleanup fires only after the company relation is set;
+- thread matching goes through companyId (names are role-only), respects the
+  active-stage set, and never guesses across companies.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
+
+from twenty_enrich import (  # noqa: E402
+    build_company_plan,
+    build_name_cleanup_plan,
+    build_salary_plan,
+    clean_domain,
+    norm_company,
+)
+from gather_recent_context import build_plan, match_thread  # noqa: E402
+
+
+def opp(id="o1", name="Backend Engineer", stage="RANKED", company_id="c1",
+        url="https://jobs.example/1", amount=None):
+    return {"id": id, "name": name, "stage": stage, "companyId": company_id,
+            "url": {"primaryLinkUrl": url}, "amount": amount}
+
+
+class NormalizationTests(unittest.TestCase):
+    def test_suffixes_and_punctuation_collapse(self):
+        self.assertEqual(norm_company("DraftKings Inc."), norm_company("draftkings"))
+        self.assertEqual(norm_company("Chainlink Labs"), norm_company("Chainlink"))
+
+    def test_clean_domain_rejects_nan_artifact(self):
+        # jobs.db stores stringified float NaN; https://nan is a real failure
+        # we shipped once - Twenty rejects it and fails the whole batch.
+        self.assertIsNone(clean_domain("nan"))
+        self.assertIsNone(clean_domain(""))
+        self.assertIsNone(clean_domain(None))
+
+    def test_clean_domain_strips_scheme_and_www(self):
+        self.assertEqual(clean_domain("https://www.example.com/"), "example.com")
+
+
+class SalaryPlanTests(unittest.TestCase):
+    def test_stated_salary_beats_levels_estimate(self):
+        ranked = [{"url": "https://jobs.example/1", "company": "Acme",
+                   "title": "Backend Engineer", "salary_max_stated": 200000}]
+        updates, unresolved = build_salary_plan(
+            ranked, [opp()], postings_salary={}, levels={"acme": 150000})
+        self.assertEqual(len(updates), 1)
+        self.assertEqual(updates[0]["source"], "seen_jobs.salary_max_stated")
+        self.assertEqual(updates[0]["patch"]["amount"]["amountMicros"], 200000 * 1_000_000)
+        self.assertEqual(unresolved, [])
+
+    def test_levels_fallback_is_labeled_estimate(self):
+        ranked = [{"url": "https://jobs.example/1", "company": "Acme",
+                   "title": "Backend Engineer"}]
+        updates, _ = build_salary_plan(ranked, [opp()], postings_salary={},
+                                       levels={norm_company("Acme"): 150000})
+        self.assertIn("estimate", updates[0]["source"])
+
+    def test_filled_amount_is_never_touched(self):
+        ranked = [{"url": "https://jobs.example/1", "company": "Acme",
+                   "title": "Backend Engineer", "salary_max_stated": 999999}]
+        o = opp(amount={"amountMicros": 1, "currencyCode": "USD"})
+        updates, unresolved = build_salary_plan(ranked, [o], {}, {})
+        self.assertEqual(updates, [])
+        self.assertEqual(unresolved, [])
+
+    def test_no_source_lands_in_unresolved_not_in_updates(self):
+        ranked = [{"url": "https://jobs.example/1", "company": "Acme",
+                   "title": "Backend Engineer"}]
+        updates, unresolved = build_salary_plan(ranked, [opp()], {}, {})
+        self.assertEqual(updates, [])
+        self.assertEqual([u["company"] for u in unresolved], ["Acme"])
+
+
+class CompanyPlanTests(unittest.TestCase):
+    DB = {norm_company("Acme"): {
+        "company_id": "db1", "linkedin_url": "http://www.linkedin.com/company/acme",
+        "country": "United States", "state": "NY", "apollo_city": "New York",
+    }}
+
+    def test_fills_only_missing_fields(self):
+        company = {"id": "c1", "name": "Acme",
+                   "domainName": {"primaryLinkUrl": "https://acme.com"},  # present
+                   "linkedinLink": {}, "address": {}}
+        updates, unresolved = build_company_plan([company], self.DB, {"db1": "acme.com"})
+        patch = updates[0]["patch"]
+        self.assertNotIn("domainName", patch)  # never overwrite
+        self.assertEqual(patch["linkedinLink"]["primaryLinkUrl"],
+                         "https://www.linkedin.com/company/acme")
+        self.assertEqual(patch["address"]["addressCity"], "New York")
+        self.assertEqual(unresolved, [])
+
+    def test_nan_domain_never_reaches_the_patch(self):
+        company = {"id": "c1", "name": "Acme", "domainName": {},
+                   "linkedinLink": {"primaryLinkUrl": "https://x"}, "address": {"addressCity": "NY"}}
+        updates, unresolved = build_company_plan([company], self.DB, {"db1": "nan"})
+        self.assertEqual(updates, [])  # nothing groundable -> no patch at all
+        self.assertEqual(unresolved[0]["missing"], ["domain"])
+
+    def test_unknown_company_is_fully_unresolved(self):
+        company = {"id": "c9", "name": "Ghost Startup",
+                   "domainName": {}, "linkedinLink": {}, "address": {}}
+        updates, unresolved = build_company_plan([company], self.DB, {})
+        self.assertEqual(updates, [])
+        self.assertEqual(unresolved[0]["missing"], ["domain", "linkedin", "address"])
+
+
+class NameCleanupTests(unittest.TestCase):
+    def test_strips_suffix_only_when_linked(self):
+        linked = opp(name="Backend Engineer — Acme")
+        unlinked = opp(id="o2", name="ML Engineer — Ghost", company_id=None)
+        clean = opp(id="o3", name="Platform Engineer")
+        updates = build_name_cleanup_plan([linked, unlinked, clean])
+        self.assertEqual(updates, [{"id": "o1", "patch": {"name": "Backend Engineer"}}])
+
+
+class ThreadMatchingTests(unittest.TestCase):
+    COMPANIES = {"c1": "Acme", "c2": "Globex"}
+
+    def test_matches_through_company_relation_not_name(self):
+        # Opportunity names are role-only after cleanup; a name-based matcher
+        # would find nothing here.
+        o = opp()
+        t = {"thread_id": "t1", "company": "Acme Inc.", "role": None}
+        self.assertIs(match_thread(t, [o], self.COMPANIES), o)
+
+    def test_final_stage_loops_never_match(self):
+        o = opp(stage="REJECTED")
+        t = {"thread_id": "t1", "company": "Acme", "role": None}
+        self.assertIsNone(match_thread(t, [o], self.COMPANIES))
+
+    def test_role_tokens_break_ties_within_a_company(self):
+        backend = opp(id="o1", name="Backend Engineer")
+        frontend = opp(id="o2", name="Frontend Engineer")
+        t = {"thread_id": "t1", "company": "Acme", "role": "Senior Frontend Engineer"}
+        self.assertEqual(match_thread(t, [backend, frontend], self.COMPANIES)["id"], "o2")
+
+    def test_plan_dedups_processed_threads_and_reports_unmatched(self):
+        o = opp()
+        companies = [{"id": "c1", "name": "Acme"}]
+        threads = [
+            {"thread_id": "old", "company": "Acme", "role": None,
+             "date": "2026-08-01", "subject": "s", "summary": "x", "signal": "other"},
+            {"thread_id": "new", "company": "Acme", "role": None,
+             "date": "2026-08-02", "subject": "s", "summary": "y", "signal": "interview"},
+            {"thread_id": "stray", "company": "Nowhere", "role": None,
+             "date": "2026-08-03", "subject": "s", "summary": "z", "signal": "other"},
+        ]
+        plan = build_plan(threads, [o], companies,
+                          state={"processed_thread_ids": ["old"]})
+        self.assertEqual(plan["skipped_already_processed"], 1)
+        self.assertEqual([n["thread_ids"] for n in plan["notes"]], [["new"]])
+        self.assertEqual([u["company"] for u in plan["unmatched_threads"]], ["Nowhere"])
+        self.assertTrue(plan["notes"][0]["title"].startswith("Email Context — Acme"))
+
+
+if __name__ == "__main__":
+    unittest.main()