ten31-database/backend/test_purge_soft_deleted.py

#!/usr/bin/env python3
"""Test for the admin soft-deleted purge (v0.1.0:104).

The purge is a deliberate, admin-only, type-to-confirm exception to never-hard-delete, for
clearing dummy/test data. It must be SAFE: only ever touch a soft-deleted row, and never
remove or mutate LIVE data via a cascade/SET-NULL. This boots the real server, seeds live +
soft-deleted graphs, and drives /api/admin/soft-deleted[/purge] over HTTP. Synthetic only.

Run: cd backend && python3 test_purge_soft_deleted.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 = []
DEL = "2026-06-01T00:00:00"


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 _post(port, path, token, payload):
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
    conn.request("POST", path, body=json.dumps(payload),
                 headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"})
    resp = conn.getresponse()
    body = resp.read().decode("utf-8", "replace")
    conn.close()
    try:
        return resp.status, (json.loads(body) if body else None)
    except ValueError:
        return resp.status, None


def _get(port, path, token):
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
    conn.request("GET", path, headers={"Authorization": "Bearer " + token})
    resp = conn.getresponse()
    body = resp.read().decode("utf-8", "replace")
    conn.close()
    try:
        return resp.status, (json.loads(body) if body else None)
    except ValueError:
        return resp.status, None


def exists(table, rid):
    c = sqlite3.connect(os.environ["CRM_DB_PATH"])
    n = c.execute(f"SELECT COUNT(*) FROM {table} WHERE id = ?", (rid,)).fetchone()[0]
    c.close()
    return n > 0


def seed():
    c = sqlite3.connect(os.environ["CRM_DB_PATH"])
    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)")
    # Soft-deleted contact with ONLY soft-deleted children -> purgeable; cascade should remove them.
    c.execute("INSERT INTO contacts (id,first_name,last_name,deleted_at) VALUES ('cClean','Dummy','Clean',?)", (DEL,))
    c.execute("INSERT INTO opportunities (id,name,contact_id,owner_id,deleted_at) VALUES ('opC','Opp','cClean','u1',?)", (DEL,))
    c.execute("INSERT INTO communications (id,contact_id,communication_date,created_by,subject,deleted_at) VALUES ('cmC','cClean','2026-05-01','u1','note',?)", (DEL,))
    # A reminder pointing at the purge target (reminders.contact_id is a bare logical FK, no ON DELETE):
    # the purge must NULL it, not leave it dangling and not delete the reminder.
    c.execute("INSERT INTO reminders (id,contact_id,investor_id,title) VALUES ('remC','cClean','inv-x','Follow up dummy')")
    # Soft-deleted contact WITH a live child -> must refuse (cascade would kill live data).
    c.execute("INSERT INTO contacts (id,first_name,last_name,deleted_at) VALUES ('cLiveKid','Has','Livekid',?)", (DEL,))
    c.execute("INSERT INTO communications (id,contact_id,communication_date,created_by,subject) VALUES ('cmLive','cLiveKid','2026-05-02','u1','live note')")
    # A live contact -> must refuse (not soft-deleted).
    c.execute("INSERT INTO contacts (id,first_name,last_name) VALUES ('cLive','Real','Person')")
    # Soft-deleted org with no live refs -> purgeable.
    c.execute("INSERT INTO organizations (id,name,deleted_at) VALUES ('orgClean','Dead Org',?)", (DEL,))
    # Soft-deleted org referenced by a LIVE contact -> must refuse (SET NULL would mutate live data).
    c.execute("INSERT INTO organizations (id,name,deleted_at) VALUES ('orgRef','Ref Org',?)", (DEL,))
    c.execute("INSERT INTO contacts (id,first_name,last_name,organization_id) VALUES ('cRef','Org','Member','orgRef')")
    c.commit()
    c.close()


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:
        print("\n[list soft-deleted]")
        st, body = _get(port, "/api/admin/soft-deleted", token)
        groups = (body or {}).get("groups", {})
        cids = {x["id"] for x in groups.get("contacts", [])}
        oids = {x["id"] for x in groups.get("organizations", [])}
        check(st == 200, f"GET soft-deleted -> 200 (got {st})")
        check({"cClean", "cLiveKid"} <= cids and "cLive" not in cids, f"lists soft-deleted contacts only (got {cids})")
        check({"orgClean", "orgRef"} <= oids, f"lists soft-deleted orgs (got {oids})")
        check("opC" in {x["id"] for x in groups.get("opportunities", [])}, "lists the soft-deleted opportunity")

        print("\n[purge guards]")
        st, _ = _post(port, "/api/admin/soft-deleted/purge", token, {"table": "contacts", "id": "cLive"})
        check(st == 400, f"purge a LIVE contact -> 400 (got {st})")
        check(exists("contacts", "cLive"), "live contact still present after refused purge")
        st, _ = _post(port, "/api/admin/soft-deleted/purge", token, {"table": "contacts", "id": "cLiveKid"})
        check(st == 409, f"purge contact with a LIVE child -> 409 (got {st})")
        check(exists("contacts", "cLiveKid") and exists("communications", "cmLive"), "contact + its live child preserved")
        st, _ = _post(port, "/api/admin/soft-deleted/purge", token, {"table": "organizations", "id": "orgRef"})
        check(st == 409, f"purge org referenced by a LIVE contact -> 409 (got {st})")
        check(exists("organizations", "orgRef") and exists("contacts", "cRef"), "org + its live referencing contact preserved")
        st, _ = _post(port, "/api/admin/soft-deleted/purge", token, {"table": "bogus", "id": "x"})
        check(st == 400, f"unknown table -> 400 (got {st})")
        st, _ = _post(port, "/api/admin/soft-deleted/purge", token, {"table": "contacts", "id": "nope"})
        check(st == 404, f"missing id -> 404 (got {st})")

        print("\n[purge happy path + cascade]")
        st, _ = _post(port, "/api/admin/soft-deleted/purge", token, {"table": "contacts", "id": "cClean"})
        check(st == 200, f"purge a clean soft-deleted contact -> 200 (got {st})")
        check(not exists("contacts", "cClean"), "purged contact is gone")
        check(not exists("opportunities", "opC") and not exists("communications", "cmC"),
              "its soft-deleted children were cascade-removed")
        _rc = sqlite3.connect(os.environ["CRM_DB_PATH"])
        _rem = _rc.execute("SELECT contact_id FROM reminders WHERE id = 'remC'").fetchone()
        _rc.close()
        check(_rem is not None and _rem[0] is None,
              "a reminder on the purged contact is KEPT but its contact_id is NULL'd (no dangling ref)")
        st, _ = _post(port, "/api/admin/soft-deleted/purge", token, {"table": "organizations", "id": "orgClean"})
        check(st == 200, f"purge a clean soft-deleted org -> 200 (got {st})")
        check(not exists("organizations", "orgClean"), "purged org is gone")
    finally:
        httpd.shutdown()

    print()
    if FAILS:
        print(f"{len(FAILS)} FAILED")
        sys.exit(1)
    print("ALL PASS (soft-deleted purge)")


if __name__ == "__main__":
    main()