diff --git a/src/quant_advisor_research/preview_bundle.py b/src/quant_advisor_research/preview_bundle.py new file mode 100644 index 0000000..e031b19 --- /dev/null +++ b/src/quant_advisor_research/preview_bundle.py @@ -0,0 +1,202 @@ +"""Concrete daily report-to-preview bundle with no legacy or runtime integration.""" +from __future__ import annotations + +import hashlib +import html +import json +import os +import tempfile +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType +from typing import Any + +from .artifact_integrity import ArtifactIntegrityError, snapshot_json_wire +from .contracts import AdvisoryValidationError, validate_advisory_report + +BUNDLE_CONTRACT = "qar.preview_bundle.v1" +SOURCE_SCHEMA_VERSION = "5" +SOURCE_CONTRACT_VERSION = "model_recommendations.v5" +_FIXED_FILES = frozenset({"report.json", "report.html", "manifest.json"}) +_MANIFEST_KEYS = frozenset({"bundle_contract", "source", "artifacts"}) +_SOURCE_KEYS = frozenset({"schema_version", "contract_version", "cadence", "as_of", "generated_at"}) +_ARTIFACT_KEYS = frozenset({"name", "role", "sha256"}) + + +class PreviewBundleError(ValueError): + """Stable, sanitized preview bundle error.""" + + def __init__(self, code: str): + self.code = code + super().__init__(code) + + +@dataclass(frozen=True, slots=True) +class PreviewBundleEvidence: + report: Mapping[str, object] + manifest: Mapping[str, object] + + def __post_init__(self) -> None: + if not isinstance(self.report, Mapping) or not isinstance(self.manifest, Mapping): + raise PreviewBundleError("readback_invalid") + + @property + def bundle_contract(self) -> str: + return BUNDLE_CONTRACT + + +def _error(code: str) -> PreviewBundleError: + return PreviewBundleError(code) + + +def _canonical_json(value: object) -> bytes: + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + except (TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("serialization_invalid") from None + + +def _validated_source(report: Mapping[str, Any]) -> dict[str, object]: + try: + snapshot = snapshot_json_wire(report) + validate_advisory_report(snapshot) + except (ArtifactIntegrityError, AdvisoryValidationError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("source_invalid") from None + if snapshot.get("schema_version") != SOURCE_SCHEMA_VERSION: + raise _error("source_schema_unsupported") + if "contract_version" in snapshot: + raise _error("source_contract_field_forbidden") + if snapshot.get("cadence") != "daily": + raise _error("daily_only") + if type(snapshot.get("as_of")) is not str or type(snapshot.get("generated_at")) is not str: + raise _error("source_time_invalid") + return snapshot + + +def _render_html(snapshot: Mapping[str, object], report_bytes: bytes) -> bytes: + escaped = html.escape(report_bytes.decode("utf-8"), quote=True) + document = ( + "QAR Daily Preview" + "" + f"

Daily report {html.escape(str(snapshot['as_of']), quote=True)}

" + f"
{escaped}
" + ) + return document.encode("utf-8") + + +def _manifest(snapshot: Mapping[str, object], report_bytes: bytes, html_bytes: bytes) -> dict[str, object]: + def artifact(name: str, role: str, content: bytes) -> dict[str, str]: + return {"name": name, "role": role, "sha256": hashlib.sha256(content).hexdigest()} + + return { + "bundle_contract": BUNDLE_CONTRACT, + "source": { + "schema_version": SOURCE_SCHEMA_VERSION, + "contract_version": SOURCE_CONTRACT_VERSION, + "cadence": "daily", + "as_of": snapshot["as_of"], + "generated_at": snapshot["generated_at"], + }, + "artifacts": { + "report.json": artifact("report.json", "source_report", report_bytes), + "report.html": artifact("report.html", "escaped_preview", html_bytes), + }, + } + + +def _output_parent(path: str | Path) -> Path: + output = Path(path) + try: + if output.exists(): + raise _error("output_exists") + if not output.parent.is_dir(): + raise _error("output_parent_invalid") + except PreviewBundleError: + raise + except (OSError, TypeError, ValueError): + raise _error("output_parent_invalid") from None + return output + + +def build_preview_bundle(report: Mapping[str, Any], output_dir: str | Path) -> PreviewBundleEvidence: + """Validate once, build three deterministic bytes, then write an empty directory.""" + snapshot = _validated_source(report) + output = _output_parent(output_dir) + report_bytes = _canonical_json(snapshot) + html_bytes = _render_html(snapshot, report_bytes) + manifest = _manifest(snapshot, report_bytes, html_bytes) + manifest_bytes = _canonical_json(manifest) + files = {"report.json": report_bytes, "report.html": html_bytes, "manifest.json": manifest_bytes} + staging_dir: str | None = None + try: + staging_dir = tempfile.mkdtemp(prefix=f".{output.name}.staging-", dir=output.parent) + for name, content in files.items(): + Path(staging_dir, name).write_bytes(content) + read_preview_bundle(staging_dir) + os.rename(staging_dir, output) + staging_dir = None + except FileExistsError: + raise _error("output_exists") from None + except (OSError, TypeError, ValueError): + raise _error("output_write_failed") from None + finally: + if staging_dir is not None: + for child in Path(staging_dir).iterdir(): + child.unlink(missing_ok=True) + Path(staging_dir).rmdir() + return PreviewBundleEvidence(MappingProxyType(snapshot), MappingProxyType(manifest)) + + +def _parse_json_bytes(content: bytes) -> object: + try: + text = content.decode("utf-8", errors="strict") + return json.loads(text) + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _error("readback_invalid") from None + + +def read_preview_bundle(output_dir: str | Path) -> PreviewBundleEvidence: + output = Path(output_dir) + try: + if not output.is_dir() or {item.name for item in output.iterdir()} != _FIXED_FILES: + raise _error("readback_file_set_invalid") + report_bytes = (output / "report.json").read_bytes() + html_bytes = (output / "report.html").read_bytes() + manifest_bytes = (output / "manifest.json").read_bytes() + except PreviewBundleError: + raise + except (OSError, TypeError, ValueError): + raise _error("readback_invalid") from None + + report = _parse_json_bytes(report_bytes) + if not isinstance(report, Mapping): + raise _error("readback_invalid") + snapshot = _validated_source(report) + if _canonical_json(snapshot) != report_bytes: + raise _error("report_bytes_noncanonical") + manifest = _parse_json_bytes(manifest_bytes) + if not isinstance(manifest, Mapping): + raise _error("manifest_invalid") + if set(manifest) != _MANIFEST_KEYS or set(manifest.get("source", {})) != _SOURCE_KEYS: + raise _error("manifest_shape_invalid") + artifacts = manifest.get("artifacts") + if not isinstance(artifacts, Mapping) or set(artifacts) != {"report.json", "report.html"}: + raise _error("manifest_shape_invalid") + for item in artifacts.values(): + if not isinstance(item, Mapping) or set(item) != _ARTIFACT_KEYS or type(item.get("name")) is not str or type(item.get("role")) is not str: + raise _error("manifest_shape_invalid") + expected_manifest = _manifest(snapshot, report_bytes, html_bytes) + if manifest != expected_manifest: + raise _error("manifest_mismatch") + expected_html = _render_html(snapshot, report_bytes) + if html_bytes != expected_html or html_bytes.count(b'href="report.json"') != 1 or html_bytes.count(b'href="manifest.json"') != 1: + raise _error("html_links_invalid") + return PreviewBundleEvidence(MappingProxyType(snapshot), MappingProxyType(dict(manifest))) + + +__all__ = [ + "BUNDLE_CONTRACT", "SOURCE_CONTRACT_VERSION", "PreviewBundleError", "PreviewBundleEvidence", + "build_preview_bundle", "read_preview_bundle", +] diff --git a/tests/test_preview_bundle.py b/tests/test_preview_bundle.py new file mode 100644 index 0000000..044be71 --- /dev/null +++ b/tests/test_preview_bundle.py @@ -0,0 +1,177 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +import quant_advisor_research.preview_bundle as preview_bundle +from quant_advisor_research.advisory_report import build_advisory_report +from quant_advisor_research.preview_bundle import ( + BUNDLE_CONTRACT, + SOURCE_CONTRACT_VERSION, + PreviewBundleError, + build_preview_bundle, + read_preview_bundle, +) + +ROOT = Path(__file__).resolve().parents[1] + + +def report(*, cadence="daily", as_of="2026-06-20"): + return build_advisory_report( + as_of=as_of, + cadence=cadence, + political_events_path=ROOT / "examples/political_events.example.csv", + political_watchlist_path=ROOT / "examples/political_watchlist.example.csv", + ) + + +def test_daily_bundle_builds_and_readback_validates(tmp_path): + output = tmp_path / "preview" + result = build_preview_bundle(report(), output) + + assert result.bundle_contract == BUNDLE_CONTRACT + assert sorted(path.name for path in output.iterdir()) == ["manifest.json", "report.html", "report.json"] + evidence = read_preview_bundle(output) + assert evidence.report["cadence"] == "daily" + manifest = json.loads((output / "manifest.json").read_text()) + assert manifest == { + "bundle_contract": BUNDLE_CONTRACT, + "source": { + "schema_version": "5", + "contract_version": SOURCE_CONTRACT_VERSION, + "cadence": "daily", + "as_of": "2026-06-20", + "generated_at": report()["generated_at"], + }, + "artifacts": { + "report.json": { + "name": "report.json", + "role": "source_report", + "sha256": hashlib.sha256((output / "report.json").read_bytes()).hexdigest(), + }, + "report.html": { + "name": "report.html", + "role": "escaped_preview", + "sha256": hashlib.sha256((output / "report.html").read_bytes()).hexdigest(), + }, + }, + } + + +@pytest.mark.parametrize("cadence", ["weekly", "monthly"]) +def test_non_daily_source_is_rejected_before_output(cadence, tmp_path): + output = tmp_path / "preview" + with pytest.raises(PreviewBundleError, match="daily_only"): + build_preview_bundle(report(cadence=cadence), output) + assert not output.exists() + + +@pytest.mark.parametrize("mutation", [ + lambda value: value.update(schema_version="6"), + lambda value: value.update(contract_version="wrong"), + lambda value: value.update(generated_at="not-a-datetime"), +]) +def test_source_contract_mutations_fail_closed_without_partial_output(mutation, tmp_path): + value = report() + mutation(value) + output = tmp_path / "preview" + with pytest.raises(PreviewBundleError): + build_preview_bundle(value, output) + assert not output.exists() + + +def test_build_is_deterministic_for_equivalent_mapping_order(tmp_path): + left = tmp_path / "left" + right = tmp_path / "right" + value = report() + reordered = {key: value[key] for key in reversed(list(value))} + build_preview_bundle(value, left) + build_preview_bundle(reordered, right) + assert {path.name: path.read_bytes() for path in left.iterdir()} == { + path.name: path.read_bytes() for path in right.iterdir() + } + + +def test_html_escapes_snapshot_and_has_only_fixed_relative_links(tmp_path): + output = tmp_path / "preview" + value = report() + value["source_artifacts"]["political_events"] = "" + build_preview_bundle(value, output) + html = (output / "report.html").read_text() + assert "<script>alert(' x')</script>" not in html + assert "<script>alert('x')</script>" in html + assert 'href="report.json"' in html + assert 'href="manifest.json"' in html + assert html.count('href="') == 2 + + +@pytest.mark.parametrize("tamper", [ + lambda path: path.joinpath("report.json").write_text(path.joinpath("report.json").read_text().replace('"cadence":"daily"', '"cadence":"weekly"')), + lambda path: path.joinpath("manifest.json").write_text(path.joinpath("manifest.json").read_text().replace('report.json', 'other.json')), + lambda path: path.joinpath("report.html").write_text(path.joinpath("report.html").read_text().replace('href="report.json"', 'href="other.json"')), + lambda path: (path / "unexpected.txt").write_text("x"), +]) +def test_readback_tamper_and_extra_file_fail_closed(tamper, tmp_path): + output = tmp_path / "preview" + build_preview_bundle(report(), output) + tamper(output) + with pytest.raises(PreviewBundleError): + read_preview_bundle(output) + + +def test_non_empty_output_fails_without_touching_sentinel(tmp_path): + output = tmp_path / "preview" + output.mkdir() + sentinel = output / "sentinel" + sentinel.write_text("keep") + with pytest.raises(PreviewBundleError, match="output_exists"): + build_preview_bundle(report(), output) + assert sentinel.read_text() == "keep" + + +def test_destination_is_not_visible_during_staging_readback(tmp_path, monkeypatch): + output = tmp_path / "preview" + observed = [] + original = preview_bundle.read_preview_bundle + + def inspect(path): + observed.append(Path(path)) + assert not output.exists() + return original(path) + + monkeypatch.setattr(preview_bundle, "read_preview_bundle", inspect) + build_preview_bundle(report(), output) + assert len(observed) == 1 + assert output.is_dir() + + +def test_concurrent_destination_winner_is_not_overwritten_or_cleaned(tmp_path, monkeypatch): + output = tmp_path / "preview" + real_rename = preview_bundle.os.rename + + def concurrent_winner(source, destination): + Path(destination).mkdir() + raise FileExistsError(destination) + + monkeypatch.setattr(preview_bundle.os, "rename", concurrent_winner) + with pytest.raises(PreviewBundleError, match="output_exists"): + build_preview_bundle(report(), output) + assert output.is_dir() + assert not list(tmp_path.glob(".preview.staging-*")) + monkeypatch.setattr(preview_bundle.os, "rename", real_rename) + + +def test_staging_directory_is_cleaned_when_install_fails(tmp_path, monkeypatch): + output = tmp_path / "preview" + + def fail_readback(_path): + raise PreviewBundleError("forced_readback_failure") + + monkeypatch.setattr(preview_bundle, "read_preview_bundle", fail_readback) + with pytest.raises(PreviewBundleError, match="output_write_failed"): + build_preview_bundle(report(), output) + assert not output.exists() + assert not list(tmp_path.glob(".preview.staging-*"))