Files
Renderive/render_3D/datoviz/testing/test_release_conformance.py
T
2026-08-14 01:54:38 +08:00

433 lines
14 KiB
Python

"""Tests for installed-wheel release conformance evidence."""
from __future__ import annotations
import json
import struct
import tarfile
import zlib
from argparse import Namespace
from pathlib import Path
from types import SimpleNamespace
from tools.release_wheels import conformance
def _chunk(kind: bytes, payload: bytes) -> bytes:
return (
struct.pack('>I', len(payload))
+ kind
+ payload
+ struct.pack('>I', zlib.crc32(kind + payload) & 0xFFFFFFFF)
)
def _png(path: Path, rgba: tuple[int, int, int, int]) -> None:
width, height = 2, 2
pixels = bytes(rgba) * width
raw = b''.join(b'\0' + pixels for _ in range(height))
path.write_bytes(
b'\x89PNG\r\n\x1a\n'
+ _chunk(b'IHDR', struct.pack('>IIBBBBB', width, height, 8, 6, 0, 0, 0))
+ _chunk(b'IDAT', zlib.compress(raw))
+ _chunk(b'IEND', b'')
)
def _evidence(
root: Path, machine_id: str, status: str = 'pass', *, physical: bool = False
) -> Path:
evidence_dir = root / machine_id
captures = evidence_dir / 'captures'
captures.mkdir(parents=True)
_png(captures / 'point.png', (20, 40, 60, 255))
environment = {
'platform': {'system': 'TestOS', 'machine': 'test64'},
'execution_class': 'physical-interactive' if physical else 'github-hosted-software-gpu',
}
(evidence_dir / 'environment.json').write_text(json.dumps(environment), encoding='utf8')
manual = {
'state': 'approved' if physical else 'not-applicable',
'scenarios': [],
}
(evidence_dir / 'manual-observations.json').write_text(json.dumps(manual), encoding='utf8')
evidence = {
'schema': conformance.EVIDENCE_SCHEMA,
'version': '0.4.0rc1',
'campaign': {
'wheel_run_id': '123',
'artifact_commit': 'artifact',
'validator_commit': 'validator',
},
'machine_id': machine_id,
'execution_class': environment['execution_class'],
'mode': 'physical' if physical else 'unattended',
'artifact_checksums': {'wheel': {'name': 'wheel.whl', 'sha256': 'abc'}},
'environment': 'environment.json',
'results': [
{
'id': 'installed-wheel-native-window-1',
'status': 'pass',
'returncode': 0,
}
],
'captures': [
{
'scenario': 'point',
'path': 'captures/point.png',
'sha256': conformance.sha256(captures / 'point.png'),
'deterministic': True,
'stats': conformance.png_stats(captures / 'point.png'),
}
],
'capture_comparisons': [
{
'id': 'c-python-point-render',
'status': 'pass',
'scenarios': ['point'],
}
],
'manual': manual,
'skips': [],
'failures': [] if status == 'pass' else [{'id': 'test'}],
'status': status,
}
(evidence_dir / 'evidence.json').write_text(json.dumps(evidence), encoding='utf8')
return evidence_dir
def test_png_stats_rgba(tmp_path: Path) -> None:
"""RGBA PNG statistics expose deterministic channel facts."""
path = tmp_path / 'capture.png'
_png(path, (20, 40, 60, 255))
stats = conformance.png_stats(path)
assert stats['width'] == 2
assert stats['height'] == 2
assert stats['channel_mean'] == [20.0, 40.0, 60.0, 255.0]
assert stats['nontransparent_fraction'] == 1.0
assert len(stats['pixel_sha256']) == 64
def test_physical_evidence_machine_matches_campaign_artifact() -> None:
"""Physical sync discovers accepted machines from authoritative artifact names."""
assert (
conformance.physical_evidence_machine(
'physical-evidence-29641789685-macbook-m3', '29641789685'
)
== 'macbook-m3'
)
assert conformance.physical_evidence_machine('physical-evidence-other', '29641789685') == ''
assert (
conformance.physical_evidence_machine(
'physical-evidence-29641789684-macbook-m3', '29641789685'
)
== ''
)
def test_capture_parity_compares_decoded_pixels() -> None:
"""Equivalent C and Python scenarios require identical decoded pixels."""
scenarios = conformance.CAPTURE_PARITY_GROUPS['c-python-point-render']
captures = [
{
'scenario': scenario,
'stats': {
'width': 2,
'height': 2,
'channels': 4,
'pixel_sha256': 'same-pixels',
},
}
for scenario in scenarios
]
comparisons = conformance.capture_parity_records(captures)
assert comparisons[0]['status'] == 'pass'
captures[-1]['stats']['pixel_sha256'] = 'different-pixels'
assert conformance.capture_parity_records(captures)[0]['status'] == 'fail'
def test_validate_wheels_run_uses_workflow_path_not_display_case() -> None:
"""Run identity follows the canonical file even when its API name is title-cased."""
run = {
'name': 'Wheels',
'path': '.github/workflows/wheels.yml',
'status': 'completed',
'conclusion': 'success',
}
conformance.validate_wheels_run(run, '123')
def test_run_parser_records_capability_skips() -> None:
"""Hosted lanes can disable unavailable checks without changing shared defaults."""
args = conformance.parse_args(
[
'run',
'--wheel',
'wheel.whl',
'--output-dir',
'evidence',
'--version',
'0.4.0rc1',
'--wheel-run-id',
'123',
'--artifact-commit',
'artifact',
'--machine-id',
'windows-arm64',
'--execution-class',
'github-hosted-no-gpu',
'--no-render',
'--no-cmake-consumer',
'--no-examples',
]
)
assert args.render is False
assert args.cmake_consumer is False
assert args.examples is False
assert args.shaderc is True
def test_aggregate_builds_self_contained_report(tmp_path: Path) -> None:
"""Aggregation preserves captures and writes HTML, JSON, and a manifest."""
inputs = tmp_path / 'inputs'
_evidence(inputs, 'linux-ci')
_evidence(inputs, 'macbook-m3')
output = tmp_path / 'report'
args = Namespace(
input=[inputs],
output_dir=output,
replace=False,
strict=True,
expected_machine=['linux-ci', 'macbook-m3'],
)
assert conformance.aggregate(args) == 0
report = json.loads((output / 'report.json').read_text(encoding='utf8'))
assert report['status'] == 'pass'
assert len(report['evidence']) == 2
assert (output / 'index.html').is_file()
assert (output / 'manifest.json').is_file()
assert (output / 'platforms/linux-ci/captures/point.png').is_file()
report_html = (output / 'index.html').read_text(encoding='utf8')
assert 'Full environment and Vulkan metadata' in report_html
assert 'Capture statistics' in report_html
assert 'Release gates' in report_html
assert report['gates']['cross_frontend_render_parity'] == 'pass'
assert 'Cross-frontend capture parity' in report_html
def test_aggregate_writes_report_before_strict_failure(tmp_path: Path) -> None:
"""A failing platform still produces the diagnostic report."""
inputs = tmp_path / 'inputs'
_evidence(inputs, 'windows-ci', status='fail')
output = tmp_path / 'report'
args = Namespace(
input=[inputs],
output_dir=output,
replace=False,
strict=True,
expected_machine=['windows-ci'],
)
assert conformance.aggregate(args) == 1
assert (output / 'index.html').is_file()
assert json.loads((output / 'report.json').read_text())['status'] == 'fail'
def test_aggregate_rejects_cross_frontend_capture_mismatch(tmp_path: Path) -> None:
"""A frontend parity mismatch fails the report even when each repeat is deterministic."""
evidence_dir = _evidence(tmp_path / 'inputs', 'macos-ci')
evidence_path = evidence_dir / 'evidence.json'
evidence = json.loads(evidence_path.read_text(encoding='utf8'))
evidence['capture_comparisons'][0]['status'] = 'fail'
evidence_path.write_text(json.dumps(evidence), encoding='utf8')
output = tmp_path / 'report'
args = Namespace(
input=[evidence_dir],
output_dir=output,
replace=False,
strict=True,
expected_machine=['macos-ci'],
)
assert conformance.aggregate(args) == 1
report = json.loads((output / 'report.json').read_text())
assert report['gates']['cross_frontend_render_parity'] == 'fail'
def test_native_window_gate_respects_declared_capability_skip() -> None:
"""A rendering-capable hosted runner need not expose an interactive desktop session."""
linux = {
'results': [{'id': 'installed-wheel-native-window-1', 'status': 'pass'}],
'skips': [],
}
hosted_macos = {
'execution_class': 'github-hosted-hardware-gpu',
'results': [],
'skips': [
{
'id': 'native-window',
'reason': 'hosted runner has no windowed graphics device',
}
],
}
gates = conformance.report_gates([linux, hosted_macos], [])
assert gates['installed_wheel_native_window'] == 'pass'
def test_native_window_gate_requires_result_without_skip() -> None:
"""Evidence that claims window capability must contain a passing window result."""
gates = conformance.report_gates([{'results': [], 'skips': []}], [])
assert gates['installed_wheel_native_window'] == 'fail'
def test_aggregate_rejects_missing_expected_machine(tmp_path: Path) -> None:
"""Missing matrix evidence is a report failure rather than an implicit pass."""
inputs = tmp_path / 'inputs'
_evidence(inputs, 'linux-ci')
output = tmp_path / 'report'
args = Namespace(
input=[inputs],
output_dir=output,
replace=False,
strict=True,
expected_machine=['linux-ci', 'windows-ci'],
)
assert conformance.aggregate(args) == 1
report = json.loads((output / 'report.json').read_text())
assert report['missing_machines'] == ['windows-ci']
def test_aggregate_keeps_pending_physical_human_gate_non_green(tmp_path: Path) -> None:
"""Passing unattended work cannot promote pending human observations."""
inputs = tmp_path / 'inputs'
evidence_dir = _evidence(inputs, 'macbook-m3', physical=True)
evidence_path = evidence_dir / 'evidence.json'
evidence = json.loads(evidence_path.read_text(encoding='utf8'))
evidence['manual']['state'] = 'pending'
evidence_path.write_text(json.dumps(evidence), encoding='utf8')
(evidence_dir / 'manual-observations.json').write_text(
json.dumps(evidence['manual']), encoding='utf8'
)
output = tmp_path / 'report'
args = Namespace(
input=[inputs],
output_dir=output,
replace=False,
strict=True,
expected_machine=['macbook-m3'],
)
assert conformance.aggregate(args) == 1
report = json.loads((output / 'report.json').read_text())
assert report['status'] == 'pending'
assert report['gates']['physical_unattended'] == 'pass'
assert report['gates']['physical_human_interaction'] == 'pending'
def test_verify_bundle_accepts_approved_physical_evidence(tmp_path: Path) -> None:
"""Intake accepts one approved bundle with matching immutable identity."""
evidence_dir = _evidence(tmp_path / 'source', 'macbook-m3', physical=True)
archive = tmp_path / 'evidence.tar.gz'
with tarfile.open(archive, 'w:gz') as tar:
tar.add(evidence_dir, arcname='macbook-m3')
output = tmp_path / 'accepted'
args = Namespace(
archive=archive,
output_dir=output,
wheel_run_id='123',
machine_id='macbook-m3',
version='0.4.0rc1',
replace=False,
)
assert conformance.verify_bundle(args) == 0
assert (output / 'macbook-m3/evidence.json').is_file()
def test_submit_physical_passes_oras_a_relative_archive(
tmp_path: Path, monkeypatch
) -> None:
"""ORAS keeps path validation enabled by receiving a layer relative to its working directory."""
evidence_dir = _evidence(tmp_path / 'source', 'macbook-m3', physical=True)
calls = []
def fake_run(argv, **kwargs):
calls.append((argv, kwargs))
if argv[0] == 'oras':
return SimpleNamespace(
returncode=0,
stdout=json.dumps(
{
'reference': 'ghcr.io/datoviz/evidence@sha256:' + 'a' * 64,
}
),
stderr='',
)
return SimpleNamespace(returncode=0)
monkeypatch.setattr(conformance.subprocess, 'run', fake_run)
args = Namespace(
evidence_dir=evidence_dir,
package='ghcr.io/datoviz/evidence',
ref='v0.4-dev',
confirm='yes',
)
assert conformance.submit_physical(args) == 0
push, kwargs = calls[0]
archive_layer = push[-1].partition(':application/')[0]
assert Path(archive_layer).name == archive_layer
assert kwargs['cwd'] == evidence_dir.parent
def test_verify_bundle_rejects_tampered_capture(tmp_path: Path) -> None:
"""Intake verifies every recorded image checksum after extraction."""
evidence_dir = _evidence(tmp_path / 'source', 'macbook-m3', physical=True)
_png(evidence_dir / 'captures/point.png', (200, 40, 60, 255))
archive = tmp_path / 'evidence.tar.gz'
with tarfile.open(archive, 'w:gz') as tar:
tar.add(evidence_dir, arcname='macbook-m3')
args = Namespace(
archive=archive,
output_dir=tmp_path / 'accepted',
wheel_run_id='123',
machine_id='macbook-m3',
version='0.4.0rc1',
replace=False,
)
try:
conformance.verify_bundle(args)
except ValueError as exc:
assert 'capture checksum mismatch' in str(exc)
else:
raise AssertionError('tampered capture was accepted')
def test_safe_extract_rejects_path_traversal(tmp_path: Path) -> None:
"""Evidence archive extraction cannot write outside its intake directory."""
payload = tmp_path / 'payload'
payload.write_text('unsafe', encoding='utf8')
archive = tmp_path / 'unsafe.tar.gz'
with tarfile.open(archive, 'w:gz') as tar:
tar.add(payload, arcname='../escaped')
try:
conformance._safe_extract(archive, tmp_path / 'accepted')
except ValueError as exc:
assert 'unsafe evidence archive path' in str(exc)
else:
raise AssertionError('path traversal archive was accepted')