"""Professional-reference export API and public-output contracts.""" from __future__ import annotations import json from io import BytesIO import pytest from scripts.jyotish_api_server import ( DEFAULT_ALLOWED_HOSTS, DEFAULT_ALLOWED_ORIGINS, BadRequest, JyotishAPIHandler, ) from scripts.jyotish_engine import ( _professional_report_reference_boundary_notice, render_pl9_markdown, sanitize_professional_report_reference, sanitize_professional_report_reference_markdown, ) from scripts.professional_report_reference import ( ProfessionalReportReferenceInputError, build_professional_report_reference, ) BIRTH = { "year": 1990, "month": 1, "day": 1, "hour": 12, "minute": 0, "lat": 39.9042, "lon": 116.4074, "tz": 8, "ayanamsa": "raman", } class _ReferenceHandler: def __init__(self) -> None: self.full_reading_calls = 0 def _high_rigor_birth_payload(self, body): # noqa: ANN001 return {**BIRTH, "hour": 12.0, "minute": 0.0, "second": 0.0, "node_mode": "mean", "today": None, "transit_date": None} def _compute_full_reading_for_thematic(self, body): # noqa: ANN001 self.full_reading_calls += 1 return {"modules": {"chart": {"status": "executed"}}, "birth": body} class _ReferenceEngine: def __init__(self, *, invalid_pack: bool = False) -> None: self.invalid_pack = invalid_pack self.calls = [] def build_professional_report_reference_packet(self, full_reading, args, packs): # noqa: ANN001 assert isinstance(args.hour, int) assert isinstance(args.minute, int) assert isinstance(args.second, int) self.calls.append((full_reading, args, packs)) if self.invalid_pack: raise ValueError("Unknown professional report pack: secret") return { "schema": "pl9_style_professional_export_v1", "reader_engine_boundary_notice": _professional_report_reference_boundary_notice(), "selected_report_pack_ids": packs or ["full"], } def render_pl9_markdown(self, packet): # noqa: ANN001 return render_pl9_markdown(packet) def test_json_and_markdown_reuse_one_full_reading_and_normalize_packs() -> None: handler = _ReferenceHandler() engine = _ReferenceEngine() json_result = build_professional_report_reference( handler, {**BIRTH, "format": "json", "packs": ["base", "base", "timing"]}, engine=engine, ) assert json_result["format"] == "json" assert json_result["report"]["selected_report_pack_ids"] == ["base", "timing"] assert handler.full_reading_calls == 1 markdown_result = build_professional_report_reference( handler, {**BIRTH, "format": "markdown", "packs": "full"}, engine=engine, ) assert handler.full_reading_calls == 2 assert markdown_result["format"] == "markdown" assert "## 多引擎口径说明" in markdown_result["markdown"] assert engine.calls[-1][2] == ["full"] def test_public_markdown_sanitizer_removes_renderer_authored_internal_references() -> None: unsafe = ( "| audit | `raw_full_reading.modules.private_engine` |\n" "source=scripts/private_engine.py::compute path=/Users/example/report.json " "call=jyotish_engine.cmd_full_reading\n" "oracle=references/oracle/private_status.json module=yoga_engine " "path=relative/private_report.json\n" ) sanitized = sanitize_professional_report_reference_markdown(unsafe) assert "internal_reference_omitted" in sanitized assert "raw_full_reading" not in sanitized assert "scripts/" not in sanitized assert ".py::" not in sanitized assert "/Users/" not in sanitized assert "jyotish_engine." not in sanitized assert "references/oracle/" not in sanitized assert "yoga_engine" not in sanitized assert "path=" not in sanitized def test_public_sanitizer_removes_raw_paths_and_internal_python_references() -> None: unsafe = { "raw_full_reading": {"modules": {"private": True}}, "fixture_path": "/Users/example/references/oracle.json", "references/oracle/private_key.json": {"status": "blocked"}, "yoga_engine": {"status": "executed"}, "path=relative/private_key.json": "blocked", "nested": { "source_path": "/private/tmp/capture.json", "reason": "scripts/yoga_engine.py::YogaContext failed", "callable": "jyotish_engine.cmd_full_reading", "oracle_reference": "references/oracle/private_status.json", "module": "yoga_engine", "trace": "path=relative/private_report.json", "plain": "parameter_sensitive", }, } sanitized = sanitize_professional_report_reference(unsafe) rendered = json.dumps(sanitized, ensure_ascii=False) assert "raw_full_reading" not in rendered assert "fixture_path" not in rendered assert "/Users/" not in rendered assert "scripts/" not in rendered assert ".py::" not in rendered assert "jyotish_engine." not in rendered assert "references/oracle/" not in rendered assert "yoga_engine" not in rendered assert "path=" not in rendered assert sanitized["nested"]["plain"] == "parameter_sensitive" assert sanitized["nested"]["reason"] == "internal_reference_omitted" assert sanitized["nested"]["oracle_reference"] == "internal_reference_omitted" assert sanitized["nested"]["module"] == "internal_reference_omitted" assert sanitized["nested"]["trace"] == "internal_reference_omitted" @pytest.mark.parametrize( "body, message", [ ({**BIRTH, "format": "pdf"}, "format must be json or markdown"), ({**BIRTH, "packs": {"full": True}}, "packs must be a string or array"), ({**BIRTH, "packs": ["full", 1]}, "packs must contain only strings"), ], ) def test_invalid_format_or_packs_fail_before_full_reading(body, message) -> None: # noqa: ANN001 handler = _ReferenceHandler() with pytest.raises(ProfessionalReportReferenceInputError, match=message): build_professional_report_reference(handler, body, engine=_ReferenceEngine()) assert handler.full_reading_calls == 0 def test_unknown_pack_becomes_public_input_error() -> None: handler = _ReferenceHandler() with pytest.raises(ProfessionalReportReferenceInputError, match="Unknown professional report pack"): build_professional_report_reference( handler, {**BIRTH, "packs": ["secret"]}, engine=_ReferenceEngine(invalid_pack=True), ) assert handler.full_reading_calls == 1 def test_missing_birth_fields_use_consultation_strict_validation() -> None: handler = JyotishAPIHandler.__new__(JyotishAPIHandler) with pytest.raises(BadRequest, match="missing birth fields: minute"): build_professional_report_reference(handler, {**BIRTH, "minute": None}, engine=_ReferenceEngine()) class _FakeHeaders(dict): def get(self, key, default=None): # noqa: ANN001 return super().get(key, default) class _FakeServer: allowed_origins = DEFAULT_ALLOWED_ORIGINS allowed_hosts = DEFAULT_ALLOWED_HOSTS class _PostCaptureHandler(JyotishAPIHandler): def __init__(self, payload: dict) -> None: raw = json.dumps(payload).encode("utf-8") self.headers = _FakeHeaders({"Content-Length": str(len(raw)), "Content-Type": "application/json"}) self.server = _FakeServer() self.path = "/api/professional_report_reference" self.rfile = BytesIO(raw) self.wfile = BytesIO() self.status_code = None self.response_headers = [] self.client_address = ("test-professional-reference", 0) def send_response(self, code, message=None): # noqa: ANN001 self.status_code = code def send_header(self, key, value): # noqa: ANN001 self.response_headers.append((key, value)) def end_headers(self): return None def payload(self) -> dict: return json.loads(self.wfile.getvalue().decode("utf-8")) def test_http_endpoint_maps_invalid_format_to_bad_request() -> None: handler = _PostCaptureHandler({**BIRTH, "format": "pdf"}) handler.do_POST() assert handler.status_code == 400 assert handler.payload()["error_code"] == "ERR_BAD_REQUEST"