orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

修复 Studio 失败测试并新增合并门禁

Browse Source 
把丢失的 box 录像放回 apps/studio/tests/data/，修掉第 3 节 3–10 号与两处过时测试桩，
注册 timing 标记并让根目录收集零错误；新增 tools/gate.ps1 与可选的 pre-push 钩子。
验证：studio 524 通过 19 跳过、shared 467 通过 18 跳过、ui_check 632 项 0 失败、gate.ps1 退出码 0。
细节见 docs/test-repair-plan.zh.md。
This commit is contained in:
ykcheungab
2026-10-18 19:24:31 +08:00
parent dcabd547fa
commit 433b76e43c
38 changed files with 1036 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/tests/test_replay_analysis_pass.py
+789
View File
@@ -0,0 +1,789 @@
"""Part A: the offline analysis pass and its result cache, checked on real footage.
Three things are being defended here, and each one is a way the feature could be
silently wrong rather than visibly broken:
* the stored payload is **what the live selector returns for that frame** — not a
close approximation of it, and not a different frame;
* a result is either complete or unreadable, so a cancelled pass can never be drawn;
* the pass still costs what it cost when it was measured, so a cache that is correct
but slower than re-running the chain is caught.
The inputs are real: the 33-frame box recording under ``dist/`` (its own ``frames.csv``,
``session.json`` and ``raw_0001.mp4``) and the shipped finger configuration and asset
bundle. The pass is driven with the **box** chain on it for the offset and round-trip
checks, and with the **finger** chain on the same footage for the keyframe, reset and
size checks. The finger chain is what produces a validity mask; the box chain's
painter grid is static without one.
"""
from __future__ import annotations
import hashlib
import inspect
import json
import os
import shutil
import threading
from pathlib import Path
import numpy as np
import pytest
from orisys_app_shared.analysis import model as analysis_model
from orisys_app_shared.analysis.cache import (
DEFAULT_INCOMPLETE_GRACE_S,
INCOMPLETE_MARKER,
LEGACY_RESULT_DIRNAME,
META_NAME,
ResultCache,
evict,
scan_results,
sweep,
)
from orisys_app_shared.analysis.model import (
FRAME_HEADER,
frame_payload_bytes,
result_key,
)
from orisys_app_shared.analysis.reader import (
AnalysisIncompleteError,
AnalysisReader,
)
from orisys_app_shared.analysis.service import (
AnalysisService,
ChainSpec,
chain_spec_for,
)
from orisys_app_shared.scene3d.visual_policy import VisualPolicy
REPO = Path(__file__).resolve().parents[2]
STUDIO = REPO / "dist" / "OrisysStudio"
RECORDING = (
STUDIO
/ "outputs"
/ "recordings"
/ "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011"
)
FINGER_CONFIG = STUDIO / "config" / "sensors" / "finger" / "sensor.json"
FINGER_EXPORT = STUDIO / "config" / "sensors" / "finger" / "assets" / "30fps-0703"
#: Step 0's digest of the finger validity mask, over ``np.packbits(mask)`` with
#: blake2b-16: one value across three videos and 327 frames. Reproduced here from the
#: stored keyframe, which is what makes the mask keyframe a *verified* copy rather than
#: a plausible-looking one. (The result stores a blake2b-8 digest of the same bytes; the
#: longer digest is the comparison because that is the one on record.)
STEP0_MASK_DIGEST = "d76349a3b901aea50915eb1308901c2e"
#: The frame id the reset is applied at. The recording's own ids run 20…52, so this is
#: deliberately not a small number: an off-by-one in the reset rule would land on a
#: frame that does not exist and be invisible.
REFERENCE_FRAME = 36
def _require_inputs() -> None:
missing = [
str(path)
for path in (
RECORDING / "frames.csv",
RECORDING / "raw_0001.mp4",
FINGER_CONFIG,
)
if not path.exists()
]
if missing:
pytest.skip("analysis inputs are not present: " + ", ".join(missing))
pytest.importorskip("orisys", reason="the SDK is needed to run the analysis pass")
def _finger_chain() -> ChainSpec:
return ChainSpec(
sensor_type="finger", config_ref=FINGER_CONFIG, export_dir=FINGER_EXPORT
)
def _box_service(root: Path, **kwargs) -> AnalysisService:
from orisys_app_shared.record.playback import RecordingReader
chain = chain_spec_for(RecordingReader(RECORDING).session())
return AnalysisService(RECORDING, chain=chain, cache=ResultCache(root), **kwargs)
def _finger_service(root: Path, **kwargs) -> AnalysisService:
return AnalysisService(
RECORDING, chain=_finger_chain(), cache=ResultCache(root), **kwargs
)
def _directory_bytes(directory: Path) -> int:
return sum(path.stat().st_size for path in directory.rglob("*") if path.is_file())
# --------------------------------------------------------------------------- fixtures
@pytest.fixture(scope="module")
def box_pass(tmp_path_factory):
_require_inputs()
root = tmp_path_factory.mktemp("analysis-box")
service = _box_service(root)
outcome = service.run()
return service, outcome
@pytest.fixture(scope="module")
def finger_pass(tmp_path_factory):
_require_inputs()
root = tmp_path_factory.mktemp("analysis-finger")
service = _finger_service(root)
outcome = service.run()
return service, outcome
@pytest.fixture(scope="module")
def finger_pass_again(tmp_path_factory):
"""A second pass over the same input, in its own cache root."""
_require_inputs()
root = tmp_path_factory.mktemp("analysis-finger-again")
service = _finger_service(root)
return service, service.run()
@pytest.fixture(scope="module")
def finger_reset_pass(tmp_path_factory):
_require_inputs()
root = tmp_path_factory.mktemp("analysis-finger-reset")
service = _finger_service(root, reference_frame=REFERENCE_FRAME)
return service, service.run()
# --------------------------------------------------------------- frame identity
def test_the_box_chain_consumes_frames_before_the_first_one_a_caller_can_read(box_pass):
"""The offset is measured, recorded, and applied — not assumed.
Constructing a box ``Sensor`` reads seven frames through its stitch probe, so the
first frame the pass can read is the recording's *eighth* video frame. Nothing about
the file says so, which is why the offset comes from the capture's own decode
position and why the resulting frame identity is checked against ``frames.csv``.
"""
_, outcome = box_pass
assert outcome.complete, outcome.reason
assert outcome.provenance.decode_order_offsets == {"1": 7}
assert outcome.provenance.decode_order_offset == 7
assert outcome.provenance.decode_order_offset_source == "measured"
assert outcome.provenance.counts["skipped"] == 7
frames = _box_service(Path(outcome.directory).parent).reader().info().frames
rows = sorted(
(frame for frame in frames if frame.in_video and frame.segment == 1),
key=lambda frame: frame.segment_frame,
)
assert len(rows) == 33
assert outcome.provenance.first_output_frame_id == rows[7].frame_id == 27
assert outcome.provenance.last_output_frame_id == rows[-1].frame_id == 52
assert outcome.frames_written == 26
assert outcome.frames_expected == 26
def test_the_box_result_round_trips_against_a_fresh_run_of_the_chain(box_pass):
"""Frame for frame, the store equals what the chain computes now."""
service, outcome = box_pass
assert outcome.complete, outcome.reason
reader = AnalysisReader(outcome.directory)
try:
checked = 0
for live in service.recompute():
stored = reader.frame(live.row.frame_id)
assert stored.t_rel_s == live.row.t_rel_s
assert live.payload_matches(stored), f"frame {live.row.frame_id} differs"
checked += 1
assert checked == outcome.frames_written == len(reader) == 26
finally:
reader.close()
# ------------------------------------------------------------------- keyframes
def test_the_finger_result_stores_the_candidate_grid_and_the_mask_once(finger_pass):
"""330 candidates on every frame, from one grid keyframe and one mask keyframe.
This is the measurement the format rests on: the candidate *location* set does not
move, so it is stored once and the per-frame record carries only values.
"""
_, outcome = finger_pass
assert outcome.complete, outcome.reason
reader = AnalysisReader(outcome.directory)
try:
assert reader.keyframe_counts == {"grid": 1, "mask": 1}
payloads = list(reader.frames())
assert len(payloads) == 33
assert {payload.candidates for payload in payloads} == {330}
assert len({payload.grid_id for payload in payloads}) == 1
mask = reader.mask()
assert mask is not None and mask.shape == (300, 280)
packed = np.packbits(mask, axis=None).tobytes()
assert hashlib.blake2b(packed, digest_size=16).hexdigest() == STEP0_MASK_DIGEST
assert outcome.provenance.counts["candidates_min"] == 330
assert outcome.provenance.counts["candidates_max"] == 330
assert outcome.provenance.counts["active_zero_frames"] == 5
assert outcome.provenance.counts["active_total"] == 152
finally:
reader.close()
def test_every_stored_frame_reconstructs_what_the_live_selector_returns(finger_pass):
"""The keyframe path is exact: the grid is the one the selector chose, per frame."""
service, outcome = finger_pass
assert outcome.complete, outcome.reason
reader = AnalysisReader(outcome.directory)
try:
policy = service.policy
assert policy.flow_grid_spacing == 10 and policy.max_arrows == 400
for live in service.recompute():
stored = reader.frame(live.row.frame_id)
assert live.payload_matches(stored), f"frame {live.row.frame_id} differs"
rebuilt = reader.samples(live.row.frame_id)
assert np.array_equal(rebuilt.rows, np.asarray(live.samples.rows, np.int32))
assert np.array_equal(rebuilt.cols, np.asarray(live.samples.cols, np.int32))
assert np.array_equal(
np.asarray(rebuilt.magnitude, np.float32),
np.asarray(live.samples.magnitude, np.float32),
)
assert np.array_equal(rebuilt.active, np.asarray(live.samples.active, bool))
finally:
reader.close()
def test_a_threshold_change_is_served_without_rebuilding_the_result(finger_pass):
"""Display settings are not in the key because the record can answer a new threshold.
Activity is stored, but magnitude is stored too — so a viewer with a different
``flow_on_px`` gets a correct answer from the same file.
"""
_, outcome = finger_pass
reader = AnalysisReader(outcome.directory)
try:
frame_id = int(reader.frame_ids[0])
recorded = reader.samples(frame_id)
impossible = reader.samples(frame_id, policy=VisualPolicy(flow_on_px=1e9))
assert impossible.active_count == 0
permissive = reader.samples(
frame_id, policy=VisualPolicy(flow_on_px=0.0, flow_off_px=0.0)
)
assert permissive.active_count >= recorded.active_count
assert np.array_equal(permissive.magnitude, recorded.magnitude)
assert np.array_equal(
permissive.active, np.asarray(permissive.magnitude >= 0.0, bool)
)
finally:
reader.close()
# ------------------------------------------------------------------ determinism
def test_two_passes_over_the_same_input_store_byte_identical_payloads(
finger_pass, finger_pass_again
):
"""Re-running the pass produces the same bytes, so a cache hit cannot drift.
The determinism contract is the **frame records** and the **keyframe digests**. The
metadata is deliberately outside it: ``elapsed_s``, ``open_seconds``, ``started_at``
and ``finished_at`` describe the run, not the measurement, and two runs of the same
pass cannot share them.
"""
_, first = finger_pass
_, second = finger_pass_again
assert first.complete and second.complete
reader_a = AnalysisReader(first.directory)
reader_b = AnalysisReader(second.directory)
try:
ids = reader_a.frame_ids
assert np.array_equal(ids, reader_b.frame_ids)
assert reader_a.frame_record_bytes == reader_b.frame_record_bytes
for frame_id in ids:
assert (
reader_a.frame(int(frame_id)).canonical_bytes()
== reader_b.frame(int(frame_id)).canonical_bytes()
), f"frame {frame_id} is not reproducible"
assert {key: value.digest for key, value in reader_a.keyframes.items()} == {
key: value.digest for key, value in reader_b.keyframes.items()
}
assert reader_a.mask_digest == reader_b.mask_digest
finally:
reader_a.close()
reader_b.close()
# ------------------------------------------------------------------ cancellation
def test_an_interrupted_pass_is_incomplete_and_is_not_served(tmp_path):
"""A cancelled pass leaves a directory that refuses to open by default."""
_require_inputs()
root = tmp_path / "cache"
cancelled = threading.Event()
seen: list[int] = []
def on_progress(progress):
seen.append(progress.frame_id)
if len(seen) >= 5:
cancelled.set()
service = _finger_service(root, on_progress=on_progress, cancelled=cancelled)
outcome = service.run()
assert outcome.complete is False
assert outcome.reason == "cancelled"
assert 0 < outcome.frames_written < 33
assert (outcome.directory / INCOMPLETE_MARKER).is_file()
meta = json.loads((outcome.directory / META_NAME).read_text(encoding="utf-8"))
assert meta["complete"] is False
assert meta["stopped_reason"] == "cancelled"
assert meta["provenance"]["complete"] is False
assert [item.complete for item in scan_results(root)] == [False]
with pytest.raises(AnalysisIncompleteError):
AnalysisReader(outcome.directory)
# The partial data is still inspectable on purpose, and says what it is.
partial = AnalysisReader(outcome.directory, allow_incomplete=True)
try:
assert partial.complete is False
assert len(partial) == outcome.frames_written
assert "incomplete" in partial.stale_reasons()
finally:
partial.close()
def test_a_declared_segment_that_is_missing_makes_the_result_incomplete(tmp_path):
"""A segment the session declares and the directory lacks is a failure, not a short read."""
_require_inputs()
recording = tmp_path / "recording"
recording.mkdir()
shutil.copy2(RECORDING / "frames.csv", recording / "frames.csv")
shutil.copy2(RECORDING / "forces.csv", recording / "forces.csv")
shutil.copy2(RECORDING / "events.csv", recording / "events.csv")
session = json.loads((RECORDING / "session.json").read_text(encoding="utf-8"))
session["stats"]["segments"] = 2
(recording / "session.json").write_text(json.dumps(session), encoding="utf-8")
try:
os.link(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4")
except OSError:
shutil.copy2(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4")
service = AnalysisService(
recording, chain=_finger_chain(), cache=ResultCache(tmp_path / "cache")
)
outcome = service.run()
assert outcome.complete is False
assert outcome.reason == "segments_missing"
assert outcome.frames_written == 0
assert outcome.provenance.segments == [1]
assert outcome.provenance.segments_declared == 2
assert outcome.provenance.segments_missing == [2]
with pytest.raises(AnalysisIncompleteError):
AnalysisReader(outcome.directory)
def test_a_reference_frame_outside_the_recording_is_refused(tmp_path):
"""The reset is applied at a named frame or the pass does not run.
Frame ids come from the acquisition session and need not start at zero or one: this
recording's run 20…52. A requested reference of 16 therefore names nothing, and
ignoring that would store a result whose ``reference_frame_id`` is not the frame
that was asked for.
"""
_require_inputs()
service = _finger_service(tmp_path / "cache", reference_frame=16)
with pytest.raises(ValueError, match="reference frame 16"):
service.run()
assert not (tmp_path / "cache").exists(), (
"a refused pass must not leave a directory"
)
# ------------------------------------------------------------------ reference reset
def test_resetting_the_reference_records_the_frame_and_clears_the_tracker(
finger_pass, finger_reset_pass
):
"""The reset is exact at the named frame and touches nothing before it.
``Sensor.reset`` clears the optical-flow tracker, so the frame the reset is applied
at measures against an empty reference: its displacement is exactly zero — the
Step 0 signature. The frames before it are byte-identical to a pass with no reset,
which is what makes "reference frame" a statement about a position rather than about
a slightly different run.
"""
_, plain = finger_pass
_, reset = finger_reset_pass
assert reset.complete, reset.reason
assert reset.provenance.reference_is_explicit is True
assert reset.provenance.reference_frame_id == REFERENCE_FRAME
assert reset.provenance.first_output_frame_id == 20
assert reset.provenance.counts["reset_count"] == 1
assert reset.provenance.counts["reset_seconds"] >= 0.0
# With no explicit reset the reference is the pass's own clean start.
assert plain.provenance.reference_is_explicit is False
assert plain.provenance.reference_frame_id == plain.provenance.first_output_frame_id
reader = AnalysisReader(plain.directory)
reset_reader = AnalysisReader(reset.directory)
try:
before = int(REFERENCE_FRAME) - 1
assert reset_reader.frame(REFERENCE_FRAME).magnitude.max() == 0.0
assert reader.frame(REFERENCE_FRAME).magnitude.max() > 0.0
assert reset_reader.frame(before).magnitude.max() > 0.0
for frame_id in reader.frame_ids:
if int(frame_id) < REFERENCE_FRAME:
assert (
reader.frame(int(frame_id)).canonical_bytes()
== reset_reader.frame(int(frame_id)).canonical_bytes()
), f"frame {frame_id} before the reference was disturbed by the reset"
assert (
reader.frame(int(REFERENCE_FRAME)).canonical_bytes()
!= reset_reader.frame(int(REFERENCE_FRAME)).canonical_bytes()
)
finally:
reader.close()
reset_reader.close()
# ------------------------------------------------------------------ the key
def test_the_key_follows_the_inputs_that_change_the_result_and_no_others(finger_pass):
"""Five components, and display settings are deliberately not one of them."""
base = dict(
recording="rec-a",
sensor_config="config-a",
calibration="cal-a",
reference_frame=36,
)
digest = result_key(**base).digest
assert result_key(**base).digest == digest
assert result_key(**{**base, "sensor_config": "config-b"}).digest != digest
assert result_key(**{**base, "algorithm": "other-chain"}).digest != digest
assert result_key(**{**base, "calibration": "cal-b"}).digest != digest
assert result_key(**{**base, "reference_frame": None}).digest != digest
assert result_key(**{**base, "recording": "rec-b"}).digest != digest
# Nothing in the signature can carry a display setting, so a change to one cannot
# move the key even by accident.
parameters = set(inspect.signature(result_key).parameters)
assert not {name for name in parameters if "policy" in name or "display" in name}
assert "policy" not in parameters
# And the concrete consequence: two services that differ only in the policy the
# samples are selected with address the same stored result.
service = _finger_service(finger_pass[1].directory.parent)
other = _finger_service(finger_pass[1].directory.parent)
other.policy = VisualPolicy(flow_on_px=1.5, max_arrows=1)
assert (
service.dependency_report()["key"].digest
== other.dependency_report()["key"].digest
== finger_pass[1].key.digest
)
recorded = finger_pass[1].provenance.policy
assert recorded["flow_on_px"] == 0.4 and recorded["max_arrows"] == 400
# ------------------------------------------------------------------ size
def test_the_stored_size_is_the_measured_encoding(box_pass, finger_pass):
"""State the encoding, then hold it: 4,048 B per frame at the finger grid.
Implicit candidate grid (a keyframe) + float32 ``dx``/``dy``/``magnitude`` +
bit-packed ``active`` + a 32-byte header + the 8-byte validity digest + the 6-byte
depth header: ``32 + 330*12 + 42 + 8 + 6 = 4,048`` B, of which the depth part is
``14`` B over the 4,034 B this record cost before the depth channel was stored. A
frame whose depth map is *not* all zeros adds its own zlib'd uint8 plane on top.
The alternative measured encodings are float16 ``dxy``+magnitude (2,022 B) and
dropping magnitude (4,002 B with the grid, no header); float32 is chosen because
the round trip has to be *exact*, and the magnitude is what serves a changed
threshold.
"""
fixed = FRAME_HEADER.size + 330 * 12 + 42 + 8
assert fixed == 4042
assert frame_payload_bytes(330, with_validity=True) == 4042
assert frame_payload_bytes(330, with_validity=True, with_depth=True) == 4048
assert frame_payload_bytes(330, with_validity=False, with_depth=True) == 4040
assert frame_payload_bytes(400, with_validity=False) == 4882
assert frame_payload_bytes(400, with_validity=False, with_depth=True) == 4888
_, finger = finger_pass
reader = AnalysisReader(finger.directory)
try:
# Every frame states the depth channel; the ones whose map is all zeros pay the
# 6-byte header and nothing else.
frames = list(reader.frames())
emptied = [item for item in frames if item.has_depth and not item.depth_nonzero]
carrying = [item for item in frames if item.depth_nonzero]
assert len(emptied) + len(carrying) == len(frames)
assert reader.frame_record_bytes == sum(
fixed + item.depth_stored_bytes for item in frames
)
assert reader.frame_record_bytes == (
len(emptied) * 4048
+ len(carrying) * fixed
+ sum(item.depth_stored_bytes for item in carrying)
)
assert reader.bytes_per_frame <= 4096, "the stated per-frame bound"
total_per_frame = _directory_bytes(finger.directory) / len(reader)
assert total_per_frame <= 4600, "keyframes and metadata included"
finally:
reader.close()
_, box = box_pass
box_reader = AnalysisReader(box.directory)
try:
# The box chain has no validity mask. Its sample grid is the live painter's
# (spacing 20 over the 365 stitch, every cell kept): 18 x 18 = 324 candidates,
# one keyframe, and the per-frame record is values plus whatever depth it carries.
frames = list(box_reader.frames())
assert {item.candidates for item in frames} == {324}
assert box_reader.frame_record_bytes == sum(
frame_payload_bytes(324, with_validity=False) + item.depth_stored_bytes
for item in frames
)
assert box_reader.keyframe_counts == {"grid": 1}
assert box_reader.mask() is None
finally:
box_reader.close()
# ------------------------------------------------------------------ reader
def test_a_stored_result_is_served_without_the_recording_or_a_replay(
finger_pass, tmp_path
):
"""A seek is a file read: the result answers with the recording and SDK out of reach."""
_, outcome = finger_pass
copy = tmp_path / "copied-result"
shutil.copytree(outcome.directory, copy)
original = AnalysisReader(outcome.directory)
copied = AnalysisReader(copy)
try:
assert np.array_equal(original.frame_ids, copied.frame_ids)
for frame_id in original.frame_ids:
assert (
original.frame(int(frame_id)).canonical_bytes()
== copied.frame(int(frame_id)).canonical_bytes()
)
# Random access by frame id and by timestamp both land on the right frame.
times = copied.times
middle = len(times) // 2
assert copied.frame(int(copied.frame_ids[middle])).t_rel_s == times[middle]
assert copied.frame_at(float(times[middle])).frame_id == int(
copied.frame_ids[middle]
)
assert copied.frame_at(float(times[middle]) + 1e-6).frame_id == int(
copied.frame_ids[middle]
)
assert copied.frame_at(-1.0).frame_id == int(copied.frame_ids[0])
assert copied.frame_at(float(times[-1]) + 100.0).frame_id == int(
copied.frame_ids[-1]
)
window = list(
copied.frames(start=int(copied.frame_ids[1]), stop=int(copied.frame_ids[2]))
)
assert [payload.frame_id for payload in window] == [
int(copied.frame_ids[1]),
int(copied.frame_ids[2]),
]
assert copied.has(int(copied.frame_ids[0]))
assert not copied.has(10_000)
finally:
original.close()
copied.close()
def test_eviction_removes_whole_results_against_the_cap(tmp_path):
"""The cap is enforced by removing complete results, oldest first, never a mistake."""
root = tmp_path / "cache"
root.mkdir()
made: list[Path] = []
for index in range(3):
directory = root / f"result-{index}"
directory.mkdir()
(directory / META_NAME).write_text(
json.dumps({"complete": True, "key": f"result-{index}", "format": "x"}),
encoding="utf-8",
)
(directory / "blob.bin").write_bytes(b"\0" * 4096)
os.utime(directory / META_NAME, (1_600_000_000 + index, 1_600_000_000 + index))
made.append(directory)
def size_of(path: Path) -> int:
return sum(item.stat().st_size for item in path.rglob("*") if item.is_file())
kept = made[2]
# A cap that one removal satisfies: the oldest complete result goes, and the
# protected one is never a candidate whatever the arithmetic says.
cap = sum(size_of(item) for item in made) - size_of(made[0])
removed = evict(root, max_bytes=cap, keep=[kept])
assert removed == [made[0]]
assert made[0].exists() is False
assert kept.is_dir() and made[1].is_dir()
assert [item.directory.name for item in scan_results(root)] == [
"result-1",
"result-2",
]
# A dry run reports without deleting, and still never proposes the protected result.
assert evict(root, max_bytes=1, dry_run=True, keep=[kept]) == [made[1]]
assert made[1].is_dir()
assert kept.is_dir()
# ------------------------------------------------- cache root and lifetime policy
def _write_result(root: Path, name: str, *, complete: bool, modified: float, blob: int = 4096):
"""One result directory as the writer leaves it, with a chosen age."""
directory = root / name
directory.mkdir(parents=True)
(directory / META_NAME).write_text(
json.dumps({"complete": complete, "key": name, "format": "x"}), encoding="utf-8"
)
(directory / "blob.bin").write_bytes(b"\0" * blob)
if not complete:
(directory / INCOMPLETE_MARKER).write_text("{}", encoding="utf-8")
for path in directory.rglob("*"):
os.utime(path, (modified, modified))
os.utime(directory, (modified, modified))
return directory
def test_the_cache_root_is_the_app_s_own_and_never_the_recording_s(tmp_path):
"""Results go to the app's derived-data root; a recording folder is never written.
This is the owner's constraint made checkable: the recording a result describes is
an input, and an input directory must come back byte-identical. It is also what makes
the cache findable at all — the *key* is path-independent, so one root for every
recording serves a copy exactly as it serves the original, while a root inside the
recording only ever served the one folder it was written beside.
"""
from orisys_app_shared import paths as paths_module
assert paths_module.analysis_cache_dir() == (
paths_module.derived_data_dir() / paths_module.ANALYSIS_CACHE_DIRNAME
)
assert paths_module.ANALYSIS_CACHE_DIRNAME in str(paths_module.analysis_cache_dir())
# The recording's own folder is not the root, and ``for_recording`` does not put it
# back: the recording is named for the key, not for the layout.
recording = tmp_path / "recording"
recording.mkdir()
cache = ResultCache.for_recording(recording)
assert cache.root != recording / LEGACY_RESULT_DIRNAME
assert not str(cache.root).startswith(str(recording))
assert cache.directory(result_key(
recording="r", sensor_config="s", calibration="c"
)).parent == cache.root
# …and the root it defaults to is the OS one, resolved at construction so a caller
# can point it elsewhere without patching this class.
assert ResultCache().root == paths_module.analysis_cache_dir()
def test_a_result_under_the_old_root_is_ignored_not_migrated(tmp_path):
"""A leftover ``<recording>/analysis`` neither breaks an open nor gets touched.
The old root is a *recording the operator owns*, so the app may neither read it
(the path is not the identity) nor delete it (it is not the app's data). A result
for the same input under the app's own root is what is served, and the old directory
is found exactly as it was left.
"""
from orisys_app_shared.analysis.model import result_key
recording = tmp_path / "recording"
legacy = recording / LEGACY_RESULT_DIRNAME
legacy.mkdir(parents=True)
key = result_key(recording="r", sensor_config="s", calibration="c")
(legacy / key.digest).mkdir()
(legacy / key.digest / META_NAME).write_text(
json.dumps({"complete": True, "key": key.digest}), encoding="utf-8"
)
before = sorted(str(path.relative_to(recording)) for path in recording.rglob("*"))
app_root = tmp_path / "app-cache"
app_root.mkdir()
cache = ResultCache(app_root)
assert cache.exists(key) is False, "a result beside the recording answered for it"
_write_result(app_root, key.digest, complete=True, modified=1_600_000_000)
assert cache.exists(key) is True
# A sweep of the app root does not reach into the recording: the old result and the
# recording's own files are all still there, untouched.
assert sweep(app_root, max_bytes=0) is not None
assert legacy.is_dir()
assert sorted(
str(path.relative_to(recording)) for path in recording.rglob("*")
) == before
def test_the_sweep_removes_unfinished_results_but_only_after_the_grace(tmp_path):
"""Garbage first: an unreadable result goes once it is old, and not before.
An incomplete result can never be served (Part A §7), so it is the cheapest thing to
lose — but it is also the only evidence of how far a cancelled pass got, which is why
the age test exists rather than "delete on sight".
"""
root = tmp_path / "cache"
root.mkdir()
now = 1_700_000_000.0
stale = _write_result(root, "stale", complete=False, modified=now - 7200)
fresh = _write_result(root, "fresh", complete=False, modified=now - 30)
finished = _write_result(root, "finished", complete=True, modified=now - 7200)
report = sweep(root, incomplete_grace_s=DEFAULT_INCOMPLETE_GRACE_S, now=now)
assert report.removed == [stale], "the sweep took something it should not have"
assert report.incomplete == [stale]
assert stale.exists() is False
assert fresh.is_dir(), "a pass that may still be running lost its directory"
assert finished.is_dir(), "a readable result was removed while it fits the cap"
assert report.freed_bytes > 0
def test_the_sweep_protects_what_is_in_use_and_honours_the_cap(tmp_path):
"""Then the cap: whole results, oldest first, and never the one being served."""
root = tmp_path / "cache"
root.mkdir()
now = 1_700_000_000.0
made = [
_write_result(root, f"result-{index}", complete=True, modified=now - 100 + index)
for index in range(3)
]
kept = made[2]
size = sum(path.stat().st_size for path in kept.rglob("*") if path.is_file())
# A cap that fits only the protected result: both others must go, and the protected
# one must survive even when the arithmetic would prefer it (it is the newest).
report = sweep(root, max_bytes=size, keep=[kept], now=now)
assert kept.is_dir()
assert made[0].exists() is False and made[1].exists() is False
assert set(report.removed) == {made[0], made[1]}
assert report.incomplete == []
assert report.remaining_bytes <= size + 64
# A dry run removes nothing and reports what would go.
again = _write_result(root, "result-3", complete=True, modified=now)
plan = sweep(root, max_bytes=1, keep=[kept], now=now, dry_run=True)
assert again.is_dir() and kept.is_dir()
assert set(plan.removed) == {again}
# A root that does not exist is not an error: the first run of an app has no cache.
missing = sweep(tmp_path / "absent", now=now)
assert missing.removed == [] and missing.remaining_bytes == 0
# ------------------------------------------------------------------ incompleteness gate
def test_a_result_without_metadata_is_never_served(tmp_path):
"""The only thing that opens is a directory whose metadata says it finished."""
directory = tmp_path / "empty-result"
directory.mkdir()
from orisys_app_shared.analysis.reader import AnalysisNotFoundError
with pytest.raises(AnalysisNotFoundError):
AnalysisReader(directory)
(directory / META_NAME).write_text(
json.dumps(
{
"format": analysis_model.ANALYSIS_FORMAT,
"format_version": analysis_model.ANALYSIS_FORMAT_VERSION,
}
),
encoding="utf-8",
)
with pytest.raises(AnalysisIncompleteError):
AnalysisReader(directory)