"""Phase 2 tests: provider registry, HMAC signing, webhook contracts, utterance/evaluation ingestion, cancel endpoint. Run: python manage.py test apps.ai_calls """ import json from django.contrib.auth import get_user_model from django.test import TestCase, override_settings from rest_framework.test import APIClient from apps.candidates.models import Candidate from apps.jobs.models import JobDescription from apps.pipeline.models import JobApplication, PipelineStage from . import security from .models import AICall, CallEvaluation, CallTranscriptUtterance from .providers import get_provider, ProviderNotConfigured User = get_user_model() class SecurityTests(TestCase): def test_sign_verify_roundtrip(self): body = b'{"call_id": "abc"}' headers = security.sign_headers("s3cret", body) self.assertTrue(security.verify_signature("s3cret", body, headers[security.HEADER])) def test_tampered_body_fails(self): headers = security.sign_headers("s3cret", b'{"a":1}') self.assertFalse(security.verify_signature("s3cret", b'{"a":2}', headers[security.HEADER])) def test_wrong_secret_fails(self): headers = security.sign_headers("s3cret", b"x") self.assertFalse(security.verify_signature("other", b"x", headers[security.HEADER])) def test_stale_timestamp_fails(self): headers = security.sign_headers("s3cret", b"x", timestamp=1000) self.assertFalse(security.verify_signature("s3cret", b"x", headers[security.HEADER], now=1000 + 301)) self.assertTrue(security.verify_signature("s3cret", b"x", headers[security.HEADER], now=1000 + 299)) def test_garbage_header_fails(self): self.assertFalse(security.verify_signature("s3cret", b"x", "not-a-signature")) self.assertFalse(security.verify_signature("s3cret", b"x", "")) class ProviderRegistryTests(TestCase): def test_registry_selects_by_setting(self): with override_settings(AI_CALL_PROVIDER="mock", AI_CALL_ALLOW_SIMULATED=True): self.assertEqual(get_provider().name, "mock") with override_settings(AI_CALL_PROVIDER="selfhosted"): self.assertEqual(get_provider().name, "selfhosted") def test_unknown_provider_raises_error(self): with override_settings(AI_CALL_PROVIDER="banana"): with self.assertRaises(ProviderNotConfigured): get_provider() class BaseCallTestCase(TestCase): """Shared fixtures: user, candidate, job, AI stages, application, call.""" def setUp(self): self.client = APIClient() self.admin = User.objects.create_user( email="admin@test.local", username="admin-test", password="x", role="ADMIN" ) self.candidate = Candidate.objects.create( first_name="Test", last_name="Candidate", phone_number="9999900000", email="cand@test.local", ) self.job = JobDescription.objects.create( title="Python Developer", location="Remote", work_details="Build APIs", status="Published", created_by=self.admin, ) for i, (name, code) in enumerate( [("AI Calling", "ai_calling"), ("AI Screened", "ai_screened"), ("AI Qualified", "ai_qualified")], start=1, ): PipelineStage.objects.get_or_create(code=code, defaults={"name": name, "sort_order": i}) self.application = JobApplication.objects.create( candidate=self.candidate, job=self.job, stage=PipelineStage.objects.get(code="ai_calling"), created_by=self.admin, ) self.call = AICall.objects.create( candidate=self.candidate, job=self.job, application=self.application, provider_call_id="platform-123", status=AICall.Status.IN_PROGRESS, agent_variables={"phone": "9999900000"}, ) def post_webhook(self, path, payload, sign_with=None, headers=None): body = json.dumps(payload) extra = dict(headers or {}) if sign_with: extra.update(security.sign_headers(sign_with, body.encode())) return self.client.post(path, data=body, content_type="application/json", headers=extra) @override_settings(AI_PLATFORM_SIGNING_SECRET="") class WebhookIngestionTests(BaseCallTestCase): def test_transcript_webhook_stores_utterances_and_blob(self): res = self.post_webhook("/api/v1/ai-calls/transcript/", { "call_id": "platform-123", "utterances": [ {"sequence": 1, "speaker": "AGENT", "message": "Hello!", "started_ms": 0, "language": "en"}, {"sequence": 2, "speaker": "CANDIDATE", "message": "Hi.", "started_ms": 2000, "language": "en"}, ], }) self.assertEqual(res.status_code, 200) self.assertEqual(CallTranscriptUtterance.objects.filter(call=self.call).count(), 2) self.call.refresh_from_db() self.assertIn("Agent: Hello!", self.call.transcript) self.assertIn("Candidate: Hi.", self.call.transcript) def test_transcript_batches_are_idempotent(self): payload = { "call_id": "platform-123", "utterances": [{"sequence": 1, "speaker": "AGENT", "message": "Hello!"}], } self.post_webhook("/api/v1/ai-calls/transcript/", payload) payload["utterances"][0]["message"] = "Hello again!" # re-delivery with update self.post_webhook("/api/v1/ai-calls/transcript/", payload) utterances = CallTranscriptUtterance.objects.filter(call=self.call) self.assertEqual(utterances.count(), 1) self.assertEqual(utterances.first().message, "Hello again!") def test_completed_webhook_stores_evaluation_and_updates_pipeline(self): res = self.post_webhook("/api/v1/ai-calls/completed/", { "call_id": "platform-123", "status": "completed", "duration": 240, "utterances": [{"sequence": 1, "speaker": "AGENT", "message": "Summary time."}], "evaluation": { "technical_score": 80, "communication_score": 75, "experience_score": 70, "confidence_score": 72, "overall_score": 78, "classification": "Strong Match", "recommendation": "Proceed", "strengths": ["Python depth"], "weaknesses": ["Cloud exposure"], "summary": "Good screening call.", "rubric": {"q1": {"score": 8}}, }, }) self.assertEqual(res.status_code, 200) self.call.refresh_from_db() evaluation = CallEvaluation.objects.get(call=self.call) self.assertEqual(evaluation.overall_score, 78) self.assertEqual(evaluation.classification, "Strong Match") # score falls back to the evaluation's overall; Proceed -> QUALIFIED self.assertEqual(self.call.score, 78) self.assertEqual(self.call.status, AICall.Status.COMPLETED) self.assertEqual(self.call.recommendation, AICall.Recommendation.QUALIFIED) # pipeline side effects self.application.refresh_from_db() self.assertEqual(self.application.stage.code, "ai_qualified") self.assertEqual(self.application.score, 78) def test_completed_hold_maps_to_review_and_screened_stage(self): self.post_webhook("/api/v1/ai-calls/completed/", { "call_id": "platform-123", "status": "completed", "evaluation": {"overall_score": 58, "classification": "Potential Match", "recommendation": "Hold"}, }) self.call.refresh_from_db() self.assertEqual(self.call.recommendation, AICall.Recommendation.REVIEW) self.application.refresh_from_db() self.assertEqual(self.application.stage.code, "ai_screened") def test_unknown_call_id_returns_404(self): res = self.post_webhook("/api/v1/ai-calls/webhook/", {"call_id": "nope", "status": "completed"}) self.assertEqual(res.status_code, 404) @override_settings(AI_PLATFORM_SIGNING_SECRET="topsecret") class WebhookSignatureTests(BaseCallTestCase): def test_valid_signature_accepted(self): res = self.post_webhook( "/api/v1/ai-calls/webhook/", {"call_id": "platform-123", "status": "in_progress"}, sign_with="topsecret", ) self.assertEqual(res.status_code, 200) self.call.refresh_from_db() self.assertEqual(self.call.status, AICall.Status.IN_PROGRESS) def test_invalid_signature_rejected(self): res = self.post_webhook( "/api/v1/ai-calls/webhook/", {"call_id": "platform-123", "status": "failed"}, sign_with="wrong-secret", ) self.assertEqual(res.status_code, 403) def test_unsigned_requests_rejected_when_secret_configured(self): res = self.post_webhook("/api/v1/ai-calls/webhook/", {"call_id": "platform-123", "status": "in_progress"}) self.assertEqual(res.status_code, 403) @override_settings(AI_PLATFORM_SIGNING_SECRET="") def test_unsigned_requests_pass_in_dev_without_secret(self): res = self.post_webhook("/api/v1/ai-calls/webhook/", {"call_id": "platform-123", "status": "in_progress"}) self.assertEqual(res.status_code, 200) @override_settings(AI_CALL_PROVIDER="mock", AI_CALL_ALLOW_SIMULATED=True) class CancelEndpointTests(BaseCallTestCase): def test_cancel_running_call(self): self.client.force_authenticate(self.admin) res = self.client.post(f"/api/v1/ai-calls/{self.call.id}/cancel/") self.assertEqual(res.status_code, 200) self.call.refresh_from_db() self.assertEqual(self.call.status, AICall.Status.CANCELLED) self.assertEqual(self.call.error_message, "Cancelled by recruiter") def test_cannot_cancel_finished_call(self): self.call.status = AICall.Status.COMPLETED self.call.save() self.client.force_authenticate(self.admin) res = self.client.post(f"/api/v1/ai-calls/{self.call.id}/cancel/") self.assertEqual(res.status_code, 400) def test_candidate_role_cannot_cancel(self): viewer = User.objects.create_user( email="cand-user@test.local", username="cand-user", password="x", role="CANDIDATE" ) self.client.force_authenticate(viewer) res = self.client.post(f"/api/v1/ai-calls/{self.call.id}/cancel/") self.assertEqual(res.status_code, 403) class VoiceAgentStatusTests(BaseCallTestCase): def test_agent_statuses_contain_expected_values(self): from .agent_serializers import AGENT_STATUSES self.assertEqual(AGENT_STATUSES, ["DRAFT", "ACTIVE", "INACTIVE"]) def test_agent_options_view_returns_all_statuses(self): self.client.force_authenticate(self.admin) res = self.client.get("/api/v1/ai-calls/agents/options/") self.assertEqual(res.status_code, 200) statuses = res.json()["data"]["statuses"] self.assertEqual(statuses, ["DRAFT", "ACTIVE", "INACTIVE"])