FortyDN Dev
v3.0.0: fully autonomous engine — remove MillionVerifier, add DoH DNS worker, AI risk scorer (HF PRO, circuit breaker), i18n, normalized verdicts, remote SMTP worker
de745bf | """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() | |