Spaces:
Running
Running
Z User
feat: ABM Engine v1 — OpenSearch store + Apify enrichment + ICP scoring + k-NN memory + optional Neo4j mirror
fb0ec21 | """Unit tests — normalization, scoring, embeddings, auth, API contract.""" | |
| import unittest | |
| from unittest.mock import AsyncMock, MagicMock, patch | |
| from fastapi.testclient import TestClient | |
| from app.main import app | |
| from app.stores.apify import normalize_account, normalize_contact | |
| from app.services.signals import score_account | |
| client = TestClient(app) | |
| class NormalizerTests(unittest.TestCase): | |
| """Actor output is schema-flexible; the normalizer must survive it.""" | |
| def test_google_maps_item_full(self): | |
| raw = { | |
| "title": "Acme Robotics", | |
| "website": "https://www.acmerobotics.com/contact", | |
| "phoneNumber": "+1 555 0100", | |
| "category": "Industrial equipment supplier", | |
| "city": "Detroit", | |
| "countryCode": "US", | |
| "totalScore": 4.6, | |
| "description": "Industrial robots", | |
| } | |
| out = normalize_account(raw) | |
| self.assertEqual(out["name"], "Acme Robotics") | |
| self.assertEqual(out["domain"], "acmerobotics.com") | |
| self.assertEqual(out["phone"], "+1 555 0100") | |
| self.assertEqual(out["source"], "google_maps") | |
| def test_google_maps_item_missing_name_is_skipped(self): | |
| self.assertIsNone(normalize_account({"website": "https://x.com"})) | |
| def test_linkedin_contact(self): | |
| raw = { | |
| "fullName": "Jane Doe", | |
| "email": "Jane.Doe@Example.com ", | |
| "position": "Head of Ops", | |
| "linkedinUrl": "https://linkedin.com/in/janedoe", | |
| "company": "Acme", | |
| } | |
| out = normalize_contact(raw) | |
| self.assertEqual(out["email"], "jane.doe@example.com") | |
| self.assertEqual(out["title"], "Head of Ops") | |
| def test_contact_requires_email_or_name(self): | |
| self.assertIsNone(normalize_contact({"title": "CEO"})) | |
| class ScoringTests(unittest.TestCase): | |
| def test_full_fit_account_scores_A(self): | |
| account = { | |
| "name": "Great Co", "domain": "great.com", "website": "great.com", | |
| "industry": "Software", "hq_country": "United States", | |
| "employee_count": 120, "rating": 4.8, | |
| } | |
| out = score_account(account, contact_count=5) | |
| self.assertGreaterEqual(out["icp_score"], 75) | |
| self.assertEqual(out["icp_label"], "A") | |
| self.assertIn("industry_fit", out["breakdown"]) | |
| def test_poor_fit_account_scores_D(self): | |
| out = score_account({"name": "Nobody"}, contact_count=0) | |
| self.assertLess(out["icp_score"], 30) | |
| self.assertEqual(out["icp_label"], "D") | |
| def test_score_is_explainable_and_bounded(self): | |
| out = score_account({"name": "X", "industry": "software", "hq_country": "usa", | |
| "employee_count": 50, "rating": 5.0}, contact_count=99) | |
| self.assertLessEqual(out["icp_score"], 100) | |
| self.assertEqual(out["breakdown"]["contact_coverage"], 10.0) | |
| class EmbeddingTests(unittest.TestCase): | |
| def test_hash_embedding_deterministic_and_normalized(self): | |
| from app.stores.embeddings import HashEmbedding | |
| e1 = HashEmbedding().encode("acme robotics detroit") | |
| e2 = HashEmbedding().encode("acme robotics detroit") | |
| e3 = HashEmbedding().encode("completely different text") | |
| self.assertEqual(e1, e2) | |
| self.assertNotEqual(e1, e3) | |
| self.assertEqual(len(e1), 384) | |
| norm = sum(v * v for v in e1) ** 0.5 | |
| self.assertAlmostEqual(norm, 1.0, places=2) | |
| class AuthTests(unittest.TestCase): | |
| def test_api_key_auth(self): | |
| from app.config import settings | |
| with patch.object(settings, "api_key", "test-key-123"): | |
| from app.auth import authenticate | |
| req = MagicMock() | |
| req.headers = {"x-api-key": "test-key-123"} | |
| import asyncio | |
| out = asyncio.run(authenticate(req)) | |
| self.assertEqual(out["via"], "api-key") | |
| self.assertEqual(out["user_id"], "service") | |
| class HealthTests(unittest.TestCase): | |
| def test_health_is_open_and_reports_components(self): | |
| res = client.get("/health") | |
| self.assertEqual(res.status_code, 200) | |
| body = res.json() | |
| self.assertIn("components", body) | |
| self.assertIn("embeddings", body["components"]) | |
| self.assertIn("status", body) | |
| class ApiContractTests(unittest.TestCase): | |
| def test_enrich_rejects_unknown_source(self): | |
| res = client.post("/api/v1/enrich", json={"source": "twitter", "queries": ["x"]}) | |
| self.assertEqual(res.status_code, 400) | |
| def test_accounts_require_domain_or_name(self): | |
| res = client.post("/api/v1/accounts", json={"industry": "software"}) | |
| self.assertEqual(res.status_code, 400) | |
| def test_memory_rejects_empty_text(self): | |
| res = client.post("/api/v1/memory", json={"text": " "}) | |
| self.assertEqual(res.status_code, 400) | |
| if __name__ == "__main__": | |
| unittest.main() | |