from io import BytesIO from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json from pathlib import Path import tempfile import threading import unittest from unittest.mock import patch from urllib.error import HTTPError, URLError from fastapi.testclient import TestClient import http_api import request_logging from scripts.tests.test_api import RecordingExecutor, _write_json_input_files, _valid_files, _write_docx class RequestLoggingTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.log = self.root / "requests.log" self.patch = patch.object(request_logging, "LOG_PATH", self.log) self.patch.start() self.addCleanup(self.patch.stop) def events(self): return [json.loads(line) for line in self.log.read_text(encoding="utf-8").splitlines()] def test_static_directory_only_preserves_root_and_filename(self): for source, expected in [ ("static/file/a.docx", "file/a.docx"), ("/static/file/static/a.docx", "/file/a.docx"), (r"C:\static\file\a.docx", r"C:\file\a.docx"), ("/data/static//file/a.docx", "/data/file/a.docx"), ("/data/mystatic/static.docx", "/data/mystatic/static.docx"), ("//server/share/static/a.docx", "//server/share/a.docx"), ]: with self.subTest(source=source): self.assertEqual(http_api._strip_static_directory(source), expected) def test_four_paths_cleaned_and_original_request_logged_for_both_modes(self): tender, procurement, reference = _write_json_input_files(self.root) cla = self.root / "clarification.docx" cla.write_bytes(b"optional") for absolute in (True, False): with self.subTest(absolute=absolute): paths = dict(zip(("TENDER_FILE", "PROCUREMENT_FILE", "REFERENCE_BID", "CLA_FILE"), (tender, procurement, reference, cla))) payload = {key: str(path.parent / "static" / path.name) if absolute else "static/" + path.name for key, path in paths.items()} payload.update(CALLBACK_URL="http://localhost/callback", txbId="TEST", TPC_IS_ABSOLUTE=absolute, REFERENCE_IS_ABSOLUTE=absolute) executor = RecordingExecutor() with patch.object(http_api, "API_WORK_ROOT", self.root / "jobs"), \ patch.object(http_api, "TPC_DIR", str(self.root)), \ patch.object(http_api, "REFERENCE_DIR", str(self.root)), \ patch.object(http_api, "_JOB_EXECUTOR", executor): response = TestClient(http_api.app).post("/api/v1/final-review", json=payload) self.assertEqual(response.status_code, 202, response.text) job = executor.submitted[0][1][0] self.assertEqual(set(job["files"]), set(paths)) records = [e for e in self.events() if e["request_id"] == response.json()["request_id"]] self.assertEqual(records[0]["body"], payload) effective = next(e for e in records if e["event"] == "processing_request") self.assertEqual(effective["source_paths"], {key: str(path.resolve()) for key, path in paths.items()}) self.assertEqual(effective["body"]["TENDER_FILE"], str(job["tender_path"].resolve())) self.assertTrue(effective["body"]["TPC_IS_ABSOLUTE"]) self.assertTrue(effective["body"]["CLEAN_INTERMEDIATE"]) self.assertEqual(effective["body"]["CALLBACK_URL"], "http://localhost/callback") self.assertEqual(records[-1]["body"], response.json()) self.assertEqual(records[-1]["status"], 202) def test_rejected_and_invalid_json_requests_are_logged(self): client = TestClient(http_api.app) for body, status in [("{broken", 400), ('{"txbId":"TEST"}', 422)]: response = client.post("/api/v1/final-review", content=body, headers={"Content-Type": "application/json"}) self.assertEqual(response.status_code, status) records = self.events()[-2:] self.assertEqual(records[0]["request_id"], records[1]["request_id"]) self.assertEqual(records[1]["body"], response.json()) self.assertFalse(any(e["event"] == "processing_request" for e in self.events())) def test_multipart_records_fields_and_file_metadata_without_binary(self): with patch.object(http_api, "API_WORK_ROOT", self.root / "jobs"), \ patch.object(http_api, "_JOB_EXECUTOR", RecordingExecutor()): response = TestClient(http_api.app).post("/api/v1/final-review", data={"CALLBACK_URL": "http://localhost/callback", "txbId": "TEST"}, files=_valid_files()) self.assertEqual(response.status_code, 202) record = next(e for e in self.events() if e["event"] == "request") fields = {entry["name"]: entry["value"] for entry in record["body"]} self.assertEqual(fields["txbId"], "TEST") self.assertGreater(fields["TENDER_FILE"]["size"], 0) self.assertNotIn("%PDF", self.log.read_text(encoding="utf-8")) def test_callback_http_error_retry_and_business_response_logged(self): class Response(BytesIO): status = 200 responses = [HTTPError("http://localhost", 503, "unavailable", {}, BytesIO(b"busy")), Response(b'{"code":400,"message":"unknown id"}')] with patch.object(http_api, "urlopen", side_effect=responses), \ patch.object(http_api.time, "sleep"), patch.object(http_api, "CALLBACK_MAX_RETRIES", 2): http_api._send_callback("http://localhost/callback", {"request_id": "TEST"}) records = self.events() self.assertEqual([e["event"] for e in records], ["callback_request", "callback_response"] * 2) self.assertEqual(records[1]["status"], 503) self.assertEqual(records[1]["response_body"], "busy") self.assertEqual(records[3]["response_body"]["code"], 400) self.assertEqual(records[3]["attempt"], 2) def test_callback_network_failure_logged(self): with patch.object(http_api, "urlopen", side_effect=URLError("offline")), \ patch.object(http_api, "CALLBACK_MAX_RETRIES", 1): with self.assertRaises(RuntimeError): http_api._send_callback("http://localhost/callback", {"request_id": "TEST"}) self.assertIn("offline", self.events()[-1]["error"]) def test_log_write_failure_does_not_fail_request(self): with patch.object(request_logging, "LOG_PATH", self.root), \ self.assertLogs("request_logging", level="ERROR"): self.assertEqual(TestClient(http_api.app).get("/health").status_code, 200) def test_local_http_callback_and_job_cleanup_keep_correlated_log(self): received = [] class Handler(BaseHTTPRequestHandler): def do_POST(self): received.append(json.loads(self.rfile.read(int(self.headers["Content-Length"])))) self.send_response(200) self.end_headers() self.wfile.write(b'{"code":200}') def log_message(self, *_args): pass server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: executor = RecordingExecutor() with patch.object(http_api, "API_WORK_ROOT", self.root / "jobs"), \ patch.object(http_api, "_JOB_EXECUTOR", executor), \ patch.object(http_api, "_run_pipeline_worker", side_effect=lambda _a, _b, _c, output, _log: _write_docx(output)): client = TestClient(http_api.app) response = client.post("/api/v1/final-review", files=_valid_files(), data={"CALLBACK_URL": f"http://127.0.0.1:{server.server_port}/callback", "txbId": "TEST"}) self.assertEqual(response.status_code, 202) function, args = executor.submitted[0] function(*args) job_id = response.json()["request_id"] self.assertTrue(client.get(f"/api/v1/jobs/{job_id}").json()["callback_delivered"]) self.assertEqual(client.get(f"/api/v1/jobs/{job_id}/file").status_code, 200) self.assertEqual(received[0]["request_id"], job_id) self.assertFalse(args[0]["job_dir"].exists()) callback = next(e for e in self.events() if e["event"] == "callback_response") self.assertEqual(callback["request_id"], job_id) self.assertEqual(callback["response_body"], {"code": 200}) self.assertEqual(self.events()[-1]["body"], "[non-JSON response omitted]") finally: server.shutdown() server.server_close() thread.join() def test_early_server_failure_still_logs_original_json(self): with patch.object(http_api, "TEMPLATE_PATH", self.root / "missing.docx"): response = TestClient(http_api.app).post("/api/v1/final-review", json={"txbId": "TEST"}) self.assertEqual(response.status_code, 500) self.assertEqual(self.events()[0]["body"], {"txbId": "TEST"}) self.assertEqual(self.events()[-1]["status"], 500)