-
Notifications
You must be signed in to change notification settings - Fork 0
feat: add private preview workspace foundation #55
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,246 @@ | ||
| """Private, build-before-return daily preview workspace foundation. | ||
|
|
||
| The caller supplies only a trusted stable parent. No fixed destination is | ||
| accepted and no publish/rename operation is performed by this API. | ||
| """ | ||
| from __future__ import annotations | ||
|
|
||
| import hashlib | ||
| import html | ||
| import json | ||
| import os | ||
| import stat | ||
| import tempfile | ||
| from collections.abc import Mapping | ||
| from pathlib import Path | ||
| from typing import Any | ||
|
|
||
| from .artifact_integrity import ArtifactIntegrityError, snapshot_json_wire | ||
| from .contracts import AdvisoryValidationError, validate_advisory_report | ||
| from .preview_bundle import BUNDLE_CONTRACT, PreviewBundleError, SOURCE_CONTRACT_VERSION, read_preview_bundle | ||
|
|
||
| _FIXED_FILES = frozenset({"report.json", "report.html", "manifest.json"}) | ||
| _NOFOLLOW = getattr(os, "O_NOFOLLOW", None) | ||
| _WORKSPACE_PREFIX = ".qar-preview-" | ||
|
|
||
|
|
||
| class PreviewWorkspaceError(ValueError): | ||
| """Stable, sanitized private-workspace error.""" | ||
|
|
||
| def __init__(self, code: str) -> None: | ||
| self.code = code | ||
| super().__init__(code) | ||
|
|
||
|
|
||
| def _error(code: str) -> PreviewWorkspaceError: | ||
| return PreviewWorkspaceError(code) | ||
|
|
||
|
|
||
| def _assert_parent_chain(parent: Path) -> None: | ||
| try: | ||
| current = parent | ||
| while True: | ||
| info = os.lstat(current) | ||
| if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): | ||
| raise _error("parent_invalid") | ||
| if current.parent == current: | ||
| return | ||
| current = current.parent | ||
| except PreviewWorkspaceError: | ||
| raise | ||
| except (OSError, TypeError, ValueError): | ||
| raise _error("parent_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") != "5": | ||
| raise _error("source_schema_unsupported") | ||
| if "contract_version" in snapshot or snapshot.get("cadence") != "daily": | ||
| raise _error("source_contract_invalid") | ||
| 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 _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 _render_html(snapshot: Mapping[str, object], report_bytes: bytes) -> bytes: | ||
| escaped = html.escape(report_bytes.decode("utf-8"), quote=True) | ||
| document = ( | ||
| "<!doctype html><html><head><meta charset=\"utf-8\"><title>QAR Daily Preview</title></head>" | ||
| "<body><nav><a href=\"report.json\">report.json</a> " | ||
| "<a href=\"manifest.json\">manifest.json</a></nav>" | ||
| f"<h1>Daily report {html.escape(str(snapshot['as_of']), quote=True)}</h1>" | ||
| f"<pre>{escaped}</pre></body></html>" | ||
| ) | ||
| 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": "5", | ||
| "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 _assert_regular(path: Path, descriptor: int | None = None) -> None: | ||
| try: | ||
| info = os.fstat(descriptor) if descriptor is not None else os.lstat(path) | ||
| except (OSError, TypeError, ValueError): | ||
| raise _error("filesystem_invalid") from None | ||
| if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: | ||
| raise _error("filesystem_invalid") | ||
| if descriptor is not None: | ||
| path_info = os.lstat(path) | ||
| if (path_info.st_dev, path_info.st_ino) != (info.st_dev, info.st_ino): | ||
| raise _error("filesystem_invalid") | ||
|
|
||
|
|
||
| def _open_flags(base: int) -> int: | ||
| if _NOFOLLOW is None: | ||
| raise _error("filesystem_unsupported") | ||
| return base | _NOFOLLOW | ||
|
|
||
|
|
||
| def _write_file(path: Path, content: bytes) -> None: | ||
| fd = -1 | ||
| try: | ||
| fd = os.open(path, _open_flags(os.O_WRONLY | os.O_CREAT | os.O_EXCL), 0o600) | ||
| _assert_regular(path, fd) | ||
| offset = 0 | ||
| while offset < len(content): | ||
| offset += os.write(fd, content[offset:]) | ||
| os.fsync(fd) | ||
| _assert_regular(path, fd) | ||
| except PreviewWorkspaceError: | ||
| raise | ||
| except (OSError, TypeError, ValueError): | ||
| raise _error("workspace_write_failed") from None | ||
| finally: | ||
| if fd >= 0: | ||
| os.close(fd) | ||
|
|
||
|
|
||
| def _read_file(path: Path) -> bytes: | ||
| fd = -1 | ||
| try: | ||
| _assert_regular(path) | ||
| fd = os.open(path, _open_flags(os.O_RDONLY)) | ||
| _assert_regular(path, fd) | ||
| chunks: list[bytes] = [] | ||
| while True: | ||
| chunk = os.read(fd, 1024 * 1024) | ||
| if not chunk: | ||
| return b"".join(chunks) | ||
| chunks.append(chunk) | ||
| except PreviewWorkspaceError: | ||
| raise | ||
| except (OSError, TypeError, ValueError): | ||
| raise _error("workspace_readback_invalid") from None | ||
| finally: | ||
| if fd >= 0: | ||
| os.close(fd) | ||
|
|
||
|
|
||
| def _read_workspace_files(workspace: Path) -> dict[str, bytes]: | ||
| try: | ||
| info = os.lstat(workspace) | ||
| if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): | ||
| raise _error("workspace_readback_invalid") | ||
| if {entry.name for entry in os.scandir(workspace)} != _FIXED_FILES: | ||
| raise _error("workspace_file_set_invalid") | ||
| return {name: _read_file(workspace / name) for name in sorted(_FIXED_FILES)} | ||
| except PreviewWorkspaceError: | ||
| raise | ||
| except (OSError, TypeError, ValueError): | ||
| raise _error("workspace_readback_invalid") from None | ||
|
|
||
|
|
||
| def _fsync_directory(path: Path) -> None: | ||
| fd = -1 | ||
| try: | ||
| fd = os.open(path, _open_flags(os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))) | ||
| os.fsync(fd) | ||
| except PreviewWorkspaceError: | ||
| raise | ||
| except (OSError, TypeError, ValueError): | ||
| raise _error("workspace_write_failed") from None | ||
| finally: | ||
| if fd >= 0: | ||
| os.close(fd) | ||
|
|
||
|
|
||
| def _cleanup_workspace(path: Path) -> None: | ||
| try: | ||
| info = os.lstat(path) | ||
| if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): | ||
| return | ||
| for entry in os.scandir(path): | ||
| child = Path(entry.path) | ||
| try: | ||
| os.unlink(child) | ||
| except IsADirectoryError: | ||
| os.rmdir(child) | ||
| os.rmdir(path) | ||
| except OSError: | ||
| pass | ||
|
|
||
|
|
||
| def build_preview_workspace(report: Mapping[str, Any], trusted_parent: str | Path) -> Path: | ||
| """Build and validate a private bundle, returning its path only on success.""" | ||
| parent = Path(trusted_parent) | ||
| _assert_parent_chain(parent) | ||
| snapshot = _validated_source(report) | ||
| report_bytes = _canonical_json(snapshot) | ||
| html_bytes = _render_html(snapshot, report_bytes) | ||
| manifest_bytes = _canonical_json(_manifest(snapshot, report_bytes, html_bytes)) | ||
| files = {"report.json": report_bytes, "report.html": html_bytes, "manifest.json": manifest_bytes} | ||
| workspace: Path | None = None | ||
| succeeded = False | ||
| try: | ||
| workspace = Path(tempfile.mkdtemp(prefix=_WORKSPACE_PREFIX, dir=parent)) | ||
| info = os.lstat(workspace) | ||
| if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode) or info.st_mode & 0o777 != 0o700: | ||
| raise _error("workspace_invalid") | ||
| for name, content in files.items(): | ||
| _write_file(workspace / name, content) | ||
| _read_workspace_files(workspace) | ||
| read_preview_bundle(workspace) | ||
| _fsync_directory(workspace) | ||
| succeeded = True | ||
| return workspace | ||
| except PreviewWorkspaceError: | ||
| raise | ||
| except PreviewBundleError: | ||
| raise _error("workspace_readback_invalid") from None | ||
| except (OSError, TypeError, ValueError): | ||
| raise _error("workspace_build_failed") from None | ||
| finally: | ||
| if workspace is not None and not succeeded: | ||
| _cleanup_workspace(workspace) | ||
|
|
||
|
|
||
| __all__ = ["PreviewWorkspaceError", "build_preview_workspace"] |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| from __future__ import annotations | ||
|
|
||
| import inspect | ||
| import os | ||
| from pathlib import Path | ||
|
|
||
| import pytest | ||
|
|
||
| import quant_advisor_research.preview_workspace as preview_workspace | ||
| from quant_advisor_research.advisory_report import build_advisory_report | ||
| from quant_advisor_research.preview_bundle import PreviewBundleError, read_preview_bundle | ||
| from quant_advisor_research.preview_workspace import PreviewWorkspaceError, build_preview_workspace | ||
|
|
||
| ROOT = Path(__file__).resolve().parents[1] | ||
|
|
||
|
|
||
| def report(): | ||
| return build_advisory_report( | ||
| as_of="2026-06-20", | ||
| cadence="daily", | ||
| political_events_path=ROOT / "examples/political_events.example.csv", | ||
| political_watchlist_path=ROOT / "examples/political_watchlist.example.csv", | ||
| ) | ||
|
|
||
|
|
||
| def test_api_accepts_only_trusted_parent_not_fixed_destination(): | ||
| assert list(inspect.signature(build_preview_workspace).parameters) == ["report", "trusted_parent"] | ||
|
|
||
|
|
||
| def test_workspace_is_private_exact_three_files_and_readback_valid(tmp_path): | ||
| workspace = build_preview_workspace(report(), tmp_path) | ||
|
|
||
| assert workspace.parent == tmp_path | ||
| assert workspace.name.startswith(".qar-preview-") | ||
| assert os.stat(workspace).st_mode & 0o777 == 0o700 | ||
| assert {path.name for path in workspace.iterdir()} == {"report.json", "report.html", "manifest.json"} | ||
| read_preview_bundle(workspace) | ||
|
|
||
|
|
||
| def test_equivalent_reports_produce_identical_bytes_in_distinct_workspaces(tmp_path): | ||
| first = build_preview_workspace(report(), tmp_path) | ||
| second = build_preview_workspace(report(), tmp_path) | ||
|
|
||
| assert first != second | ||
| assert {p.name: p.read_bytes() for p in first.iterdir()} == { | ||
| p.name: p.read_bytes() for p in second.iterdir() | ||
| } | ||
|
|
||
|
|
||
| def test_symlinked_or_non_directory_parent_fails_closed(tmp_path): | ||
| real = tmp_path / "real" | ||
| real.mkdir() | ||
| alias = tmp_path / "alias" | ||
| alias.symlink_to(real, target_is_directory=True) | ||
| with pytest.raises(PreviewWorkspaceError, match="parent_invalid"): | ||
| build_preview_workspace(report(), alias) | ||
|
|
||
| file_parent = tmp_path / "file" | ||
| file_parent.write_text("not a directory") | ||
| with pytest.raises(PreviewWorkspaceError, match="parent_invalid"): | ||
| build_preview_workspace(report(), file_parent) | ||
|
|
||
|
|
||
| def test_injected_member_symlink_never_touches_external_target(tmp_path, monkeypatch): | ||
| external = tmp_path / "external.json" | ||
| external.write_bytes(b"keep") | ||
| real_mkdtemp = preview_workspace.tempfile.mkdtemp | ||
|
|
||
| def inject_member(**kwargs): | ||
| workspace = real_mkdtemp(**kwargs) | ||
| (Path(workspace) / "report.json").symlink_to(external) | ||
| return workspace | ||
|
|
||
| monkeypatch.setattr(preview_workspace.tempfile, "mkdtemp", inject_member) | ||
| with pytest.raises(PreviewWorkspaceError, match="workspace_write_failed"): | ||
| build_preview_workspace(report(), tmp_path) | ||
| assert external.read_bytes() == b"keep" | ||
| assert not list(tmp_path.glob(".qar-preview-*")) | ||
|
|
||
|
|
||
| def test_readback_failure_cleans_only_private_workspace(tmp_path, monkeypatch): | ||
| def fail_readback(_path): | ||
| raise PreviewBundleError("forced_readback_failure") | ||
|
|
||
| monkeypatch.setattr(preview_workspace, "read_preview_bundle", fail_readback) | ||
| with pytest.raises(PreviewWorkspaceError, match="workspace_readback_invalid"): | ||
| build_preview_workspace(report(), tmp_path) | ||
| assert not list(tmp_path.glob(".qar-preview-*")) | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When the two
report()calls in this test straddle a one-second boundary,build_advisory_report()writes a freshgenerated_atvalue viautc_now_iso(), soreport.json,report.html, andmanifest.jsondiffer even though the workspace builder is deterministic. Reuse a single report snapshot or inject a fixedgenerated_atbefore asserting byte-for-byte equality to avoid nondeterministic CI failures.Useful? React with 👍 / 👎.