#!/usr/bin/env python3 """Tests for the mobile add-investor flow (Phase 8g). Boots the REAL server against a temp DB and exercises the create path the mobile "New investor" sheet drives: - POST /api/fundraising/log-communication with create_investor_if_missing honors an optional initial `priority` flag on the NEW row (and defaults it to False when omitted); - the brand-new row's source_row_id resolves immediately for the follow-on POST /api/fundraising/pipeline/link (the relational sync runs inside the create), so the create -> link-at-stage handshake the UI does works end to end; - a follow-on POST /api/reminders with the new row's source_row_id resolves to the synced investor (the create -> reminder handshake). Synthetic data only. Run: cd backend && python3 test_grid_add_investor.py """ import http.client import json import os import sqlite3 import sys import tempfile import threading from http.server import ThreadingHTTPServer _DATA = tempfile.mkdtemp() os.environ["CRM_DATA_DIR"] = _DATA os.environ["CRM_DB_PATH"] = os.path.join(_DATA, "crm.db") sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import server # noqa: E402 FAILS = [] def check(cond, msg): print((" PASS " if cond else " FAIL ") + msg) if not cond: FAILS.append(msg) class _Quiet(server.CRMHandler): def log_message(self, *a): pass def _req(port, method, path, token=None, body=None): conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) headers = {} if token: headers["Authorization"] = "Bearer " + token payload = None if body is not None: payload = json.dumps(body) headers["Content-Type"] = "application/json" conn.request(method, path, body=payload, headers=headers) resp = conn.getresponse() raw = resp.read().decode("utf-8", "replace") conn.close() data = None if raw: try: data = json.loads(raw) except ValueError: pass return resp.status, data def _db(): return sqlite3.connect(os.environ["CRM_DB_PATH"]) def seed(): c = _db() c.execute("INSERT INTO users (id,username,email,password_hash,full_name,role,is_active) " "VALUES ('u1','grant','grant@ten31.example','x','Grant','admin',1)") c.commit() c.close() def _create(port, token, name, contact_name, **extra): body = { "investor_name": name, "create_investor_if_missing": True, "contact": {"name": contact_name, "email": contact_name.split(" ")[0].lower() + "@firm.com"}, "type": "note", "body": extra.pop("note", ""), "append_note": bool(extra.pop("note_append", False)), } body.update(extra) return _req(port, "POST", "/api/fundraising/log-communication", token, body) def _grid_rows(port, token): st, d = _req(port, "GET", "/api/fundraising/state", token) return {r["id"]: r for r in (d or {}).get("data", {}).get("grid", {}).get("rows", [])} def main(): server.init_db() seed() token = server.create_token("u1", "grant", "admin") httpd = ThreadingHTTPServer(("127.0.0.1", 0), _Quiet) port = httpd.server_address[1] threading.Thread(target=httpd.serve_forever, daemon=True).start() try: # ── create with priority:true seeds the row's Priority flag ── print("\n[create: optional initial priority flag honored]") st, d = _create(port, token, "Acme Capital", "Jane Doe", priority=True, note="Intro call", note_append=True) row = (d or {}).get("data", {}).get("row") or {} acme_id = row.get("id") check(st == 201, f"create -> 201 (got {st})") check(row.get("priority") is True, f"returned row carries priority=true (got {row.get('priority')!r})") rows = _grid_rows(port, token) check(rows.get(acme_id, {}).get("priority") is True, f"GET /state shows the new row priority=true (got {rows.get(acme_id, {}).get('priority')!r})") check(len(rows.get(acme_id, {}).get("contacts", [])) == 1, f"new row has its first contact (got {rows.get(acme_id, {}).get('contacts')})") # ── create without priority defaults to False (no accidental flag) ── print("\n[create: priority defaults False when omitted]") st, d = _create(port, token, "Beta Partners", "Pat Roe") # no note, no priority beta = (d or {}).get("data", {}).get("row") or {} beta_id = beta.get("id") check(st == 201, f"no-note create -> 201 (got {st})") check(beta.get("priority") is False, f"omitted priority -> False (got {beta.get('priority')!r})") # ── priority is honored ONLY on the create branch: logging against an EXISTING row # with priority:true must not flip its flag (Beta was created without priority) ── print("\n[invariant: priority on an existing-row log does NOT change its flag]") st, _ = _req(port, "POST", "/api/fundraising/log-communication", token, { "row_id": beta_id, "type": "note", "body": "follow-up", "append_note": True, "priority": True, }) check(st in (200, 201), f"log against existing Beta -> ok (got {st})") rows = _grid_rows(port, token) check(rows.get(beta_id, {}).get("priority") is False, f"existing-row priority untouched by the log's priority flag (got {rows.get(beta_id, {}).get('priority')!r})") # ── create -> link handshake: the brand-new row links at the chosen stage ── print("\n[create -> link: freshly-created row resolves for pipeline link at stage]") st, d = _req(port, "POST", "/api/fundraising/pipeline/link", token, { "source_row_id": acme_id, "contact_index": 0, "name": "Acme Capital — Pipeline", "stage": "engaged", "expected_amount": 0, "probability": 55, "fund_name": "", }) opp = (d or {}).get("data") or {} check(st == 201 and opp.get("stage") == "engaged", f"link new row @engaged -> 201 (got {st}, stage={opp.get('stage')})") rows = _grid_rows(port, token) check(rows.get(acme_id, {}).get("pipeline") is True and rows.get(acme_id, {}).get("pipeline_stage") == "engaged", f"new row now in pipeline @engaged (got {rows.get(acme_id, {}).get('pipeline')}, " f"{rows.get(acme_id, {}).get('pipeline_stage')})") # ── create -> reminder handshake: source_row_id resolves to the synced investor ── print("\n[create -> reminder: source_row_id resolves to the new investor]") st, d = _req(port, "POST", "/api/reminders", token, { "source_row_id": acme_id, "investor_name": "Acme Capital", "title": "Send Fund III deck", "due_date": "2026-07-01", "details": "", }) rem = (d or {}).get("data") or {} check(st == 201, f"reminder create -> 201 (got {st})") check(bool(rem.get("investor_id")) and rem.get("investor_name") == "Acme Capital", f"reminder linked to the new investor (got id={rem.get('investor_id')!r}, " f"name={rem.get('investor_name')!r})") # ── card-intake contact fields land on the canonical contact (phone/city/linkedin) ── # The Matrix card flow sends these on the contact dict; the upsert must persist them. print("\n[contact fields: phone + city + linkedin persist on the contact]") st, d = _req(port, "POST", "/api/fundraising/log-communication", token, { "investor_name": "Fortitude Investment Group", "create_investor_if_missing": True, "contact": {"name": "Daniel Raupp", "email": "draupp@fortitude.example", "phone": "631-474-5610", "mobile": "631-922-1195", "city": "Setauket, NY", "linkedin_url": "linkedin.com/in/danielraupp"}, "type": "note", "body": "from a business card", "append_note": True, }) check(st == 201, f"create with contact fields -> 201 (got {st})") c = _db() crow = c.execute("SELECT phone, mobile, city, linkedin_url FROM contacts WHERE lower(email) = ?", ("draupp@fortitude.example",)).fetchone() c.close() check(crow is not None, "contact row exists") check(bool(crow) and crow[0] == "631-474-5610", f"phone (office) persisted (got {crow[0] if crow else None!r})") check(bool(crow) and crow[1] == "631-922-1195", f"mobile (cell) persisted (got {crow[1] if crow else None!r})") check(bool(crow) and crow[2] == "Setauket, NY", f"city persisted (got {crow[2] if crow else None!r})") check(bool(crow) and crow[3] == "linkedin.com/in/danielraupp", f"linkedin persisted (got {crow[3] if crow else None!r})") # ── unknown source_row_id is refused (guard) ── print("\n[guard: reminder on an unknown source_row_id -> 404]") st, _ = _req(port, "POST", "/api/reminders", token, { "source_row_id": "nope", "title": "x", }) check(st == 404, f"unknown source_row_id -> 404 (got {st})") finally: httpd.shutdown() print("\n" + ("ALL PASS" if not FAILS else f"{len(FAILS)} FAILURE(S):")) for f in FAILS: print(" - " + f) sys.exit(1 if FAILS else 0) if __name__ == "__main__": main()