import json import os import sqlite3 import threading import unittest from http.client import HTTPConnection from tempfile import TemporaryDirectory from app.main import create_server, hash_password class ApiSmokeTests(unittest.TestCase): def setUp(self): self.tmp = TemporaryDirectory() self.old_env = {key: os.environ.get(key) for key in ("BOOTSTRAP_ADMIN_EMAIL", "BOOTSTRAP_ADMIN_PASSWORD")} os.environ["BOOTSTRAP_ADMIN_EMAIL"] = "owner@example.test" os.environ["BOOTSTRAP_ADMIN_PASSWORD"] = "development-password" self.db_path = self.tmp.name + "/test.db" self.server = create_server("127.0.0.1", 0, self.db_path) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.conn = HTTPConnection("127.0.0.1", self.server.server_port, timeout=3) self.cookie = None status, _ = self.request("POST", "/api/v1/auth/login", {"email": "owner@example.test", "password": "development-password"}) self.assertEqual(status, 200) def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=2) for key, value in self.old_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value self.tmp.cleanup() def request(self, method, path, payload=None, cookie=True): body = json.dumps(payload).encode() if payload is not None else None headers = {"Content-Type": "application/json"} if body else {} if cookie and self.cookie: headers["Cookie"] = self.cookie self.conn.request(method, path, body, headers) response = self.conn.getresponse() set_cookie = response.getheader("Set-Cookie") if set_cookie and "session=" in set_cookie: self.cookie = set_cookie.split(";", 1)[0] return response.status, json.loads(response.read() or b"{}") def test_auth_login_me_logout_and_protected_route(self): self.assertEqual(self.request("GET", "/api/v1/auth/me")[0], 200) self.assertEqual(self.request("POST", "/api/v1/auth/logout")[0], 200) self.assertEqual(self.request("GET", "/api/v1/dashboard/summary")[0], 401) def test_viewer_cannot_mutate(self): password_hash, salt = hash_password("viewer-password") db = sqlite3.connect(self.db_path) db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ("demo-tenant", "viewer@example.test", password_hash, salt, "viewer")) db.commit(); db.close() self.cookie = None self.assertEqual(self.request("POST", "/api/v1/auth/login", {"email": "viewer@example.test", "password": "viewer-password"})[0], 200) self.assertEqual(self.request("POST", "/api/v1/businesses", {"name": "Nope"})[0], 403) def test_cross_organization_businesses_are_isolated(self): password_hash, salt = hash_password("other-password") db = sqlite3.connect(self.db_path) db.execute("INSERT INTO organizations (id,name) VALUES (?,?)", ("other-tenant", "Other")) db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ("other-tenant", "other@example.test", password_hash, salt, "owner")) db.commit(); db.close() self.cookie = None self.assertEqual(self.request("POST", "/api/v1/auth/login", {"email": "other@example.test", "password": "other-password"})[0], 200) self.assertEqual(self.request("POST", "/api/v1/businesses", {"name": "Other Co", "website": "https://other.test"})[0], 201) self.assertEqual(self.request("GET", "/api/v1/businesses")[1]["organization_id"], "other-tenant") self.cookie = None self.assertEqual(self.request("POST", "/api/v1/auth/login", {"email": "owner@example.test", "password": "development-password"})[0], 200) status, businesses = self.request("GET", "/api/v1/businesses") self.assertEqual(status, 200) self.assertEqual(businesses["items"], []) def test_missing_auth_is_rejected(self): self.cookie = None self.assertEqual(self.request("GET", "/api/v1/businesses", cookie=False)[0], 401) def test_health_create_get_summary_and_import_preview(self): self.assertEqual(self.request("GET", "/api/v1/health/live")[0], 200) status, created = self.request("POST", "/api/v1/businesses", {"name": "Acme", "website": "https://acme.co.za", "email": "a@acme.co.za"}) self.assertEqual(status, 201) self.assertEqual(created["organization_id"], "demo-tenant") status, fetched = self.request("GET", "/api/v1/businesses/" + str(created["id"])) self.assertEqual(status, 200) self.assertEqual(fetched["name"], "Acme") status, summary = self.request("GET", "/api/v1/dashboard/summary") self.assertEqual(status, 200) self.assertEqual(summary["businesses"], 1) status, preview = self.request("POST", "/api/v1/imports/preview", {"rows": [{"name": "Acme", "website": "https://acme.co.za"}, {"name": "New Co"}]}) self.assertEqual(status, 200) self.assertEqual(preview["accepted"], 1) self.assertEqual(preview["duplicates"], 1) def test_suppression_blocks_new_business(self): status, _ = self.request("POST", "/api/v1/suppressions", {"kind": "domain", "value": "blocked.co.za"}) self.assertEqual(status, 201) status, response = self.request("POST", "/api/v1/businesses", {"name": "Blocked", "website": "https://blocked.co.za"}) self.assertEqual(status, 409) self.assertEqual(response["error"], "suppressed") def test_business_detail_contains_nested_phase_three_resources_and_mutations_audit(self): status, business = self.request("POST", "/api/v1/businesses", {"name": "Nested Co", "website": "https://nested.test"}) self.assertEqual(status, 201) bid = business["id"] for path, payload in [ ("contacts", {"name": "Jane", "email": "jane@nested.test"}), ("domains", {"domain": "nested.test", "kind": "primary"}), ("websites", {"url": "https://nested.test", "website_class": "business_site"}), ("evidence", {"kind": "source", "url": "https://source.test", "claim": "Founded 2020"}), ("notes", {"body": "Call next week"}), ]: self.assertEqual(self.request("POST", f"/api/v1/businesses/{bid}/{path}", payload)[0], 201) self.assertEqual(self.request("PATCH", f"/api/v1/businesses/{bid}/pipeline", {"stage": "qualified"})[0], 200) self.assertEqual(self.request("POST", f"/api/v1/businesses/{bid}/verify", {"verified": True})[0], 200) status, detail = self.request("GET", f"/api/v1/businesses/{bid}") self.assertEqual(status, 200) for key in ("contacts", "domains", "websites", "evidence", "pipeline", "notes"): self.assertEqual(len(detail[key]), 1, key) self.assertTrue(detail["verified"]) db = sqlite3.connect(self.db_path) self.assertGreaterEqual(db.execute("SELECT COUNT(*) FROM audit_log WHERE organization_id='demo-tenant'").fetchone()[0], 8) db.close() def test_suppressed_contact_is_do_not_contact(self): self.assertEqual(self.request("POST", "/api/v1/suppressions", {"kind": "email", "value": "blocked@co.test"})[0], 201) _, business = self.request("POST", "/api/v1/businesses", {"name": "Contact Co"}) status, contact = self.request("POST", f"/api/v1/businesses/{business['id']}/contacts", {"email": "blocked@co.test"}) self.assertEqual(status, 201) self.assertTrue(contact["do_not_contact"]) def test_business_list_pagination_and_filters(self): for name, website in [("Alpha", "https://alpha.test"), ("Beta", "https://beta.test"), ("Gamma", "https://gamma.test")]: self.assertEqual(self.request("POST", "/api/v1/businesses", {"name": name, "website": website})[0], 201) status, page = self.request("GET", "/api/v1/businesses?page=1&page_size=2&score_min=20&q=Alpha") self.assertEqual(status, 200) self.assertEqual([x["name"] for x in page["items"]], ["Alpha"]) status, invalid = self.request("GET", "/api/v1/businesses?page_size=0") self.assertEqual(status, 400) self.assertEqual(invalid["error"], "invalid_pagination") def test_child_resources_are_tenant_scoped_and_validated(self): _, business = self.request("POST", "/api/v1/businesses", {"name": "Private Co"}) self.assertEqual(self.request("POST", f"/api/v1/businesses/{business['id']}/contacts", {"email": "bad"})[0], 400) password_hash, salt = hash_password("other-password") db = sqlite3.connect(self.db_path) db.execute("INSERT INTO organizations (id,name) VALUES (?,?)", ("other-tenant", "Other")) db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ("other-tenant", "other@example.test", password_hash, salt, "owner")) db.commit(); db.close() self.cookie = None self.assertEqual(self.request("POST", "/api/v1/auth/login", {"email": "other@example.test", "password": "other-password"})[0], 200) self.assertEqual(self.request("POST", f"/api/v1/businesses/{business['id']}/notes", {"body": "leak"})[0], 404) if __name__ == "__main__": unittest.main()