DCCONSTRUCTIONS
e515ab1b8c
Preserve the completed teach-and-repeat laboratory stage: reference preparation, cascaded acquisition, local tracking and recovery, recording lifecycle, replay qualification, and persistent Rerun scene controls. Document the open grid-picking regression and Rerun upgrade contract. No autonomous driving or loop-closure optimization is claimed.
197 lines
11 KiB
Python
197 lines
11 KiB
Python
"""Project browsing binds to existing evidence and has no acquisition authority."""
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
import numpy as np
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from k1link.missions.projects import PlanningProjects
|
|
from k1link.web.mission_registration_api import build_mission_registration_router
|
|
|
|
|
|
def fixture(tmp_path):
|
|
draft = dict(id=str(uuid4()), revision=1, name='Reference experiment', updated_at_utc='2026-09-11T10:00:00Z',
|
|
zone=dict(label='A'), route=dict(length_m=30, points=[dict(position=[0,0,0]), dict(position=[30,0,0])]))
|
|
root=tmp_path/'recorded'; root.mkdir()
|
|
live_root=tmp_path/'live'; live_root.mkdir()
|
|
def read(identity): return json.loads((root/identity/'report.json').read_text())
|
|
runs=SimpleNamespace(root=root,get=read,directory=lambda identity:root/identity,
|
|
drafts=SimpleNamespace(list=lambda:[draft],get=lambda identity:draft))
|
|
live=SimpleNamespace(root=live_root,directory=lambda identity:live_root/identity)
|
|
return draft,runs,live,PlanningProjects(runs,live)
|
|
|
|
|
|
def save(owner,doc):
|
|
directory=owner.directory(doc['id']);directory.mkdir()
|
|
(directory/'report.json').write_text(json.dumps(doc))
|
|
return directory
|
|
|
|
|
|
def test_each_result_is_a_project_and_uses_frozen_draft(tmp_path):
|
|
draft,runs,live,projects=fixture(tmp_path)
|
|
ids=[str(uuid4()),str(uuid4())]
|
|
for n,identity in enumerate(ids):
|
|
save(runs,dict(id=identity,draft=dict(draft),state='ready',revision=1,created_at_utc=f'2026-09-11T10:0{n}:00Z',
|
|
scene_url=f'/recorded/{identity}.rrd',result=dict(status='candidate',matched_query_indices=[1,2])))
|
|
draft.update(name='Later edited name',revision=8)
|
|
items=projects.list()
|
|
assert len(items)==2 and items[0]['key']=='recorded:'+ids[1]
|
|
assert all(p['name']=='Reference experiment' and p['revision']==1 for p in items)
|
|
detail=projects.get('recorded',ids[0])
|
|
assert detail['draft']['revision']==1 and detail['scene_url']==f'/recorded/{ids[0]}.rrd'
|
|
assert 'matched_query_indices' not in detail['result']
|
|
assert not detail['vehicle_control'] and not detail['localization_confirmed']
|
|
|
|
|
|
def test_unstarted_draft_and_failed_live_are_honest_states(tmp_path):
|
|
draft,runs,live,projects=fixture(tmp_path)
|
|
assert projects.list()[0]['kind']=='draft'
|
|
identity=str(uuid4())
|
|
save(live,dict(id=identity,draft=draft,state='error',query_session_id='B',created_at_utc='2026-09-11T11:00:00Z',message='No fit',result=None))
|
|
assert len(projects.list())==1
|
|
detail=projects.get('live',identity)
|
|
assert detail['scene_url'] is None and detail['result'] is None and detail['message']=='No fit'
|
|
with pytest.raises(ValueError): projects.live_scene(identity)
|
|
|
|
|
|
def test_preparation_only_probe_is_not_a_passage_project(tmp_path):
|
|
draft,runs,live,projects=fixture(tmp_path)
|
|
save(live,dict(id=str(uuid4()),draft=draft,state='cancelled',query_session_id=None,created_at_utc='2026-09-11T11:00:00Z'))
|
|
assert [p['kind'] for p in projects.list()]==['draft']
|
|
|
|
|
|
def test_scene_hash_is_verified_and_browsing_router_never_starts_work(tmp_path):
|
|
draft,runs,live,projects=fixture(tmp_path)
|
|
identity=str(uuid4());payload=b'frozen-scene'
|
|
directory=save(runs,dict(id=identity,draft=draft,state='ready',created_at_utc='2026-09-11T11:00:00Z',
|
|
result={'status':'candidate'},artifacts={'scene.rrd':hashlib.sha256(payload).hexdigest()}))
|
|
(directory/'scene.rrd').write_bytes(payload)
|
|
app=FastAPI();app.include_router(build_mission_registration_router(runs,live))
|
|
with TestClient(app) as client:
|
|
assert client.get('/api/v1/mission-planner/projects').json()['items'][0]['key']=='recorded:'+identity
|
|
assert client.get('/api/v1/mission-planner/projects/recorded/'+identity).status_code==200
|
|
assert client.get('/api/v1/mission-planner/registration-runs/'+identity+'/scene.rrd').content==payload
|
|
(directory/'scene.rrd').write_bytes(b'different-scene')
|
|
assert client.get('/api/v1/mission-planner/registration-runs/'+identity+'/scene.rrd').status_code==409
|
|
assert client.get('/api/v1/mission-planner/projects/unknown/'+identity).status_code==404
|
|
|
|
|
|
def test_completed_live_uses_committed_fit_not_unregistered_terminal_preview(tmp_path,monkeypatch):
|
|
import k1link.missions.registration_scene as renderer
|
|
draft,runs,live,projects=fixture(tmp_path)
|
|
identity=str(uuid4());transform=np.eye(4);transform[0,3]=3
|
|
result={'status':'candidate','T_reference_query':transform.tolist(),'matched_query_indices':[0]}
|
|
doc=dict(id=identity,draft=draft,state='completed',created_at_utc='2026-09-11T11:00:00Z',query_session_id='B',
|
|
result=result,result_source_sequence=8,artifacts={})
|
|
directory=save(live,doc);step=directory/'step-001';step.mkdir()
|
|
(step/'source.json').write_text(json.dumps({'sequence':8,'query_path':[[1,2,3],[4,5,6]]}))
|
|
(step/'registration-result.json').write_text(json.dumps(result))
|
|
np.savez(step/'registration-input.npz',reference=np.zeros((2,3)),query=np.ones((2,3)),initial=np.eye(4))
|
|
for path in step.iterdir(): doc['artifacts'][str(path.relative_to(directory))]=hashlib.sha256(path.read_bytes()).hexdigest()
|
|
(directory/'report.json').write_text(json.dumps(doc))
|
|
before=(directory/'report.json').read_bytes()
|
|
captured=[]
|
|
def writer(path,run_id,reference,query,actual,ref_path,query_path):
|
|
captured.append((run_id,query.copy(),actual,query_path.copy()));path.write_bytes(b'scene')
|
|
monkeypatch.setattr(renderer,'write_scene',writer)
|
|
scene=projects.live_scene(identity)
|
|
assert scene.read_bytes()==b'scene' and len(captured)==1
|
|
assert captured[0][2]['T_reference_query']==transform.tolist()
|
|
np.testing.assert_equal(captured[0][1],np.ones((2,3)))
|
|
assert (directory/'report.json').read_bytes()==before
|
|
projects.live_scene(identity);assert len(captured)==1
|
|
(step/'registration-result.json').write_text('{}')
|
|
with pytest.raises(ValueError,match='целостности'): projects.live_scene(identity)
|
|
|
|
|
|
def test_spatial_renderer_transforms_query_cloud_and_path_once(tmp_path,monkeypatch):
|
|
import k1link.missions.registration_scene as renderer
|
|
logged={}
|
|
class Recording:
|
|
def __init__(self,*args,**kwargs): pass
|
|
def log(self,name,value,**kwargs): logged[name]=value
|
|
def save(self,*args): pass
|
|
def send_blueprint(self,*args): pass
|
|
def flush(self): pass
|
|
def disconnect(self): pass
|
|
monkeypatch.setattr(renderer.rr,'RecordingStream',Recording)
|
|
monkeypatch.setattr(renderer.rr,'Points3D',lambda xyz,**kw:np.asarray(xyz))
|
|
monkeypatch.setattr(renderer.rr,'LineStrips3D',lambda xyz,**kw:np.asarray(xyz))
|
|
t=np.eye(4);t[:3,3]=[3,-2,1]
|
|
points=np.array([[1.,2,3],[4,5,6]])
|
|
renderer.write_scene(tmp_path/'scene.rrd','test',points,points,{'status':'rejected','T_reference_query':t.tolist()},points,points)
|
|
np.testing.assert_equal(logged['world/reference'],points)
|
|
np.testing.assert_equal(logged['world/query'],points+[3,-2,1])
|
|
np.testing.assert_equal(logged['world/query_path'][0],points+[3,-2,1])
|
|
|
|
|
|
@pytest.mark.parametrize('kind,state', [('recorded', 'ready'), ('recorded', 'error'),
|
|
('live', 'completed'), ('live', 'cancelled'), ('live', 'error'), ('live', 'interrupted')])
|
|
def test_delete_one_project_preserves_evidence_and_survives_restart(tmp_path, kind, state):
|
|
draft, runs, live, projects = fixture(tmp_path)
|
|
owner = runs if kind == 'recorded' else live
|
|
ids = [str(uuid4()), str(uuid4())]
|
|
preserved = {}
|
|
for identity in ids:
|
|
directory = save(owner, dict(id=identity, draft=draft, state=state,
|
|
query_session_id='original-passage', created_at_utc='2026-09-11T11:00:00Z'))
|
|
(directory / 'scene.rrd').write_bytes(b'original-derived-scene')
|
|
for path in directory.iterdir(): preserved[path] = path.read_bytes()
|
|
key = kind + ':' + ids[0]
|
|
app = FastAPI(); app.include_router(build_mission_registration_router(runs, live))
|
|
with TestClient(app) as client:
|
|
url = '/api/v1/mission-planner/projects/' + kind + '/' + ids[0]
|
|
for _ in range(2):
|
|
reply = client.request('DELETE', url, json={'revision': 1})
|
|
assert reply.status_code == 200 and reply.json() == {'key': key, 'deleted': True}
|
|
assert client.get(url).status_code == 404
|
|
assert [item['key'] for item in client.get('/api/v1/mission-planner/projects').json()['items']] == [kind + ':' + ids[1]]
|
|
# Same-name sibling remains; the shared draft cannot resurface after the last deletion.
|
|
projects.remove(kind, ids[1], 1)
|
|
assert PlanningProjects(runs, live).list() == []
|
|
assert all(path.read_bytes() == data for path, data in preserved.items())
|
|
assert runs.drafts.get(draft['id']) == draft
|
|
|
|
|
|
def test_delete_draft_revision_and_project_kind_are_exact(tmp_path):
|
|
draft, runs, live, projects = fixture(tmp_path)
|
|
with pytest.raises(ValueError, match='изменён'):
|
|
projects.remove('draft', draft['id'], 9)
|
|
with pytest.raises(KeyError): projects.remove('unknown', draft['id'], 1)
|
|
assert len(projects.list()) == 1
|
|
projects.remove('draft', draft['id'], 1)
|
|
assert PlanningProjects(runs, live).list() == []
|
|
assert runs.drafts.get(draft['id']) == draft
|
|
# A later immutable run has an independent identity, not the draft tombstone.
|
|
save(runs, dict(id=draft['id'], draft=draft, state='ready', created_at_utc='2026-09-11T11:00:00Z'))
|
|
assert projects.list()[0]['kind'] == 'recorded'
|
|
|
|
|
|
@pytest.mark.parametrize('kind,state', [('recorded', 'queued'), ('recorded', 'running'),
|
|
('live', 'preparing'), ('live', 'waiting'), ('live', 'running'), ('live', 'unknown')])
|
|
def test_delete_fails_closed_for_active_or_unknown_states(tmp_path, kind, state):
|
|
draft, runs, live, projects = fixture(tmp_path)
|
|
identity = str(uuid4())
|
|
save(runs if kind == 'recorded' else live, dict(id=identity, draft=draft,
|
|
state=state, created_at_utc='2026-09-11T11:00:00Z'))
|
|
app = FastAPI(); app.include_router(build_mission_registration_router(runs, live))
|
|
with TestClient(app) as client:
|
|
url = '/api/v1/mission-planner/projects/' + kind + '/' + identity
|
|
assert client.request('DELETE', url, json={'revision': 1}).status_code == 409
|
|
assert client.request('DELETE', url, json={'revision': 1, 'delete_sources': True}).status_code == 422
|
|
assert client.request('DELETE', '/api/v1/mission-planner/projects/draft/' + draft['id'], json={'revision': 1}).status_code == 409
|
|
assert client.get(url).status_code == 200
|
|
assert len(projects.list()) == 1
|
|
|
|
|
|
def test_delete_api_rejects_missing_and_invalid_identity(tmp_path):
|
|
_, runs, live, _ = fixture(tmp_path)
|
|
app = FastAPI(); app.include_router(build_mission_registration_router(runs, live))
|
|
with TestClient(app) as client:
|
|
for kind in ['unknown', 'live']:
|
|
assert client.request('DELETE', f'/api/v1/mission-planner/projects/{kind}/{uuid4()}', json={'revision': 1}).status_code == 404
|
|
assert client.request('DELETE', '/api/v1/mission-planner/projects/live/not-a-uuid', json={'revision': 1}).status_code == 422
|