expand manual prospect intelligence workflows

2026-09-02 17:57:17 +02:00
parent d6a5354eea
commit bc33b03075
12 changed files with 477 additions and 347 deletions
@@ -101,6 +101,58 @@ class ApiSmokeTests(unittest.TestCase):
        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()