#!/usr/bin/env python3 """Fixture-based tests for the report.md builders in generate.py. Runs without a database or network: it stubs `pymysql`, feeds synthetic cohorts that mix populated and not-evaluated (no-data / unsupported) mapped fields, and asserts the derived counts and the rendered markdown. Run directly: python3 scripts/sync_recon/test_report.py """ import datetime as dt import importlib.util import json import os import re import sys import types import tempfile sys.modules.setdefault("pymysql", types.ModuleType("pymysql")) _spec = importlib.util.spec_from_file_location( "generate", os.path.join(os.path.dirname(__file__), "generate.py")) g = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(g) def _cell(r, d, ms, p, se): return {"r": r, "m": p - d - ms, "d": d, "ms": ms, "p": p, "se": se, "be": 0, "vx": 0} def _win(): return { "auth_bj": dt.date(2026, 7, 22), "today_bj": dt.date(2026, 7, 23), "partial_note": "as of 22:47", "generated": "2026-07-23T22:47+08:00", "window_label": "2026-07-16 → 2026-07-23 22:47 (Beijing time)", } def _plans(): # 4 mapped fields: 2 populated, 1 no-data, 1 unsupported. mk = lambda tgt, src, unsup=False: { "tgt": tgt, "src": src, "ftype": "string", "has_transform": False, "has_vmap": False, "unsupported": unsup} return [mk("Email", "email"), mk("Industry", "salesforce_industry"), mk("AnnualRevenue", "annual_revenue"), mk("Disqualified_Date__c", "disq", unsup=True)] def _fd(): plans = _plans() y_cells = {"Email": _cell(100.0, 0, 0, 300, 0), "Industry": _cell(15.7, 0, 250, 300, 0)} t_cells = {"Email": _cell(100.0, 0, 0, 200, 0)} # Industry absent today return g.assemble_field(_win(), plans, y_cells, t_cells, y_pairs_n=300, t_pairs_n=200, y_leads_n=300, t_leads_n=200) def _dd(): def pair(nx, hs, coupling, rows, roll, ctx30=None): p = {"nx": nx, "hs": hs, "title": f"{nx} {hs}", "rows": rows, "coupling": coupling, "roll": roll} if ctx30: p["ctx30"] = ctx30 return p lrow = lambda day, nx, hs, sfdc=0, partial=False: { "day": day, "nx": nx, "hs": hs, "gap": nx - hs, "gappct": round(abs(nx - hs) / max(nx, hs, 1) * 100, 1), "uuid": nx - sfdc, "sfdc": sfdc, "sfdc_share": round(sfdc / nx * 100) if nx else 0, "partial": partial, "status": "green"} arow = lambda day, n, l, partial=False: { "day": day, "nx": n, "hs": l, "gap": l - n, "gappct": round(l / n * 100, 1) if n else None, "uuid": l, "sfdc": 0, "partial": partial, "status": "green"} return {"pairs": [ pair("Lead", "contacts", "coupled", [lrow("2026-07-22", 394, 392, sfdc=14), lrow("2026-07-23", 236, 246, partial=True)], {"nx": 394, "hs": 392, "net": 2, "avg_gap": 2, "avg_gap_pct": 0.5, "max_gap_pct": 0.5, "green": 1, "days": 1}), pair("Account", "companies", "cohort", [arow("2026-07-22", 5, 5), arow("2026-07-23", 8, 6, True)], {"nx": 5, "hs": 5, "net": 0, "avg_gap": 0, "avg_gap_pct": 100.0, "max_gap_pct": 0, "green": 1, "days": 1}, ctx30={"n": 71, "linked": 69, "rate": 97.2}), ]} def test_mapped_total_counts_not_evaluated_fields(): fd = _fd() # regression guard: the "N mapped fields" label must NOT equal rows length, # which omits the no-data and unsupported mappings. assert fd["mapped_total"] == 4, fd["mapped_total"] assert len(fd["rows"]) == 4, len(fd["rows"]) assert fd["mapped_total"] == len(fd["rows"]) assert fd["unsupported"] == 1 assert fd["nodata"] == 1 and fd["nodata_fields"] == ["AnnualRevenue"] def test_report_md_covers_both_cohorts_and_cohort_headers(): fd, dd = _fd(), _dd() md = g.build_report_md(_win(), dd, fd, {"rate": 97.2, "linked": 69, "n": 71}) assert not re.search(r"[一-鿿]", md), "no CJK in report.md" # Both cohorts have their status plus 5 metrics. hdr = next(l for l in md.splitlines() if l.startswith("| Field")) assert hdr.count("|") - 1 == 13, hdr # Account cohort uses its own headers, not the coupled ones, and drops # Status (5 columns); title has no "(targeted cohort · outbound)" suffix assert "### Accounts → Companies" in md assert "### Accounts → Companies (targeted cohort · outbound)" not in md assert "| Date | New (sync-on) | Company built | Diff (H−N) | Coverage% |" in md assert "| Date | New (sync-on) | Company built | Diff (H−N) | Coverage% | Status |" not in md ai = next(i for i, l in enumerate(md.splitlines()) if l.startswith("### Accounts")) lines2 = md.splitlines() note_i = next(i for i, l in enumerate(lines2) if i > ai and "Success target" in l) table_i = next(i for i, l in enumerate(lines2) if i > ai and l.startswith("| Date | New")) assert ai < note_i < table_i, (ai, note_i, table_i) assert "Diff = 0" in lines2[note_i] # today cohort with an absent field renders "—", a real 0 stays 0 ind = next(l for l in md.splitlines() if l.startswith("| `Industry`")) assert ind.rstrip().endswith("| — | — | — | — | — |"), ind assert "Not evaluated" in md and "AnnualRevenue" in md def test_report_md_sfdc_column_and_target(): fd, dd = _fd(), _dd() win = _win() md = g.build_report_md(win, dd, fd, {"rate": 97.2, "linked": 69, "n": 71}) lines = md.splitlines() # Lead ledger gains an SFDC% column but drops Status (8 columns): the two # pass/fail conditions are stated in the note, not collapsed into a chip. lead_hdr = next(l for l in lines if l.startswith("| Date | NexusCRM")) assert "SFDC%" in lead_hdr and "Status" not in lead_hdr assert lead_hdr.count("|") - 1 == 8, lead_hdr # summary headline bullet also carries the SFDC-channel share indicator head_bullet = next(l for l in lines if l.startswith("- Leads↔Contacts")) assert "SFDC-channel share" in head_bullet and "target 0" in head_bullet # success-target line sits under the Leads heading, before the table, and # states both conditions plus which row is authoritative hi = next(i for i, l in enumerate(lines) if l.startswith("### Leads")) ti = next(i for i, l in enumerate(lines) if l.startswith("| Date | NexusCRM")) note_i = next(i for i, l in enumerate(lines) if "Success target" in l) assert hi < note_i < ti, (hi, note_i, ti) assert "Gap = 0 and SFDC = 0" in lines[note_i] assert "yesterday" in lines[note_i].lower() # per-day + total rows carry the SFDC% cell but no Status cell (8 columns) assert next(l for l in lines if l.startswith("| 2026-07-22 |")).count("|") - 1 == 8 assert next(l for l in lines if "-day total" in l).count("|") - 1 == 8 def test_report_md_opportunity_title_and_columns(): # Opportunity: no "(low-volume · non-1:1)" suffix on the title, its own # "Gap = 0" success-target note, and no UUID/SFDC/SFDC%/Status columns — # those don't apply (Opportunities are pushed NX->HS; the write-path # split is a Lead-inbound concept). fd = _fd() orow = lambda day, nx, hs, partial=False: { "day": day, "nx": nx, "hs": hs, "gap": nx - hs, "gappct": round(abs(nx - hs) / max(nx, hs, 1) * 100, 1), "uuid": nx, "sfdc": 0, "sfdc_share": 0, "partial": partial, "status": "green"} lrow = {"day": "2026-07-22", "nx": 10, "hs": 10, "gap": 0, "gappct": 0.0, "uuid": 10, "sfdc": 0, "sfdc_share": 0, "partial": False, "status": "green"} dd = {"pairs": [ {"nx": "Lead", "hs": "contacts", "title": "L", "coupling": "coupled", "rows": [lrow], "roll": {"nx": 10, "hs": 10, "net": 0, "avg_gap": 0, "avg_gap_pct": 0.0, "max_gap_pct": 0, "green": 1, "days": 1}}, {"nx": "Opportunity", "hs": "deals", "title": "O", "coupling": "lowvol", "rows": [orow("2026-07-22", 5, 5)], "roll": {"nx": 5, "hs": 5, "net": 0, "avg_gap": 0, "avg_gap_pct": 0.0, "max_gap_pct": 0, "green": 1, "days": 1}}, ]} md = g.build_report_md(_win(), dd, fd, {"rate": 100, "linked": 1, "n": 1}) lines = md.splitlines() assert "### Opportunities ↔ Deals" in lines assert "### Opportunities ↔ Deals (low-volume · non-1:1)" not in lines # two "| Date | NexusCRM ... |" headers exist (Lead, then Opportunity); # Opportunity's has only 5 columns (no UUID/SFDC/SFDC%/Status) hdrs = [l for l in lines if l.startswith("| Date | NexusCRM")] assert len(hdrs) == 2, hdrs assert hdrs[0].count("|") - 1 == 8, hdrs[0] # Lead: unchanged assert hdrs[1].count("|") - 1 == 5, hdrs[1] for col in ("UUID", "SFDC", "Status"): assert col not in hdrs[1], hdrs[1] data_rows = [l for l in lines if l.startswith("| 2026-07-22 |")] assert data_rows[1].rstrip().endswith("| +0 | 0.0 |") # Opportunity gets its own Gap-only success-target note, under its # heading and before its table oi = next(i for i, l in enumerate(lines) if l.startswith("### Opportunities")) note_i = next(i for i, l in enumerate(lines) if i > oi and "Success target" in l) table_i = next(i for i, l in enumerate(lines) if i > oi and l.startswith("| Date | NexusCRM")) assert oi < note_i < table_i, (oi, note_i, table_i) assert "Gap = 0" in lines[note_i] and "SFDC" not in lines[note_i] def test_lead_sfdc_share_cell_uses_exact_ratio_rounding(): # 51/1000 = 5.1%: the SFDC% cell is the rounded share (an integer per # sfdc_share's own rounding), shown as-is — there's no separate Status # cell to misclassify, since the Lead table has none (see modification 2). row = {"day": "2026-07-01", "nx": 1000, "hs": 1000, "gap": 0, "gappct": 0.0, "uuid": 949, "sfdc": 51, "sfdc_share": round(51 / 1000 * 100), "partial": False, "status": "green"} assert row["sfdc_share"] == 5 dd = {"pairs": [{"nx": "Lead", "hs": "contacts", "title": "L", "coupling": "coupled", "rows": [row], "roll": {"nx": 1000, "hs": 1000, "net": 0, "avg_gap": 0, "avg_gap_pct": 0.0, "max_gap_pct": 0, "green": 1, "days": 1}}]} md = g.build_report_md(_win(), dd, _fd(), {"rate": 100, "linked": 1, "n": 1}) line = next(l for l in md.splitlines() if l.startswith("| 2026-07-01 |")) # last cell is the SFDC% value (5) assert line.rstrip().endswith("| 5 |"), line def test_field_section_heading_note_and_order(): plans = [ {"tgt": "Zebra_Field__c", "src": "zebra", "ftype": "string", "has_transform": False, "has_vmap": False, "unsupported": False}, {"tgt": "LastName", "src": "lastname", "ftype": "string", "has_transform": False, "has_vmap": False, "unsupported": False}, {"tgt": "Apple_Field__c", "src": "apple", "ftype": "string", "has_transform": False, "has_vmap": False, "unsupported": False}, {"tgt": "FirstName", "src": "firstname", "ftype": "string", "has_transform": False, "has_vmap": False, "unsupported": False}, {"tgt": "Email", "src": "email", "ftype": "string", "has_transform": False, "has_vmap": False, "unsupported": False}, ] # give Zebra_Field__c (non-priority) the worst NX-missing to prove the # display order is NOT nx_missing_desc — assemble_field sorts fd["rows"] # by marketing priority (then alpha), unconditionally, since report.md is # the only remaining consumer of the row order. y_cells = {"Zebra_Field__c": _cell(10.0, 0, 90, 100, 0), "LastName": _cell(100.0, 0, 0, 100, 0), "Apple_Field__c": _cell(100.0, 0, 0, 100, 0), "FirstName": _cell(100.0, 0, 0, 100, 0), "Email": _cell(100.0, 0, 0, 100, 0)} fd = g.assemble_field(_win(), plans, y_cells, {}, y_pairs_n=100, t_pairs_n=0, y_leads_n=100, t_leads_n=0) # fd["rows"] itself is in priority order: FirstName (rank 0) is rows[0], # not Zebra_Field__c despite its worse NX-missing. assert fd["rows"][0]["t"] == "FirstName" assert [r["t"] for r in fd["rows"]] == \ ["FirstName", "LastName", "Email", "Apple_Field__c", "Zebra_Field__c"] dd = _dd() md = g.build_report_md(_win(), dd, fd, {"rate": 100, "linked": 1, "n": 1}) lines = md.splitlines() assert "## Field-level reconciliation (HubSpot Contact ↔ NexusCRM Lead)" in md assert "HubSpot Contact → NexusCRM Lead, inbound" not in md assert "Sorted by Y·missing desc" not in md note_i = next(i for i, l in enumerate(lines) if "Comparison target" in l) hdr_i = next(i for i, l in enumerate(lines) if l.startswith("## Field-level")) table_i = next(i for i, l in enumerate(lines) if l.startswith("| Field (NX")) assert hdr_i < note_i < table_i assert "rounded 100% does not establish complete alignment" in lines[note_i] # report.md table order matches fd["rows"]: priority fields in the given # order (FirstName, LastName, Email), then remaining fields alphabetically # (Apple_Field__c before Zebra_Field__c) field_rows = [l for l in lines[table_i + 2:] if l.startswith("| `")] order = [l.split("`")[1] for l in field_rows] assert order == ["FirstName", "LastName", "Email", "Apple_Field__c", "Zebra_Field__c"], order def _company_plan(): expr = 'LEFT(BLANKVALUE($Record.company, "HubSpot Lead"), 255)' return {"tgt": "Company", "src": "company", "ftype": "string", "node": g.parse_expr(expr), "vmap": None, "on_unmapped": "skip", "has_transform": True, "has_vmap": False, "unsupported": None} def test_company_etl_placeholder_equivalence(): # A parallel SFDC ETL full-record-replace writes the literal # "[not provided]" into Lead.Company for many blank-HubSpot-company # records, instead of the pack's declared BLANKVALUE default # ("HubSpot Lead") — verified against live prod data 2026-07-30 (204/213 # of that day's Company "differ" rows were exactly this pair). These # should compare as consistent, NOT differ. plans = [_company_plan()] blank_co_leads = [ ({"Id": f"blank{i}", "Company": "[not provided]"}, {"company": None}) for i in range(3) ] cells = g.field_recon(blank_co_leads, plans, len(blank_co_leads)) c = cells["Company"] assert c["m"] == 3 and c["d"] == 0 and c["r"] == 100.0, c # Regression guard: a REAL HubSpot company value must still be compared # for real against NexusCRM's value — the equivalence map must not # swallow genuine mismatches (including the case where a real HubSpot # company was itself overwritten to "[not provided]" by the same ETL, # which is a real defect, not a placeholder pair, and must stay Differ). real_mismatch_leads = [ ({"Id": "real1", "Company": "[not provided]"}, {"company": "UC Davis"}), ({"Id": "real2", "Company": "Compass Polling"}, {"company": "Compass"}), ] cells2 = g.field_recon(real_mismatch_leads, plans, len(real_mismatch_leads)) c2 = cells2["Company"] assert c2["m"] == 0 and c2["d"] == 2, c2 # A genuine match (both sides hold the same real company) is unaffected. real_match_leads = [({"Id": "real3", "Company": "Acme Inc"}, {"company": "Acme Inc"})] cells3 = g.field_recon(real_match_leads, plans, len(real_match_leads)) assert cells3["Company"]["m"] == 1 and cells3["Company"]["d"] == 0 _INDUSTRY_CASE_EXPR = ( 'CASE($Record.salesforce_industry, "FSI-FinTech", "FSI-FinTech", ' '"DNB-Software", "DNB-Software", "Others")' ) def _industry_plan(): return {"tgt": "Industry", "src": "salesforce_industry", "ftype": "string", "node": g.parse_expr(_INDUSTRY_CASE_EXPR), "vmap": None, "on_unmapped": "skip", "has_transform": True, "has_vmap": False, "unsupported": None} def test_industry_blank_source_counts_as_src_empty(): # Industry's CASE(...,"Others") else-default doesn't reflect runtime # behavior when salesforce_industry is blank — the real connector leaves # NexusCRM Industry genuinely blank too, it does not write "Others". # Verified against live prod data 2026-07-30: 236/252 of that day's # Industry "nx_missing" rows were exactly blank-source, NX-blank pairs. plans = [_industry_plan()] # field_recon()'s cells output omits a field entirely when populated==0 # (nothing to score), so each all-blank-source batch below is padded with # one real, matched record purely to keep "Industry" present in `cells`; # the assertions account for that extra match. padder = ({"Id": "pad", "Industry": "DNB-Software"}, {"salesforce_industry": "DNB-Software"}) blank_src_leads = [({"Id": f"blank{i}", "Industry": ""}, {"salesforce_industry": None}) for i in range(3)] + [padder] cells = g.field_recon(blank_src_leads, plans, len(blank_src_leads)) c = cells["Industry"] assert c["p"] == 1 and c["se"] == 3, c # 3 blanks excluded, not counted as nx_missing # Regression guard #1: this must key off the SOURCE being blank, not # whether NX happens to literally hold "Others" — a lead where the # connector coincidentally wrote "Others" for a blank source is STILL # excluded as src_empty (matching how every other field without a # declared default already treats a blank source: unconditionally, not # only for the cases that happen to disagree). blank_src_others_leads = [({"Id": "b1", "Industry": "Others"}, {"salesforce_industry": None}), padder] cells2 = g.field_recon(blank_src_others_leads, plans, len(blank_src_others_leads)) assert cells2["Industry"]["p"] == 1 and cells2["Industry"]["se"] == 1 # Regression guard #2: a REAL, non-blank HubSpot value that legitimately # falls to the CASE else-branch (e.g. an industry not in the pack's list) # must still be compared for real — this is a genuine sync gap (16 such # records on 2026-07-30) that this fix must NOT hide. real_unmatched_leads = [({"Id": "r1", "Industry": ""}, {"salesforce_industry": "Some Other Industry"})] cells3 = g.field_recon(real_unmatched_leads, plans, len(real_unmatched_leads)) c3 = cells3["Industry"] assert c3["se"] == 0 and c3["ms"] == 1, c3 # still nx_missing, not swallowed # Regression guard #3: a real, matched CASE value still compares normally. real_match_leads = [({"Id": "r2", "Industry": "DNB-Software"}, {"salesforce_industry": "DNB-Software"})] cells4 = g.field_recon(real_match_leads, plans, len(real_match_leads)) assert cells4["Industry"]["m"] == 1 and cells4["Industry"]["se"] == 0 # Company is untouched by this Industry-only override. co_plans = [_company_plan()] co_leads = [({"Id": "c1", "Company": "[not provided]"}, {"company": None})] assert g.field_recon(co_leads, co_plans, 1)["Company"]["m"] == 1 def _vertical_plan(): expr = '$Record.salesforce_industry & "|" & $Record.salesforce_industry__vertical_' vmap = {"FSI-FinTech|Payment": "Payment", "DNB-Software|Mobile App Development": "Mobile App Development"} return {"tgt": "Vertical__c", "src": "salesforce_industry__vertical_", "ftype": "string", "node": g.parse_expr(expr), "vmap": vmap, "on_unmapped": "skip", "has_transform": True, "has_vmap": True, "unsupported": None} def test_vertical_blank_concat_counts_as_src_empty(): # Vertical__c concatenates two source properties with "|". When BOTH are # blank, the raw (pre-valueMap) result is the literal "|" — never a # valueMap key (every key has real text on both sides) — so today it # falls to vmap_miss, indistinguishable from a genuine unmapped value. # Verified against live prod data 2026-07-30: all 265 of that day's # Vertical__c vmap_miss rows were exactly "|". These should count as # src_empty, not vmap_miss. plans = [_vertical_plan()] both_blank_leads = [({"Id": f"blank{i}", "Vertical__c": ""}, {"salesforce_industry": None, "salesforce_industry__vertical_": None}) for i in range(3)] # field_recon()'s cells output omits a field entirely when populated==0, # so pad with one real matched record to keep "Vertical__c" in `cells`. padder = ({"Id": "pad", "Vertical__c": "Payment"}, {"salesforce_industry": "FSI-FinTech", "salesforce_industry__vertical_": "Payment"}) cells = g.field_recon(both_blank_leads + [padder], plans, 4) c = cells["Vertical__c"] assert c["p"] == 1 and c["se"] == 3 and c["vx"] == 0, c # Regression guard: a REAL concatenated value that's simply not in the # (necessarily incomplete, in this fixture) valueMap must still be # vmap_miss, not swallowed as src_empty — only the exact "both blank" # literal gets reclassified. real_unmapped_leads = [({"Id": "r1", "Vertical__c": ""}, {"salesforce_industry": "Some Industry", "salesforce_industry__vertical_": "Some Vertical"}), padder] cells2 = g.field_recon(real_unmapped_leads, plans, 2) assert cells2["Vertical__c"]["vx"] == 1 and cells2["Vertical__c"]["se"] == 0 # A real, matched value still compares normally. real_match_leads = [({"Id": "r2", "Vertical__c": "Payment"}, {"salesforce_industry": "FSI-FinTech", "salesforce_industry__vertical_": "Payment"})] cells3 = g.field_recon(real_match_leads, plans, 1) assert cells3["Vertical__c"]["m"] == 1 and cells3["Vertical__c"]["se"] == 0 def test_hs_batch_read_matches_via_secondary_email(): # HubSpot's batch/read-by-idProperty=email can resolve a contact via a # SECONDARY email (hs_additional_emails), returning that contact's # PRIMARY email in `properties.email` — different from the email that was # actually looked up. Keying the result only by the primary email then # makes `.get()` silently miss a real match. Verified # against a real live case 2026-07-31: a HubSpot contact created # 2026-07-08 whose primary email is "...@webmarketing.fr" resolved when # queried by "saoussane.tlaghi@gmail.com" (one of its hs_additional_emails). calls = [] def fake_hs_req(token, path, method, body): calls.append(body) assert "hs_additional_emails" in body["properties"] # fetched even if caller didn't ask return { "results": [{ "id": "233985700451", "properties": { "email": "saoussane.tlaghi@webmarketing.fr", "hs_additional_emails": "saoussane.tlaghi@smartolyp.ch;saoussane.tlaghi@sater.online;saoussane.tlaghi@gmail.com", }, }], "errors": [{"category": "OBJECT_NOT_FOUND", "context": {"ids": ["kenichi.yamane@g.softbank.co.jp"]}}], "numErrors": 1, } orig = g.hs_req g.hs_req = fake_hs_req try: by_email = g.hs_batch_read("tok", "contacts", ["kenichi.yamane@g.softbank.co.jp", "saoussane.tlaghi@gmail.com"], ["email"], id_property="email") finally: g.hs_req = orig # the secondary email that was actually looked up now resolves assert "saoussane.tlaghi@gmail.com" in by_email assert by_email["saoussane.tlaghi@gmail.com"]["email"] == "saoussane.tlaghi@webmarketing.fr" # the primary email and every other additional email also resolve to the # same contact (any of a contact's known emails should find it) for k in ("saoussane.tlaghi@webmarketing.fr", "saoussane.tlaghi@smartolyp.ch", "saoussane.tlaghi@sater.online"): assert by_email[k]["email"] == "saoussane.tlaghi@webmarketing.fr" # regression guard: a genuinely-absent email is NOT in the result assert "kenichi.yamane@g.softbank.co.jp" not in by_email def test_complete_inventory_today_only_missing_column_and_unsupported(): mappings = [ {"sourceField": "today", "targetField": "Today"}, {"sourceField": "absent", "targetField": "Absent"}, {"sourceField": "date", "targetField": "Date", "transformExpression": 'BLANKVALUE($Record.date, IF(ISBLANK($Record.date), "", "later"))'}, {"sourceField": "projection", "projectionOnly": True}, ] with tempfile.NamedTemporaryFile(mode="w", suffix=".json") as f: json.dump([{"mappings": mappings}], f) f.flush() plans = g.load_mapping(f.name, {"Today", "Date"}, {}) assert len(plans) == 3 assert g.src_props_cols(plans) == ["Date", "Today"] assert plans[2]["unsupported"] == "transform:IF" # hidden branch is still checked y = g.field_recon([], plans, 0) t = g.field_recon([({"Today": "value"}, {"today": "value"})], plans, 1) fd = g.assemble_field(_win(), plans, y, t, 0, 1, 0, 1) assert fd["mapped_total"] == len(fd["rows"]) == 3 assert fd["missing_columns"] == fd["unsupported"] == 1 assert fd["nodata_fields"] == [] assert y["Today"]["status"] == "no_pairs" and t["Today"]["m"] == 1 assert y["Absent"]["status"] == "missing_column" assert y["Date"]["status"] == "unsupported" md = g.build_report_md(_win(), _dd(), fd, {"rate": 0, "linked": 0, "n": 0}) assert "target column absent from Lead schema" in md and "transform:IF" in md assert "| `Today` ← `today` | no matched records in cohort | —" in md assert "no data on either side" not in md and "None%" not in md def test_runtime_errors_and_exclusions_remain_visible_and_accounted(): p = {"tgt": "Slice", "src": "text", "ftype": "string", "node": g.parse_expr("LEFT($Record.text, $Record.length)"), "vmap": None, "on_unmapped": "skip", "has_transform": True, "has_vmap": False, "unsupported": None} pairs = [({"Slice": "ok"}, {"text": "okay", "length": "2"}), ({"Slice": "private-value"}, {"text": "private-value", "length": "not-an-int"})] y = g.field_recon(pairs, [p], 2) c = y["Slice"] assert c["m"] == 1 and c["errors"] == 1 and c["status"] == "partial" assert sum(c[k] for k in ("p", "se", "vx", "be", "errors")) == 2 t = g.field_recon(pairs[1:], [p], 1) assert t["Slice"]["status"] == "partial" and t["Slice"]["r"] is None fd = g.assemble_field(_win(), [p], y, t, 2, 1, 2, 1) assert fd["perfect"] == 0 and fd["nodata_fields"] == [] md = g.build_report_md(_win(), _dd(), fd, {"rate": 0, "linked": 0, "n": 0}) assert "1 evaluation errors" in md and "private-value" not in md assert "private-value" not in json.dumps(fd) vertical = g.field_recon([({"Vertical__c": "kept"}, { "salesforce_industry": "unknown", "salesforce_industry__vertical_": "value"})], [_vertical_plan()], 1)["Vertical__c"] assert vertical["status"] == "partial" and vertical["vx"] == 1 and vertical["p"] == 0 blank = g.field_recon([({"Company": "kept"}, {"company": ""})], [{ **_company_plan(), "node": None}], 1)["Company"] assert blank["status"] == "no_comparable_values" and blank["se"] == 1 def test_current_mapping_inventory_includes_all_conditional_date_fields(): path = os.path.join(os.path.dirname(__file__), "../../packs/PingCAP-HubSpot-Companion/field_mappings/contact_lead_inbound.json") with open(path) as f: mappings = json.load(f)[0]["mappings"] targets = {m["targetField"] for m in mappings if m.get("targetField")} plans = g.load_mapping(path, targets, {}) assert len(plans) == len(targets) cells = g.field_recon([], plans, 0) fd = g.assemble_field(_win(), plans, cells, cells, 0, 0, 0, 0) assert {r["t"] for r in fd["rows"]} == targets unsupported = {p["tgt"] for p in plans if p["unsupported"]} assert {"agreement_time__c", "created_at__c", "Tied_Card_Time__c", "Lead_Stage_DateV2__c", "Disqualified_Date__c"} <= unsupported assert fd["perfect"] == 0 and fd["y_overall"] is None if __name__ == "__main__": fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")] for fn in fns: fn() print(f"ok {fn.__name__}") print(f"\n{len(fns)} passed")