#!/usr/bin/env python3 """Tests for recipe R1 (lead capture). Offline: no network, no n8n, no Google. Run from the recipes folder: python3 -m pytest r1/test_lead_capture.py -v What is being proved here, in plain words: a lead goes in, one row and one draft come out; the same lead a second time changes nothing; bad input is named instead of guessed; and nothing in the recipe - the Python, the n8n workflow, or the Apps Script - has a way to write a message out. Read-date: 2026-11-10. """ from __future__ import annotations import ast import json import re import sys from pathlib import Path import pytest HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import lead_capture_to_sheets as lc # noqa: E402 SAMPLES = HERE / "samples" / "incoming" WORKFLOW = HERE / "lead_capture.n8n.json" CODE_FILE = HERE / "n8n_normalize_code.js" APPS_SCRIPT = HERE / "lead_capture.apps-script.gs" BUILDER = HERE / "build_workflow_json.py" #: Names that would mean this recipe could put a message on a wire. The point #: of this list is that it comes back empty for every file in the recipe. FORBIDDEN_CAPABILITY = re.compile( r"send|smtp|urlopen|urllib|requests|httpx|smtplib|socket|postmark|mailgun|" r"sendgrid|twilio|urlfetch", re.IGNORECASE, ) #: An n8n parameter value that would turn a draft into a message going out. FORBIDDEN_N8N_VALUES = { "send", "sendemail", "send and wait", "createandsend", "create and send", "reply", "replyall", "reply all", "forward", "batchsend", "batch send", } def run_once(tmp_path: Path, incoming: Path, extra: dict | None = None) -> lc.Report: """One run against throwaway sheet and outbox paths.""" options = { "incoming": incoming, "sheet": tmp_path / "data" / "leads_sheet.csv", "outbox": tmp_path / "outbox" / "leads_drafts", "now_iso": "2026-11-10T12:00:00Z", } options.update(extra or {}) return lc.run(**options) def sheet_lines(sheet: Path) -> list[list[str]]: import csv with sheet.open(encoding="utf-8", newline="") as handle: return [row for row in csv.reader(handle) if any(cell.strip() for cell in row)] # ---------------------------------------------------------------- behaviour def test_shipped_samples_produce_rows_and_drafts(tmp_path): """The sample folder in the pack is enough to see the recipe work.""" report = run_once(tmp_path, SAMPLES) assert report.new_count == 3 assert sorted(report.new_emails) == [ "amara@example.com", "devin.park@example.net", "ines@example.org", ] assert len(report.draft_paths) == 3 assert report.sheet_rows_after == 3 assert report.leads_seen == 7 assert len(report.skipped) == 3 assert len(report.duplicates) == 1 def test_re_running_the_same_folder_writes_nothing_new(tmp_path): """The guarantee the buyer checks: run it twice, the second run is a no-op.""" first = run_once(tmp_path, SAMPLES) sheet = tmp_path / "data" / "leads_sheet.csv" lines_after_first = sheet_lines(sheet) drafts_after_first = sorted(p.name for p in (tmp_path / "outbox" / "leads_drafts").iterdir()) second = run_once(tmp_path, SAMPLES) lines_after_second = sheet_lines(sheet) drafts_after_second = sorted(p.name for p in (tmp_path / "outbox" / "leads_drafts").iterdir()) assert first.new_count == 3 assert second.new_count == 0 assert second.draft_paths == [] assert lines_after_second == lines_after_first assert drafts_after_second == drafts_after_first assert second.sheet_rows_after == 3 # 4, not 3: amara's address appears in both sample files, so on the second # run both of her rows are held out by the sheet. assert len(second.duplicates) == 4 assert all(where in ("on the sheet", "in this run") for _, where, _ in second.duplicates) def test_email_is_matched_case_insensitively(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "a.json").write_text( json.dumps([{"name": "A", "email": "Person@Example.COM"}]), encoding="utf-8" ) (incoming / "b.json").write_text( json.dumps([{"name": "A again", "email": "person@example.com"}]), encoding="utf-8" ) report = run_once(tmp_path, incoming) assert report.new_count == 1 assert report.duplicates == [("person@example.com", "in this run", "b.json record 1")] # And on the next run the same address is held out by the sheet itself. again = run_once(tmp_path, incoming) assert again.new_count == 0 assert {where for _, where, _ in again.duplicates} == {"on the sheet"} def test_same_address_twice_inside_one_run_is_counted_once(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "a.json").write_text( json.dumps([ {"name": "One", "email": "twice@example.com"}, {"name": "Two", "email": "TWICE@example.com", "message": "second submission"}, ]), encoding="utf-8", ) report = run_once(tmp_path, incoming) assert report.new_count == 1 assert report.duplicates[0][1] == "in this run" assert report.duplicates[0][0].lower() == "twice@example.com" def test_a_bad_address_is_named_not_guessed(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "a.json").write_text( json.dumps([ {"name": "Fine", "email": "fine@example.com"}, {"name": "Broken", "email": "no-at-sign-here"}, {"name": "Empty"}, ]), encoding="utf-8", ) report = run_once(tmp_path, incoming) assert report.new_count == 1 reasons = {reason for reason, _ in report.skipped} assert any("no email value" in reason for reason in reasons) assert any("'no-at-sign-here'" in reason for reason in reasons) # Nothing was guessed: only the good address reached the sheet. body = (tmp_path / "data" / "leads_sheet.csv").read_text(encoding="utf-8") assert "no-at-sign-here" not in body assert "fine@example.com" in body def test_draft_is_named_by_number_and_marks_itself_a_draft(tmp_path): report = run_once(tmp_path, SAMPLES) names = sorted(p.name for p in report.draft_paths) assert all(re.match(r"^\d{4}-", name) for name in names), names text = report.draft_paths[0].read_text(encoding="utf-8") assert text.startswith(lc.DRAFT_BANNER) assert lc.READ_DATE in text assert "Subject: Re: your enquiry" in text assert "To: amara@example.com" in text assert "08:12" not in text or "Monday sales report" in text def test_draft_sequence_continues_on_the_next_run(tmp_path): """A new lead later does not overwrite the earlier drafts.""" run_once(tmp_path, SAMPLES) incoming = tmp_path / "incoming_extra" incoming.mkdir() (incoming / "later.json").write_text( json.dumps([{"name": "Late", "email": "late@example.com"}]), encoding="utf-8" ) second = run_once(tmp_path, incoming) assert [p.name for p in second.draft_paths] == ["0004-late-example.com.txt"] def test_signature_and_subject_are_the_buyers_words(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "a.json").write_text( json.dumps([{"name": "Priya Raman", "email": "priya@example.com", "source": "newsletter", "message": "Can you take my Monday report?"}]), encoding="utf-8", ) report = run_once( tmp_path, incoming, extra={"subject": "Your automation request", "signature": "Leah - Back Office"}, ) text = report.draft_paths[0].read_text(encoding="utf-8") assert "Subject: Your automation request" in text assert text.rstrip().endswith("Leah - Back Office") assert "Hi Priya," in text assert "about newsletter" in text def test_the_run_claims_no_response_time_it_cannot_keep(tmp_path): """A reply written in the buyer's name must not promise a deadline for them.""" report = run_once(tmp_path, SAMPLES) for path in report.draft_paths: text = path.read_text(encoding="utf-8").lower() for phrase in ("within 24 hours", "within one business day", "asap", "right away"): assert phrase not in text, (path.name, phrase) def test_a_file_that_cannot_be_read_writes_nothing_at_all(tmp_path): """Half a bad folder must not leave half a sheet.""" incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "good.json").write_text( json.dumps([{"name": "Good", "email": "good@example.com"}]), encoding="utf-8" ) (incoming / "broken.json").write_text("{not json at all", encoding="utf-8") with pytest.raises(lc.InputFileError) as caught: run_once(tmp_path, incoming) message = str(caught.value) assert "broken.json" in message assert not (tmp_path / "data" / "leads_sheet.csv").exists() assert not (tmp_path / "outbox" / "leads_drafts").exists() def test_a_csv_with_no_email_column_names_the_file_and_its_columns(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "odd.csv").write_text( "contact,telephone,memo\nSam,555,Fell through\n", encoding="utf-8" ) with pytest.raises(lc.InputFileError) as caught: run_once(tmp_path, incoming) message = str(caught.value) assert "odd.csv" in message assert "no email column" in message assert "contact" in message and "memo" in message def test_a_headerless_csv_is_named(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "blank.csv").write_text("", encoding="utf-8") with pytest.raises(lc.InputFileError) as caught: run_once(tmp_path, incoming) assert "blank.csv" in str(caught.value) assert "no header row" in str(caught.value) def test_an_unknown_suffix_is_reported_and_the_run_continues(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "notes.xlsx").write_text("a spreadsheet this recipe does not read", encoding="utf-8") (incoming / "fine.csv").write_text( "email,name\nok@example.com,OK\n", encoding="utf-8" ) report = run_once(tmp_path, incoming) assert report.new_count == 1 assert any("notes.xlsx" in origin for _, origin in report.skipped) def test_a_foreign_sheet_is_refused_not_appended_to(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() (incoming / "a.csv").write_text("email\nx@example.com\n", encoding="utf-8") sheet = tmp_path / "data" / "leads_sheet.csv" sheet.parent.mkdir(parents=True) sheet.write_text("something,else\n1,2\n", encoding="utf-8") with pytest.raises(lc.SheetFileError) as caught: run_once(tmp_path, incoming, extra={"sheet": sheet}) assert "its header is" in str(caught.value) def test_missing_incoming_folder_is_named(tmp_path): with pytest.raises(lc.InputFileError) as caught: run_once(tmp_path, tmp_path / "nowhere") assert "no such folder" in str(caught.value) def test_command_line_exit_codes_and_stdout(tmp_path, capsys): ok = lc.main([ "--incoming", str(SAMPLES), "--sheet", str(tmp_path / "a.csv"), "--outbox", str(tmp_path / "drafts"), "--now", "2026-11-10T12:00:00Z", ]) printed = capsys.readouterr().out assert ok == 0 assert "appended : 3" in printed assert "nothing was sent" in printed bad = lc.main([ "--incoming", str(tmp_path / "absent"), "--sheet", str(tmp_path / "b.csv"), "--outbox", str(tmp_path / "drafts2"), ]) assert bad == 2 assert "no such folder" in capsys.readouterr().err def test_quiet_and_archive_and_bad_template(tmp_path, capsys): source = tmp_path / "incoming" source.mkdir() (source / "a.csv").write_text("email\nq@example.com\n", encoding="utf-8") assert lc.main([ "--incoming", str(source), "--sheet", str(tmp_path / "s.csv"), "--outbox", str(tmp_path / "d"), "--archive", str(tmp_path / "used"), "--quiet", ]) == 0 assert capsys.readouterr().out == "" assert (tmp_path / "used" / "a.csv").exists() assert not (source / "a.csv").exists() assert lc.main([ "--incoming", str(source), "--sheet", str(tmp_path / "s2.csv"), "--outbox", str(tmp_path / "d2"), "--body-template", str(tmp_path / "nope.txt"), ]) == 2 def test_a_tsv_file_and_a_wrapped_json_payload_both_work(tmp_path): incoming = tmp_path / "incoming" incoming.mkdir() rows = [["email", "Full Name", "Message"], ["", "", ""], ["one@tsv.example", "One Tsv", "Hello there"]] (incoming / "leads.tsv").write_text( "\n".join("\t".join(row) for row in rows) + "\n", encoding="utf-8" ) (incoming / "wrapped.json").write_text( json.dumps({"records": [{"Email Address": "two@json.example", "name": "Two"}]}), encoding="utf-8", ) report = run_once(tmp_path, incoming) assert sorted(report.new_emails) == ["one@tsv.example", "two@json.example"] assert any("empty row" in reason for reason, _ in report.skipped) def test_summary_text_is_readable_and_counts_every_row(tmp_path): report = run_once(tmp_path, SAMPLES) text = report.summary() assert f"leads seen : {report.leads_seen}" in text assert "appended : 3" in text assert "duplicates : 1 (0 already on the sheet, 1 repeated inside this run)" in text assert "read-date 2026-11-10" in text for _, origin in report.skipped: assert origin in text # ------------------------------------------------- nothing in it can send def test_python_module_exposes_no_writing_out_capability(): offenders = [ name for name, value in vars(lc).items() if not name.startswith("__") and callable(value) and FORBIDDEN_CAPABILITY.search(name) ] assert offenders == [] def test_python_module_imports_only_the_standard_library(): tree = ast.parse((HERE / "lead_capture_to_sheets.py").read_text(encoding="utf-8")) roots = set() for node in ast.walk(tree): if isinstance(node, ast.Import): roots.update(alias.name.split(".")[0] for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: roots.add(node.module.split(".")[0]) assert roots <= {"__future__", "argparse", "csv", "json", "re", "sys", "dataclasses", "datetime", "pathlib", "datetime"}, roots def test_no_file_in_the_recipe_hands_the_buyer_a_way_to_write_out(): """The claim is about capability, so it is checked as code, not as prose. The word "send" appears in this recipe's comments and docstrings, where it explains that nothing sends. So the check reads what each file actually imports and calls, and for the two scripts it reads the code with the comments taken away. """ for path in sorted(HERE.glob("*.py")): offenders = sorted( name for name in imported_and_called(path) if FORBIDDEN_CAPABILITY.search(name) ) assert offenders == [], (path.name, offenders) for path in (CODE_FILE, APPS_SCRIPT): code = re.sub(r"(?s)/\*.*?\*/|//[^\n]*", "", path.read_text(encoding="utf-8")) hits = sorted(set(FORBIDDEN_CAPABILITY.findall(code))) assert hits == [], (path.name, hits) def imported_and_called(path: Path) -> set[str]: """Every module a Python file imports, and every function it calls by name.""" tree = ast.parse(path.read_text(encoding="utf-8")) names: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.Import): names.update(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: names.add(node.module) elif isinstance(node, ast.Call): if isinstance(node.func, ast.Name): names.add(node.func.id) elif isinstance(node.func, ast.Attribute): names.add(node.func.attr) return names # ------------------------------------------------- the two other artifacts def test_n8n_workflow_loads_and_chains_in_one_direction(): workflow = json.loads(WORKFLOW.read_text(encoding="utf-8")) assert workflow["active"] is False, "a shipped workflow must not be live" names = [node["name"] for node in workflow["nodes"]] assert names == [ "New lead arrives", "Normalize and check", "Add the lead row", "Create the reply draft", ] types = [node["type"] for node in workflow["nodes"]] assert types == [ "n8n-nodes-base.webhook", "n8n-nodes-base.code", "n8n-nodes-base.googleSheets", "n8n-nodes-base.gmail", ] assert len(set(node["id"] for node in workflow["nodes"])) == 4 for source, links in workflow["connections"].items(): assert source in names for branch in links["main"]: for link in branch: assert link["node"] in names, (source, link) assert "Create the reply draft" not in workflow["connections"], "the draft ends the chain" def test_n8n_workflow_makes_a_draft_and_cannot_send(): workflow = json.loads(WORKFLOW.read_text(encoding="utf-8")) gmail = next(n for n in workflow["nodes"] if n["type"] == "n8n-nodes-base.gmail") assert gmail["parameters"]["resource"] == "draft" assert gmail["parameters"]["operation"] == "create" assert gmail["parameters"]["sendTo"] == "={{ $json.email }}" for node in workflow["nodes"]: for key, value in node["parameters"].items(): if isinstance(value, str): assert value.strip().lower() not in FORBIDDEN_N8N_VALUES, (node["name"], key, value) assert node["type"] != "n8n-nodes-base.emailSend" def test_n8n_workflow_holds_no_credentials_and_no_placeholders(): text = WORKFLOW.read_text(encoding="utf-8") workflow = json.loads(text) for node in workflow["nodes"]: for kind, held in node.get("credentials", {}).items(): assert held["id"] == "", (node["name"], kind) for marker in ("TODO", "FIXME", "XXX", "PLACEHOLDER", "CHANGEME", "your_id"): assert marker not in text, marker def test_n8n_code_node_holds_the_shipped_script_word_for_word(): """The readable .js file and the workflow cannot drift apart unnoticed.""" workflow = json.loads(WORKFLOW.read_text(encoding="utf-8")) code_node = next(n for n in workflow["nodes"] if n["type"] == "n8n-nodes-base.code") assert code_node["parameters"]["jsCode"] == CODE_FILE.read_text(encoding="utf-8") def test_n8n_script_rejects_a_bad_address_and_a_second_submission(): """The same rules the Python recipe enforces, read out of the script.""" script = CODE_FILE.read_text(encoding="utf-8") assert "var SHOW_REJECTED = false;" in script assert "stored.seenEmails.push(email)" in script assert "EMAIL.test(email)" in script assert "n8n_normalize_code.js" in script, "it says what file it came from" assert "$getWorkflowStaticData" in script def test_apps_script_creates_a_draft_and_never_lets_one_out_by_itself(): script = APPS_SCRIPT.read_text(encoding="utf-8") assert "GmailApp.createDraft(" in script for forbidden in ("GmailApp.sendEmail", "MailApp.send", "Session.getActiveUser"): assert forbidden not in script, forbidden assert "http://" not in script and "https://" not in script assert "DRAFT - NOT SENT BY THIS SCRIPT" in script def test_apps_script_dedupes_and_refuses_a_bad_address(): script = APPS_SCRIPT.read_text(encoding="utf-8") assert "function findLeadRow_" in script assert "'duplicate'" in script assert "'rejected'" in script assert "EMAIL_RE.test(" in script def test_apps_script_ships_a_self_check_that_cleans_up_after_itself(): script = APPS_SCRIPT.read_text(encoding="utf-8") assert "function verifyR1()" in script assert "@example.invalid" in script, "its test address can never receive mail" assert "sheet.deleteRow(row)" in script def test_every_shipped_file_carries_the_read_date(): for path in sorted(HERE.glob("*")): if path.suffix in {".py", ".json", ".gs", ".js"}: assert lc.READ_DATE in path.read_text(encoding="utf-8"), path.name def test_the_sample_folder_holds_only_the_two_synthetic_lead_files(): """What this check does: the incoming folder is exactly the two sample files, in formats this recipe reads. (It does not check the rest of the folder - `data/` and `outbox/` are the shipped output of one run on these samples, which RECIPE.md says plainly.)""" stray = [ p.relative_to(HERE).as_posix() for p in HERE.rglob("*") if p.is_file() and p.parent.name == "incoming" and p.suffix not in {".json", ".csv", ".tsv"} ] assert stray == [] assert sorted(p.name for p in SAMPLES.iterdir()) == [ "leads_batch_1.json", "leads_batch_2.csv", ]