mirror of
https://github.com/prdlk/jobyssey.git
synced 2026-11-02 13:12:49 +00:00
173 lines
8.1 KiB
Python
173 lines
8.1 KiB
Python
"""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()
|