#!/usr/bin/env python3 """Normalise and verify the local synthetic fixture. No network or CRM access.""" import argparse import copy import csv from datetime import datetime from decimal import Decimal from fractions import Fraction import hashlib import io import json from pathlib import Path import re import zipfile HERE = Path(__file__).resolve().parent HEADERS = ("snapshot_at", "account_key", "deal_id", "deal_title", "status", "value", "currency", "pipeline_id", "pipeline_name", "stage_id", "stage_name", "owner_id", "deal_probability_pct", "stage_probability_pct", "effective_probability_pct", "probability_source", "expected_close_date", "won_time", "lost_time", "update_time") MEMBERS = ("pipedrive-source.json", "deals-snapshot.csv", "normalise-pipedrive.py", "measures.dax", "expected-results.csv", "README.txt") def require(condition, detail): if not condition: raise ValueError(detail) def number(value, label, maximum=None): require(isinstance(value, (int, Decimal)) and not isinstance(value, bool), f"{label}: expected number") require(value >= 0 and (maximum is None or value <= maximum), f"{label}: out of fixture range") return value def utc(value): require(isinstance(value, str) and value.endswith("Z"), "Fixture timestamps must explicitly use UTC") return datetime.fromisoformat(value.replace("Z", "+00:00")) def page_rows(pages): require(isinstance(pages, list) and len(pages) > 0, "Expected at least one page, including an empty terminal page if needed") expected_cursor = None rows = [] used = set() for index, page in enumerate(pages): require(page["request_cursor"] == expected_cursor, "Missing, repeated or out-of-order page cursor") response = page["response"] require(response["success"] is True and isinstance(response["data"], list), "Unsuccessful or malformed source page") rows.extend(response["data"]) expected_cursor = response["additional_data"]["next_cursor"] if expected_cursor is None: require(index == len(pages) - 1, "Rows supplied after terminal page") else: require(expected_cursor not in used, "Repeated next cursor") used.add(expected_cursor) require(expected_cursor is None, "Incomplete pagination: final next_cursor must be null") return rows def lookup(response, label): require(response["success"] is True and response["additional_data"]["next_cursor"] is None, f"{label}: lookup must be complete") rows = response["data"] result = {row["id"]: row for row in rows} require(len(result) == len(rows), f"{label}: duplicate IDs") return result def normalise(source): metadata = source["metadata"] require(metadata["account_key"] == "SYNTHETIC-ACCOUNT" and "synthetic" in metadata["kind"], "Teaching provenance missing") snapshot = utc(metadata["snapshot_at"]) pipelines = lookup(source["pipelines"], "pipelines") stages = lookup(source["stages"], "stages") current = page_rows(source["deals_pages"]) archived = page_rows(source["archived_deals_pages"]) all_rows = current + archived require(len({row["id"] for row in all_rows}) == len(all_rows), "Duplicate deal ID across source scope") result = [] for deal in current: require(isinstance(deal["id"], int) and deal["title"].startswith("SYNTHETIC "), "Synthetic deal keys/titles required") require(deal["status"] in ("open", "won", "lost", "deleted") and isinstance(deal["is_deleted"], bool), "Deal status/deletion type") if deal["is_deleted"] or deal["status"] == "deleted": continue require(deal["currency"] == metadata["currency"] == "GBP", "Mixed/unsupported included currency; convert through a separately agreed policy first") pipeline = pipelines[deal["pipeline_id"]] stage = stages[deal["stage_id"]] require(stage["pipeline_id"] == pipeline["id"], "Stage belongs to a different pipeline") require(pipeline["is_deleted"] is False and stage["is_deleted"] is False, "Fixture requires current lookup rows") require(pipeline["is_deal_probability_enabled"] is True, "Fixture specifically models an enabled deal-probability pipeline") number(deal["value"], "value") deal_probability = deal["probability"] stage_probability = stage["deal_probability"] for value in (deal_probability, stage_probability): if value is not None: number(value, "probability", 100) if deal_probability is not None: probability, origin = deal_probability, "deal" elif stage_probability is not None: probability, origin = stage_probability, "stage" else: probability, origin = 100, "default" require(utc(deal["update_time"]) <= snapshot, "Deal update is later than snapshot") for field in ("won_time", "lost_time"): if deal[field] is not None: require(utc(deal[field]) <= snapshot, f"{field} later than snapshot") if deal["expected_close_date"] is not None: require(datetime.strptime(deal["expected_close_date"], "%Y-%m-%d").strftime("%Y-%m-%d") == deal["expected_close_date"], "Date-only expected close date") result.append(dict(zip(HEADERS, (metadata["snapshot_at"], metadata["account_key"], deal["id"], deal["title"], deal["status"], deal["value"], deal["currency"], pipeline["id"], pipeline["name"], stage["id"], stage["name"], deal["owner_id"], deal_probability, stage_probability, probability, origin, deal["expected_close_date"], deal["won_time"], deal["lost_time"], deal["update_time"])))) return sorted(result, key=lambda row: row["deal_id"]) def as_csv(rows): output = io.StringIO(newline="") writer = csv.DictWriter(output, HEADERS, lineterminator="\n") writer.writeheader() writer.writerows(rows) return output.getvalue().encode() def controls(source, rows): active = [row for row in rows if row["status"] == "open"] won = [row for row in rows if row["status"] == "won"] lost = [row for row in rows if row["status"] == "lost"] closed = len(won) + len(lost) f = lambda value: Fraction(str(value)) return {"source_deals": len(page_rows(source["deals_pages"])) + len(page_rows(source["archived_deals_pages"])), "excluded_archived": len(page_rows(source["archived_deals_pages"])), "excluded_deleted": sum(row["is_deleted"] or row["status"] == "deleted" for row in page_rows(source["deals_pages"])), "staged_deals": len(rows), "open_deals": len(active), "open_pipeline_value": sum((f(row["value"]) for row in active), Fraction(0)), "weighted_open_pipeline": sum((f(row["value"]) * f(row["effective_probability_pct"]) / 100 for row in active), Fraction(0)), "won_deal_value": sum((f(row["value"]) for row in won), Fraction(0)), "lost_deal_value": sum((f(row["value"]) for row in lost), Fraction(0)), "closed_deals": closed, "closed_deal_win_rate": Fraction(len(won), closed) if closed else None} def verify(source, rows): actual = controls(source, rows) expected = list(csv.DictReader(io.StringIO((HERE / "expected-results.csv").read_text()))) require(len(expected) == len(actual) and len({row["control"] for row in expected}) == len(expected), "Unique numerical controls") for row in expected: require(Fraction(row["expected_value"]) == actual[row["control"]], f"Control differs: {row['control']}") # Explicit, independent source-input controls catch the two common weighting mistakes. require(Fraction(10000) * 20 / 100 + Fraction(20000) * 50 / 100 + Fraction(5000) * 0 / 100 == actual["weighted_open_pipeline"] == 12000, "Independent source-weighting control") require(actual["open_pipeline_value"] == 35000 and actual["won_deal_value"] == 12000 and actual["lost_deal_value"] == 8000, "Independent status/value controls") require(actual["closed_deal_win_rate"] == Fraction(1, 2), "Closed-deal denominator must exclude open deals") require(rows[0]["effective_probability_pct"] == 20 and rows[0]["probability_source"] == "stage", "Null falls back to stage") require(rows[1]["effective_probability_pct"] == 50 and rows[1]["probability_source"] == "deal", "Deal probability overrides stage") require(rows[2]["effective_probability_pct"] == 0 and rows[2]["probability_source"] == "deal", "Explicit zero must stay zero") require(as_csv(normalise(source)) == as_csv(normalise(source)), "Repeated local load must be deterministic") require({row["deal_id"] for row in rows} == {9101, 9102, 9103, 9104, 9105}, "Archive/deletion scope") checks = ["source-to-CSV joins", "exact independent totals", "null fallback", "deal override", "explicit zero preserved", "archive/deletion exclusions", "repeat load replaces same snapshot"] def rejected(label, mutation): changed = copy.deepcopy(source) mutation(changed) try: normalise(changed) except (ValueError, KeyError): checks.append(label) return raise ValueError(f"Guard failed: {label}") rejected("duplicate deal rejected", lambda data: data["deals_pages"][1]["response"]["data"].append(copy.deepcopy(data["deals_pages"][0]["response"]["data"][0]))) rejected("mixed included currency rejected", lambda data: data["deals_pages"][0]["response"]["data"][0].update(currency="USD")) rejected("incomplete page chain rejected", lambda data: data["deals_pages"].pop()) rejected("failed page rejected", lambda data: data["deals_pages"][1]["response"].update(success=False)) rejected("missing stage rejected", lambda data: data["deals_pages"][0]["response"]["data"][0].update(stage_id=123)) rejected("invalid probability rejected", lambda data: data["deals_pages"][0]["response"]["data"][0].update(probability=101)) no_probability = copy.deepcopy(source) no_probability["stages"]["data"][0]["deal_probability"] = None require(normalise(no_probability)[0]["effective_probability_pct"] == 100, "Neither configured defaults to 100 percent") checks.append("neither configured defaults to 100 percent") dax = (HERE / "measures.dax").read_text() require(len(re.findall(r"^[A-Za-z][A-Za-z ]+ =\s*$", dax, re.M)) == 7, "Seven DAX measure definitions") require('PipedriveDeals[status] IN { "won", "lost" }' in dax and 'PipedriveDeals[effective_probability_pct]' in dax, "DAX status and weighting definitions") for name in MEMBERS: text = (HERE / name).read_text() require("\u2014" not in text, f"Em dash in download copy: {name}") require(not re.search(r"(?i)(?:[?&](?:access_token|api_token)=[A-Za-z0-9_-]{12,}|-----BEGIN [A-Z ]+PRIVATE KEY-----|shpat_[A-Za-z0-9]{20,}|sk_live_[A-Za-z0-9]{20,})", text), f"Credential-shaped content: {name}") return actual, checks def zip_bytes(): stream = io.BytesIO() with zipfile.ZipFile(stream, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive: for name in sorted(MEMBERS): info = zipfile.ZipInfo(name, (2026, 10, 17, 0, 0, 0)) info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = 0o100644 << 16 archive.writestr(info, (HERE / name).read_bytes(), compress_type=zipfile.ZIP_DEFLATED, compresslevel=9) return stream.getvalue() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--check", action="store_true", help="Check existing staged CSV without rewriting it") parser.add_argument("--write-zip", action="store_true", help="Write deterministic ZIP after fixture verification") parser.add_argument("--evidence", type=Path, help="Write verification evidence to this path") args = parser.parse_args() source = json.loads((HERE / "pipedrive-source.json").read_text(), parse_float=Decimal) rows = normalise(source) output = as_csv(rows) path = HERE / "deals-snapshot.csv" if args.check: require(path.read_bytes() == output, "Staged CSV differs from normalised source") else: path.write_bytes(output) actual, checks = verify(source, rows) content = zip_bytes() zip_path = HERE / "pipedrive-power-bi-example.zip" if args.write_zip: zip_path.write_bytes(content) elif zip_path.is_file(): require(zip_path.read_bytes() == content, "ZIP differs from deterministic member bytes") if zip_path.is_file(): with zipfile.ZipFile(zip_path) as archive: require(archive.testzip() is None and set(archive.namelist()) == set(MEMBERS), "ZIP CRC/member integrity") require(len(archive.namelist()) == len(MEMBERS), "Duplicate ZIP members") for name in MEMBERS: require(archive.read(name) == (HERE / name).read_bytes(), f"ZIP member differs: {name}") checks.append("ZIP deterministic bytes, members and CRC") else: checks.append("ZIP absent after extraction: local fixture verified") files = MEMBERS + ((zip_path.name,) if zip_path.is_file() else ()) evidence = {"status": "passed", "execution": "Local Python mapping, exact arithmetic and download integrity only; no live API, DAX engine, Power BI Desktop or Service refresh execution", "snapshot_at": source["metadata"]["snapshot_at"], "currency": "GBP", "controls": {key: str(value) for key, value in actual.items()}, "checks": checks, "files": {name: {"bytes": (HERE / name).stat().st_size, "sha256": hashlib.sha256((HERE / name).read_bytes()).hexdigest()} for name in files}} if args.evidence: args.evidence.parent.mkdir(parents=True, exist_ok=True) args.evidence.write_text(json.dumps(evidence, indent=2) + "\n") print(json.dumps(evidence, indent=2)) if __name__ == "__main__": main()