|
2 | 2 |
|
3 | 3 | import datetime as dt |
4 | 4 | import json |
| 5 | +import traceback |
5 | 6 | from collections import OrderedDict |
6 | 7 | from collections.abc import Mapping |
7 | 8 | from pathlib import Path |
8 | 9 | from types import MappingProxyType |
9 | 10 |
|
10 | 11 | import pytest |
11 | 12 |
|
| 13 | +import quant_advisor_research.artifact_integrity as artifact_integrity |
12 | 14 | from quant_advisor_research.advisory_report import build_advisory_report |
13 | 15 | from quant_advisor_research.artifact_integrity import ( |
14 | 16 | ARTIFACT_INTEGRITY_VERSION, |
|
18 | 20 | make_artifact_integrity_evidence, |
19 | 21 | ) |
20 | 22 | from quant_advisor_research.publisher import report_content_fingerprint |
21 | | -from quant_advisor_research.time_contract import canonical_reference_time, normalize_aware_datetime |
| 23 | +from quant_advisor_research.period_contract import PeriodContractError |
| 24 | +from quant_advisor_research.time_contract import TimeContractError, canonical_reference_time, normalize_aware_datetime |
22 | 25 |
|
23 | 26 |
|
24 | 27 | ROOT = Path(__file__).resolve().parents[1] |
@@ -203,3 +206,36 @@ def items(self): |
203 | 206 | evidence = make_artifact_integrity_evidence(MutatingMapping(report)) |
204 | 207 | assert report["generated_at"] != original |
205 | 208 | assert evidence.digest == artifact_integrity_digest(dict(report, generated_at=original)) |
| 209 | + |
| 210 | + |
| 211 | +@pytest.mark.parametrize("exception_type", [artifact_integrity.AdvisoryValidationError, TimeContractError, PeriodContractError]) |
| 212 | +@pytest.mark.parametrize("operation", ["canonicalize", "digest", "evidence"]) |
| 213 | +def test_known_validator_exceptions_are_sanitized(monkeypatch, exception_type, operation: str) -> None: |
| 214 | + marker = "UNTRUSTED_VALIDATOR_EXCEPTION" |
| 215 | + |
| 216 | + def raise_known(_report) -> None: |
| 217 | + raise exception_type(marker) |
| 218 | + |
| 219 | + monkeypatch.setattr(artifact_integrity, "validate_advisory_report", raise_known) |
| 220 | + report = build_report() |
| 221 | + call = { |
| 222 | + "canonicalize": artifact_integrity.canonicalize_validated_report, |
| 223 | + "digest": artifact_integrity_digest, |
| 224 | + "evidence": make_artifact_integrity_evidence, |
| 225 | + }[operation] |
| 226 | + |
| 227 | + with pytest.raises(ArtifactIntegrityError, match="report_invalid") as error: |
| 228 | + call(report) |
| 229 | + assert error.value.__suppress_context__ is True |
| 230 | + assert marker not in repr(error.value) |
| 231 | + assert marker not in "".join(traceback.format_exception(error.value)) |
| 232 | + |
| 233 | + |
| 234 | +def test_unexpected_runtime_error_is_not_swallowed(monkeypatch) -> None: |
| 235 | + def raise_unexpected(_report) -> None: |
| 236 | + raise RuntimeError("programming failure") |
| 237 | + |
| 238 | + monkeypatch.setattr(artifact_integrity, "validate_advisory_report", raise_unexpected) |
| 239 | + |
| 240 | + with pytest.raises(RuntimeError, match="programming failure"): |
| 241 | + artifact_integrity_digest(build_report()) |
0 commit comments