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