SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix(observatory): refine catalog and replay UX

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-30 20:36:12 +03:00
parent 023151c186
commit be83283ee0
18 changed files with 2137 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_session_api.py
+136
View File
@@ -38,6 +38,7 @@ from k1link.sessions.media import _manifest_generation_sha256
from k1link.viewer.recorded import recorded_blueprint_rrd
from k1link.viewer.rerun_bridge import RerunSceneSettings
from k1link.web.camera_archive import CameraArchiveWriter
from k1link.web.observatory_api import build_observatory_router
from k1link.web.session_api import (
LayoutPutRequest,
RecordedBlueprintRequest,
@@ -422,6 +423,141 @@ def test_session_router_rolls_capability_projections_out_only_in_v2(
).status_code == 422
def test_observatory_projection_api_renames_alias_and_deletes_only_projection(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
repository = tmp_path / "repo"
sessions = repository / "sessions"
source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
source_payload = source / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
source_payload_sha256 = hashlib.sha256(source_payload.read_bytes()).hexdigest()
store = SessionStore(repository, data_dir=tmp_path / "data")
store.reconcile_archive(xgrids_k1_archive_source(sessions))
source_snapshot = store.get_session_with_catalog_snapshot(source.name)[1]
capability = LabReplayCapability(
schema_version="missioncore.observation-lab-replay-capability/v1",
kind="canonical-recorded-rerun",
viewer_profile="recorded-session",
timeline="session_time",
activation="explicit",
commands_enabled=False,
)
projection = store.publish_lab_instance(
session_id="lab-observatory-api-projection",
source_session_id=source.name,
display_name="LAB V1 · canonical API projection",
lab_id="LAB V1",
result_kind="recorded-perception-qualification",
result_id="lab-v1-vegetation-shadow-" + "8" * 64,
source_result_id="lab-v1-vegetation-shadow-" + "9" * 64,
run_created_at_utc="2026-08-29T18:05:11.329061+00:00",
include_recorded_media=False,
replay_capability=capability,
provenance={
"replay_capability": capability.as_dict(),
"method": lab_method(),
},
)
cache_sentinel = store.data_dir / "recordings" / projection.session_id / "cached.rrd"
cache_sentinel.parent.mkdir(parents=True)
cache_sentinel.write_bytes(b"must-not-be-touched")
monkeypatch.setattr(
store,
"delete_session",
lambda _session_id: pytest.fail(
"Observatory router must not enter generic evidence/cache deletion"
),
)
application = FastAPI()
application.include_router(build_observatory_router(store))
client = TestClient(application)
url = f"/api/v1/observatory/lab-projections/{projection.session_id}"
assert client.patch(
url,
json={
"schema_version": "missioncore.observatory-lab-projection-rename/v0",
"display_name": "Неверная версия",
},
).status_code == 422
assert client.patch(
url,
json={
"schema_version": "missioncore.observatory-lab-projection-rename/v1",
"display_name": "Новый разбор",
"unexpected": True,
},
).status_code == 422
assert client.patch(
url,
json={
"schema_version": "missioncore.observatory-lab-projection-rename/v1",
"display_name": " ",
},
).status_code == 422
renamed = client.patch(
url,
json={
"schema_version": "missioncore.observatory-lab-projection-rename/v1",
"display_name": " Маршрут у школы ",
},
)
assert renamed.status_code == 200
assert renamed.json() == {
"schema_version": "missioncore.observatory-lab-projection/v1",
"session_id": projection.session_id,
"display_name": "Маршрут у школы",
}
assert store.get_session(projection.session_id).summary.display_name == "Маршрут у школы"
deleted = client.delete(url)
assert deleted.status_code == 204
assert deleted.content == b""
assert client.delete(url).status_code == 404
assert store.get_session_with_catalog_snapshot(source.name)[1] == source_snapshot
assert hashlib.sha256(source_payload.read_bytes()).hexdigest() == source_payload_sha256
assert source.is_dir()
assert cache_sentinel.read_bytes() == b"must-not-be-touched"
def test_observatory_projection_api_rejects_source_and_legacy_lab(
tmp_path: Path,
) -> None:
repository = tmp_path / "repo"
sessions = repository / "sessions"
source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
store = SessionStore(repository, data_dir=tmp_path / "data")
store.reconcile_archive(xgrids_k1_archive_source(sessions))
legacy = store.publish_lab_instance(
session_id="lab-observatory-api-legacy",
source_session_id=source.name,
display_name="LAB E21 · legacy API row",
lab_id="LAB E21",
result_kind="e21-realtime-envelope",
result_id="e21-realtime-envelope-" + "a" * 64,
run_created_at_utc="2026-07-23T15:55:15.548Z",
provenance={"method": lab_method()},
)
application = FastAPI()
application.include_router(build_observatory_router(store))
client = TestClient(application)
rename_document = {
"schema_version": "missioncore.observatory-lab-projection-rename/v1",
"display_name": "Нельзя изменить",
}
for session_id in (source.name, legacy.session_id):
url = f"/api/v1/observatory/lab-projections/{session_id}"
assert client.patch(url, json=rename_document).status_code == 409
assert client.delete(url).status_code == 409
assert source.is_dir()
assert store.get_session(source.name).summary.session_id == source.name
assert store.get_session(legacy.session_id).summary.display_name == "LAB E21 · legacy API row"
def test_delete_session_removes_evidence_and_cache_but_refuses_an_open_recording(
tmp_path: Path,
) -> None:
tests/test_session_store.py
+263
View File
@@ -907,6 +907,9 @@ def test_lab_replay_capability_column_migrates_existing_catalog(
connection.execute(
"ALTER TABLE observation_lab_instances DROP COLUMN include_recorded_media"
)
connection.execute(
"ALTER TABLE observation_lab_instances DROP COLUMN operator_display_name"
)
connection.commit()
migrated = SessionStore(repository, data_dir=data_dir)
@@ -919,6 +922,7 @@ def test_lab_replay_capability_column_migrates_existing_catalog(
}
assert "replay_capability_json" in columns
assert "include_recorded_media" in columns
assert "operator_display_name" in columns
assert migrated.get_lab_instance(legacy.session_id).replay_capability is None
@@ -1248,6 +1252,265 @@ def test_lab_instance_persists_typed_explicit_recorded_replay_capability(
store.get_lab_instance(binding.session_id)
def test_capability_projection_rename_is_only_an_operator_alias(
tmp_path: Path,
) -> None:
repository = tmp_path / "repo"
sessions = repository / "sessions"
source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
source_payload = source / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
source_payload_sha256 = hashlib.sha256(source_payload.read_bytes()).hexdigest()
store = SessionStore(repository, data_dir=tmp_path / "data")
store.reconcile_archive(xgrids_k1_archive_source(sessions))
_source_detail, source_snapshot = store.get_session_with_catalog_snapshot(source.name)
capability = LabReplayCapability(
schema_version="missioncore.observation-lab-replay-capability/v1",
kind="canonical-recorded-rerun",
viewer_profile="recorded-session",
timeline="session_time",
activation="explicit",
commands_enabled=False,
)
parameters = {
"session_id": "lab-recorded-operator-alias",
"source_session_id": source.name,
"display_name": "LAB V1 · canonical recorded replay",
"lab_id": "LAB V1",
"result_kind": "recorded-perception-qualification",
"result_id": "lab-v1-vegetation-shadow-" + "a" * 64,
"source_result_id": "lab-v1-vegetation-shadow-" + "b" * 64,
"run_created_at_utc": "2026-08-29T18:05:11.329061+00:00",
"include_recorded_media": False,
"replay_capability": capability,
"provenance": {
"replay_capability": capability.as_dict(),
"method": lab_method(),
},
}
binding = store.publish_lab_instance(**parameters)
immutable_binding = store.get_lab_instance(binding.session_id)
_projection_detail, projection_snapshot = store.get_session_with_catalog_snapshot(
binding.session_id
)
assert store.rename_capability_lab_projection(
binding.session_id,
" Маршрут у школы ",
) == "Маршрут у школы"
assert store.get_session(binding.session_id).summary.display_name == "Маршрут у школы"
assert (
store.list_recent(scope="laboratory").items[0].display_name
== "Маршрут у школы"
)
assert store.get_lab_instance(binding.session_id) == immutable_binding
assert store.get_session_with_catalog_snapshot(source.name)[1] == source_snapshot
assert (
store.get_session_with_catalog_snapshot(binding.session_id)[1]
== projection_snapshot
)
assert hashlib.sha256(source_payload.read_bytes()).hexdigest() == source_payload_sha256
with sqlite3.connect(store.database_path) as connection:
row = connection.execute(
"SELECT sessions.display_name, lab.operator_display_name "
"FROM observation_sessions AS sessions "
"JOIN observation_lab_instances AS lab USING (session_id) "
"WHERE sessions.session_id = ?",
(binding.session_id,),
).fetchone()
assert row == ("LAB V1 · canonical recorded replay", "Маршрут у школы")
# A strict publisher retry still sees the untouched canonical name and
# immutable provenance, even while the catalog presents the alias.
assert store.publish_lab_instance(**parameters) == binding
assert store.get_session(binding.session_id).summary.display_name == "Маршрут у школы"
assert store.rename_capability_lab_projection(
binding.session_id,
parameters["display_name"],
) == parameters["display_name"]
with sqlite3.connect(store.database_path) as connection:
override = connection.execute(
"SELECT operator_display_name FROM observation_lab_instances "
"WHERE session_id = ?",
(binding.session_id,),
).fetchone()[0]
assert override is None
def test_capability_projection_delete_removes_only_catalog_projection(
tmp_path: Path,
) -> None:
repository = tmp_path / "repo"
sessions = repository / "sessions"
source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
source_payload = source / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
source_payload_sha256 = hashlib.sha256(source_payload.read_bytes()).hexdigest()
store = SessionStore(repository, data_dir=tmp_path / "data")
store.reconcile_archive(xgrids_k1_archive_source(sessions))
source_detail, source_snapshot = store.get_session_with_catalog_snapshot(source.name)
source_replay = store.prepare_replay(source.name)
capability = LabReplayCapability(
schema_version="missioncore.observation-lab-replay-capability/v1",
kind="canonical-recorded-rerun",
viewer_profile="recorded-session",
timeline="session_time",
activation="explicit",
commands_enabled=False,
)
binding = store.publish_lab_instance(
session_id="lab-recorded-delete-only-projection",
source_session_id=source.name,
display_name="LAB V1 · disposable catalog projection",
lab_id="LAB V1",
result_kind="recorded-perception-qualification",
result_id="lab-v1-vegetation-shadow-" + "c" * 64,
source_result_id="lab-v1-vegetation-shadow-" + "d" * 64,
run_created_at_utc="2026-08-29T18:05:11.329061+00:00",
include_recorded_media=False,
replay_capability=capability,
provenance={
"replay_capability": capability.as_dict(),
"method": lab_method(),
},
)
projection_replay = store.prepare_replay(binding.session_id)
cache_sentinel = store.data_dir / "recordings" / binding.session_id / "sentinel.rrd"
cache_sentinel.parent.mkdir(parents=True)
cache_sentinel.write_bytes(b"cache-owned-by-eviction-policy")
assert projection_replay.primary_artifact.path == source_replay.primary_artifact.path
store.delete_capability_lab_projection(binding.session_id)
with pytest.raises(SessionNotFoundError):
store.get_session(binding.session_id)
retained, retained_snapshot = store.get_session_with_catalog_snapshot(source.name)
assert retained == source_detail
assert retained_snapshot == source_snapshot
assert source.is_dir()
assert source_payload.is_file()
assert hashlib.sha256(source_payload.read_bytes()).hexdigest() == source_payload_sha256
assert cache_sentinel.read_bytes() == b"cache-owned-by-eviction-policy"
assert (
store.prepare_replay(source.name).primary_artifact.path
== source_replay.primary_artifact.path
)
with sqlite3.connect(store.database_path) as connection:
assert connection.execute(
"SELECT COUNT(*) FROM observation_lab_instances WHERE session_id = ?",
(binding.session_id,),
).fetchone()[0] == 0
assert connection.execute(
"SELECT COUNT(*) FROM observation_session_artifacts WHERE session_id = ?",
(binding.session_id,),
).fetchone()[0] == 0
def test_capability_projection_mutations_reject_source_legacy_and_corrupt_rows(
tmp_path: Path,
) -> None:
repository = tmp_path / "repo"
sessions = repository / "sessions"
source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
store = SessionStore(repository, data_dir=tmp_path / "data")
store.reconcile_archive(xgrids_k1_archive_source(sessions))
legacy = store.publish_lab_instance(
session_id="lab-legacy-not-observatory-owned",
source_session_id=source.name,
display_name="LAB E21 · legacy",
lab_id="LAB E21",
result_kind="e21-realtime-envelope",
result_id="e21-realtime-envelope-" + "e" * 64,
run_created_at_utc="2026-07-23T15:55:15.548Z",
provenance={"method": lab_method()},
)
capability = LabReplayCapability(
schema_version="missioncore.observation-lab-replay-capability/v1",
kind="canonical-recorded-rerun",
viewer_profile="recorded-session",
timeline="session_time",
activation="explicit",
commands_enabled=False,
)
corrupt = store.publish_lab_instance(
session_id="lab-recorded-corrupt-owner",
source_session_id=source.name,
display_name="LAB V1 · corrupt owner",
lab_id="LAB V1",
result_kind="recorded-perception-qualification",
result_id="lab-v1-vegetation-shadow-" + "f" * 64,
run_created_at_utc="2026-08-29T18:05:11.329061+00:00",
include_recorded_media=False,
replay_capability=capability,
provenance={
"replay_capability": capability.as_dict(),
"method": lab_method(),
},
)
mismatch = store.publish_lab_instance(
session_id="lab-recorded-provenance-mismatch",
source_session_id=source.name,
display_name="LAB V1 · provenance mismatch",
lab_id="LAB V1",
result_kind="recorded-perception-qualification",
result_id="lab-v1-vegetation-shadow-" + "1" * 64,
run_created_at_utc="2026-08-29T18:05:11.329061+00:00",
include_recorded_media=False,
replay_capability=capability,
provenance={
"replay_capability": capability.as_dict(),
"method": lab_method(),
},
)
with sqlite3.connect(store.database_path) as connection:
connection.execute(
"UPDATE observation_sessions SET archive_id = ? WHERE session_id = ?",
("not-observatory-owned", corrupt.session_id),
)
mismatched_provenance = {
"replay_capability": {
**capability.as_dict(),
"commands_enabled": True,
},
"method": lab_method(),
}
connection.execute(
"UPDATE observation_lab_instances SET provenance_json = ? "
"WHERE session_id = ?",
(
json.dumps(mismatched_provenance, separators=(",", ":")),
mismatch.session_id,
),
)
connection.commit()
for session_id in (source.name, legacy.session_id, corrupt.session_id):
with pytest.raises(SessionIntegrityError, match="capability-owned"):
store.rename_capability_lab_projection(session_id, "Нельзя изменить")
with pytest.raises(SessionIntegrityError, match="capability-owned"):
store.delete_capability_lab_projection(session_id)
with pytest.raises(SessionIntegrityError, match="does not match provenance"):
store.rename_capability_lab_projection(mismatch.session_id, "Нельзя изменить")
with pytest.raises(SessionIntegrityError, match="does not match provenance"):
store.delete_capability_lab_projection(mismatch.session_id)
assert store.get_session(source.name).summary.session_id == source.name
assert store.get_session(legacy.session_id).summary.display_name == "LAB E21 · legacy"
with sqlite3.connect(store.database_path) as connection:
retained_ids = {
row[0]
for row in connection.execute(
"SELECT session_id FROM observation_sessions"
).fetchall()
}
assert {
source.name,
legacy.session_id,
corrupt.session_id,
mismatch.session_id,
} <= retained_ids
@pytest.mark.parametrize("value", [0, 1, None, True])
def test_lab_replay_capability_rejects_non_literal_false(value: object) -> None:
with pytest.raises(ValueError, match="replay capability"):