Spaces:
Running
Running
File size: 4,915 Bytes
fb0ec21 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """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()
|