from __future__ import annotations from collections.abc import Callable from typing import Any from fastapi.testclient import TestClient from app.core.errors import ErrorCode from tests.conftest import FakeScorer, make_encrypted_pdf, make_pdf, resume_pdf JD = "Backend engineer. Required: Python, FastAPI, Docker. Preferred: AWS, Kubernetes." def upload(name: str, data: bytes, content_type: str = "application/pdf") -> tuple[str, Any]: return ("resumes", (name, data, content_type)) def post(client: TestClient, files: list[Any], job_description: str = JD) -> Any: return client.post( "/api/v1/score", data={"job_description": job_description}, files=files, ) class TestHappyPath: def test_returns_a_score_sorted_leaderboard(self, client: TestClient) -> None: response = post( client, [ upload("mid.pdf", resume_pdf("Mid", 55)), upload("top.pdf", resume_pdf("Top", 92)), upload("low.pdf", resume_pdf("Low", 20)), ], ) assert response.status_code == 200 body = response.json() assert body["total"] == 3 assert body["succeeded"] == 3 assert body["failed"] == 0 assert [item["filename"] for item in body["results"]] == [ "top.pdf", "mid.pdf", "low.pdf", ] assert [item["match_score"] for item in body["results"]] == [92, 55, 20] def test_completed_results_match_the_documented_schema(self, client: TestClient) -> None: body = post(client, [upload("a.pdf", resume_pdf("Ada", 70))]).json() assert set(body) == {"request_id", "total", "succeeded", "failed", "results"} result = body["results"][0] assert set(result) == { "filename", "status", "candidate_name", "job_title", "current_company", "years_experience", "match_score", "matched_keywords", "missing_keywords", "summary_critique", } assert result["status"] == "completed" def test_response_carries_a_request_id_header(self, client: TestClient) -> None: response = post(client, [upload("a.pdf", resume_pdf("Ada", 70))]) assert response.headers["X-Request-ID"] assert response.json()["request_id"] == response.headers["X-Request-ID"] def test_inbound_request_id_is_honoured(self, client: TestClient) -> None: response = client.post( "/api/v1/score", data={"job_description": JD}, files=[upload("a.pdf", resume_pdf("Ada", 70))], headers={"X-Request-ID": "trace-123"}, ) assert response.headers["X-Request-ID"] == "trace-123" def test_health_endpoint(self, client: TestClient) -> None: assert client.get("/api/v1/health").json() == {"status": "ok"} class TestPartialFailure: def test_one_unreadable_resume_does_not_fail_the_batch(self, client: TestClient) -> None: response = post( client, [ upload("good.pdf", resume_pdf("Good", 88)), upload("scanned.pdf", make_pdf([[]])), upload("locked.pdf", make_encrypted_pdf()), upload("broken.pdf", b"%PDF-1.4\nnot really a pdf"), ], ) assert response.status_code == 200 body = response.json() assert body["total"] == 4 assert body["succeeded"] == 1 assert body["failed"] == 3 results = body["results"] assert results[0]["filename"] == "good.pdf" # Failures come after completions, in upload order. assert [item["filename"] for item in results[1:]] == [ "scanned.pdf", "locked.pdf", "broken.pdf", ] assert [item["error_code"] for item in results[1:]] == [ ErrorCode.PDF_TEXT_UNAVAILABLE, ErrorCode.PDF_ENCRYPTED, ErrorCode.INVALID_PDF, ] def test_failed_results_match_the_documented_schema(self, client: TestClient) -> None: body = post(client, [upload("scanned.pdf", make_pdf([[]]))]).json() result = body["results"][0] assert set(result) == {"filename", "status", "error_code", "error_message"} assert result["status"] == "failed" def test_a_provider_failure_is_isolated_to_its_candidate( self, make_client: Callable[..., TestClient] ) -> None: def handler(text: str) -> Any: if "Boom" in text: raise RuntimeError("upstream detail that must not leak") from tests.conftest import default_score return default_score(text) client = make_client(scorer=FakeScorer(handler)) body = post( client, [ upload("ok.pdf", resume_pdf("Fine", 65)), upload("bad.pdf", resume_pdf("Boom", 10)), ], ).json() assert body["succeeded"] == 1 failure = body["results"][1] assert failure["error_code"] == ErrorCode.INTERNAL_ERROR assert "upstream detail" not in failure["error_message"] def test_a_batch_of_only_failures_still_returns_200(self, client: TestClient) -> None: response = post(client, [upload("scanned.pdf", make_pdf([[]]))]) assert response.status_code == 200 assert response.json()["succeeded"] == 0 class TestRequestValidation: def test_blank_job_description_is_400(self, client: TestClient) -> None: response = post(client, [upload("a.pdf", resume_pdf("Ada"))], job_description=" ") assert response.status_code == 400 assert response.json()["error_code"] == ErrorCode.INVALID_REQUEST def test_missing_job_description_is_400(self, client: TestClient) -> None: response = client.post("/api/v1/score", files=[upload("a.pdf", resume_pdf("Ada"))]) assert response.status_code == 400 assert response.json()["error_code"] == ErrorCode.INVALID_REQUEST def test_missing_resumes_is_400(self, client: TestClient) -> None: response = client.post("/api/v1/score", data={"job_description": JD}) assert response.status_code == 400 def test_over_length_job_description_is_422( self, make_client: Callable[..., TestClient] ) -> None: client = make_client(max_jd_chars=100) response = post(client, [upload("a.pdf", resume_pdf("Ada"))], job_description="x" * 200) assert response.status_code == 422 assert response.json()["error_code"] == ErrorCode.UNPROCESSABLE_FIELD def test_too_many_files_is_413(self, make_client: Callable[..., TestClient]) -> None: client = make_client(max_resumes_per_request=2) response = post( client, [upload(f"c{i}.pdf", resume_pdf(f"C{i}", 50)) for i in range(3)], ) assert response.status_code == 413 assert response.json()["error_code"] == ErrorCode.PAYLOAD_TOO_LARGE def test_oversized_file_is_413(self, make_client: Callable[..., TestClient]) -> None: client = make_client(max_pdf_size_mb=1) big = make_pdf( [[f"Padding line number {i} for the oversized document." for i in range(30_000)]] ) assert len(big) > 1024 * 1024 response = post(client, [upload("big.pdf", big)]) assert response.status_code == 413 def test_non_pdf_extension_is_415(self, client: TestClient) -> None: response = post(client, [upload("resume.docx", b"whatever", "application/msword")]) assert response.status_code == 415 assert response.json()["error_code"] == ErrorCode.UNSUPPORTED_FILE_TYPE def test_disallowed_content_type_is_415(self, client: TestClient) -> None: response = post(client, [upload("resume.pdf", resume_pdf("Ada"), "text/html")]) assert response.status_code == 415 def test_octet_stream_is_accepted_when_the_extension_is_pdf(self, client: TestClient) -> None: """Clients disagree on the PDF MIME type; the %PDF- signature is the real gate.""" response = post( client, [upload("a.pdf", resume_pdf("Ada", 60), "application/octet-stream")], ) assert response.status_code == 200 def test_error_envelope_shape(self, client: TestClient) -> None: body = post(client, [upload("a.pdf", resume_pdf("Ada"))], job_description="").json() assert set(body) == {"request_id", "error_code", "error_message"} class TestFilenameHandling: def test_traversal_in_the_upload_name_is_stripped(self, client: TestClient) -> None: body = post( client, [upload(r"..\..\windows\system32\evil.pdf", resume_pdf("Ada", 60))], ).json() assert body["results"][0]["filename"] == "evil.pdf" class TestUI: def test_root_serves_the_test_ui(self, client: TestClient) -> None: response = client.get("/") assert response.status_code == 200 assert response.headers["content-type"].startswith("text/html") assert "Bulk ATS Scoring" in response.text def test_ui_is_not_in_the_openapi_schema(self, client: TestClient) -> None: assert "/" not in client.get("/openapi.json").json()["paths"] class TestConcurrency: def test_batch_respects_the_configured_bound( self, make_client: Callable[..., TestClient] ) -> None: scorer = FakeScorer(delay=0.01) client = make_client(scorer=scorer, scoring_concurrency=2, max_resumes_per_request=10) post(client, [upload(f"c{i}.pdf", resume_pdf(f"C{i}", 50)) for i in range(6)]) assert scorer.max_concurrent <= 2