File size: 4,822 Bytes
de745bf | 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 | """Worker endpoint tests (spec §3.3) — POST /api/v1/smtp/probe*.
Uses FastAPI TestClient against the real router so auth open-prefix handling
and the X-Worker-Token guard are exercised end-to-end.
"""
import unittest
from unittest.mock import AsyncMock, patch
from fastapi.testclient import TestClient
from app.config import settings
from app.main import app
from app.models import SMTPResult
def _client():
return TestClient(app)
class WorkerEndpointTests(unittest.TestCase):
def setUp(self):
self.probe_payload = {"email": "a@corp.test", "mx_host": "mx.corp.test"}
def test_mailbox_probe_returns_smtp_result(self):
with patch(
"app.validator.mailbox_provider.AutonomousSmtpProvider.verify",
new=AsyncMock(return_value=SMTPResult(verified=True, smtp_code=250)),
):
with _client() as client:
res = client.post("/api/v1/smtp/probe", json=self.probe_payload)
self.assertEqual(res.status_code, 200)
body = res.json()
self.assertTrue(body["verified"])
self.assertEqual(body["smtp_code"], 250)
def test_catch_all_probe_shape(self):
with patch(
"app.validator.mailbox_provider.AutonomousSmtpProvider.catch_all",
new=AsyncMock(return_value=(True, 0.95)),
):
with _client() as client:
res = client.post(
"/api/v1/smtp/probe",
json={"domain": "corp.test", "mx_host": "mx.corp.test",
"probe": "catch_all"},
)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.json(), {"is_catch_all": True, "confidence": 0.95})
def test_batch_probe_roundtrip(self):
with patch(
"app.validator.mailbox_provider.AutonomousSmtpProvider.verify_many",
new=AsyncMock(return_value=(
{"a@corp.test": SMTPResult(verified=True, smtp_code=250)},
{"corp.test": False},
)),
):
with _client() as client:
res = client.post(
"/api/v1/smtp/probe/batch",
json={"pairs": [["a@corp.test", "mx.corp.test"]]},
)
self.assertEqual(res.status_code, 200)
body = res.json()
self.assertTrue(body["results"]["a@corp.test"]["verified"])
self.assertEqual(body["catch_all"], {"corp.test": False})
def test_token_guard_rejects_bad_token(self):
with patch.object(settings, "smtp_worker_token", "s3cret"):
with _client() as client:
res = client.post("/api/v1/smtp/probe", json=self.probe_payload)
self.assertEqual(res.status_code, 401)
def test_token_guard_accepts_good_token(self):
with patch.object(settings, "smtp_worker_token", "s3cret"), patch(
"app.validator.mailbox_provider.AutonomousSmtpProvider.verify",
new=AsyncMock(return_value=SMTPResult(verified=True, smtp_code=250)),
):
with _client() as client:
res = client.post(
"/api/v1/smtp/probe",
json=self.probe_payload,
headers={"X-Worker-Token": "s3cret"},
)
self.assertEqual(res.status_code, 200)
def test_probe_requires_email_or_domain(self):
with _client() as client:
res = client.post("/api/v1/smtp/probe", json={"mx_host": "mx.corp.test"})
self.assertEqual(res.status_code, 422)
class OpsEndpointTests(unittest.TestCase):
def test_engine_status_shape(self):
with _client() as client:
res = client.get("/api/v1/engine/status")
self.assertEqual(res.status_code, 200)
body = res.json()
self.assertEqual(body["version"], settings.app_version)
self.assertIn("dns", body)
self.assertIn("smtp", body)
self.assertIn("ai_scorer", body)
self.assertIn("egress_port25", body["smtp"])
def test_inference_metrics_shape(self):
with _client() as client:
res = client.get("/api/v1/metrics/inference")
self.assertEqual(res.status_code, 200)
body = res.json()
for key in ("verifications_scored", "inference_calls", "heuristic_calls",
"burn_rate", "burn_rate_target", "burn_rate_ok"):
self.assertIn(key, body)
def test_health_carries_workers_summary(self):
with _client() as client:
res = client.get("/api/v1/health")
self.assertEqual(res.status_code, 200)
body = res.json()
self.assertIn("workers", body)
self.assertIn("ai_scorer", body["workers"])
self.assertEqual(body["version"], settings.app_version)
if __name__ == "__main__":
unittest.main()
|