130 lines
6.6 KiB
Python
130 lines
6.6 KiB
Python
from types import SimpleNamespace
|
|
from pathlib import Path
|
|
import json
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from test_stream_summary import _write_capture, _pcl_payload, _pose_payload
|
|
from test_session_recording import _command
|
|
from k1link.device_plugins.xgrids_k1.planning_source import export_planning_source
|
|
from k1link.missions.sources import PlanningSources
|
|
from k1link.missions.drafts import MissionDrafts, DraftConflict, route_from_source
|
|
from k1link.web.mission_planner_api import DraftRequest, build_mission_planner_router
|
|
|
|
|
|
def source_doc():
|
|
return {'session_id': 'session-a', 'generation': 'a'*64, 'label': 'Route A', 'units': 'm',
|
|
'frame_id': 'session/session-a', 'source_digests': {'primary': 'b'*64}, 'decode_errors': 0,
|
|
'poses': [{'index': i, 'position': [i*3, i*4, 0]} for i in range(4)]}
|
|
|
|
|
|
class Sources:
|
|
changed = False
|
|
def bound(self, id, generation):
|
|
if self.changed or id != 'session-a' or generation != 'a'*64:
|
|
raise ValueError('source changed')
|
|
return source_doc()
|
|
verify = bound
|
|
|
|
|
|
def request(**values):
|
|
return DraftRequest(**dict({'name': 'Route', 'session_id': 'session-a', 'generation': 'a'*64,
|
|
'start_index': 0, 'end_index': 2}, **values))
|
|
|
|
|
|
def test_export_preserves_all_pose_indices_and_does_not_transform_twice(tmp_path):
|
|
src = tmp_path / 'mqtt.raw.k1mqtt'
|
|
_write_capture(src, [('lixel/application/report/lio_pcl', _pcl_payload(scaler=1000, point_count=4)),
|
|
*[('lixel/application/report/lio_pose', _pose_payload(xyz)) for xyz in [(0, 0, 0), (3, 4, 0), (0, 0, 0)]]])
|
|
path = tmp_path / 'planning.json'
|
|
export_planning_source(src, path)
|
|
doc = json.loads(path.read_text())
|
|
assert [p['index'] for p in doc['poses']] == [0, 1, 2]
|
|
assert doc['poses'][1]['position'] == [3, 4, 0]
|
|
assert doc['poses'][1]['distance_m'] == 5
|
|
assert doc['path_m'] == 10
|
|
assert all(p['elapsed_s'] is None for p in doc['poses'])
|
|
|
|
|
|
def test_export_rejects_camera_or_pose_only_recording(tmp_path):
|
|
src = tmp_path / 'mqtt.raw.k1mqtt'
|
|
_write_capture(src, [('lixel/application/report/lio_pose', _pose_payload((0, 0, 0)))] * 2)
|
|
with pytest.raises(ValueError): export_planning_source(src, tmp_path / 'out.json')
|
|
|
|
|
|
def test_draft_persists_full_route_and_optimistic_revision(tmp_path):
|
|
service = MissionDrafts(tmp_path, Sources())
|
|
first = service.save(request(direction='reverse'))
|
|
assert first['vehicle_id'] is None and first['revision'] == 1
|
|
assert [p['source_index'] for p in first['route']['points']] == [2, 1, 0]
|
|
assert first['route']['length_m'] == 10
|
|
restarted = MissionDrafts(tmp_path, Sources())
|
|
assert restarted.get(first['id']) == first
|
|
next = restarted.save(request(id=first['id'], revision=1, name='Changed'))
|
|
assert next['revision'] == 2
|
|
with pytest.raises(DraftConflict): service.save(request(id=first['id'], revision=1))
|
|
assert service.get(first['id']) == next
|
|
|
|
|
|
def test_whole_recording_is_resolved_by_server_and_old_report_stays_frozen(tmp_path):
|
|
service = MissionDrafts(tmp_path, Sources())
|
|
old = service.save(request())
|
|
full = service.save(request(id=old['id'], revision=1, whole_recording=True,
|
|
start_index=1, end_index=2))
|
|
assert (full['route']['start_index'], full['route']['end_index']) == (0, 3)
|
|
assert full['route']['length_m'] == 15
|
|
assert old['route']['length_m'] == 10
|
|
app = FastAPI()
|
|
app.include_router(build_mission_planner_router(service))
|
|
with TestClient(app) as client:
|
|
body = {'name': 'Full', 'session_id': 'session-a', 'generation': 'a'*64,
|
|
'whole_recording': True, 'direction': 'reverse'}
|
|
response = client.post('/api/v1/mission-planner/drafts', json=body)
|
|
assert response.status_code == 200
|
|
assert [p['source_index'] for p in response.json()['route']['points']] == [3, 2, 1, 0]
|
|
assert client.post('/api/v1/mission-planner/drafts', json={**body, 'whole_recording': False}).status_code == 409
|
|
|
|
|
|
@pytest.mark.parametrize('start,end', [(2, 2), (3, 1), (-1, 2), (0, 4)])
|
|
def test_route_rejects_out_of_bounds(start, end):
|
|
with pytest.raises(ValueError): route_from_source(source_doc(), start, end, 'forward')
|
|
|
|
|
|
def test_check_is_saved_revision_bound_and_never_claims_localization(tmp_path):
|
|
sources = Sources(); service = MissionDrafts(tmp_path, sources)
|
|
draft = service.save(request())
|
|
result = service.check(draft['id'], 1)
|
|
assert result['localization'] == 'not_run' and result['vehicle_control'] is False
|
|
assert result['warnings'] and result['source_verified']
|
|
with service.connect() as db: assert db.execute('SELECT COUNT(*) FROM checks').fetchone()[0] == 1
|
|
sources.changed = True
|
|
with pytest.raises(ValueError): service.check(draft['id'], 1)
|
|
assert service.get(draft['id']) == draft # unavailable evidence never discards the draft
|
|
|
|
|
|
def test_api_forbids_vehicle_authority_and_checks_revision(tmp_path):
|
|
app = FastAPI(); app.include_router(build_mission_planner_router(MissionDrafts(tmp_path, Sources())))
|
|
with TestClient(app) as client:
|
|
body = request().model_dump(mode='json')
|
|
assert client.post('/api/v1/mission-planner/drafts', json={**body, 'vehicle_id': 'rover'}).status_code == 422
|
|
draft = client.post('/api/v1/mission-planner/drafts', json=body).json()
|
|
assert client.post('/api/v1/mission-planner/drafts/'+draft['id']+'/checks', json={'revision': 2}).status_code == 409
|
|
assert client.get('/api/v1/mission-planner/drafts/'+draft['id']).json() == draft
|
|
|
|
|
|
def test_source_cache_is_bound_to_validated_archive_and_rejects_lab(tmp_path):
|
|
command = _command(tmp_path / 'source')
|
|
detail = SimpleNamespace(plugin_id=command.plugin_id, summary=SimpleNamespace(replayable=True, lab=None), as_dict=lambda: {'display_name': 'A'})
|
|
store = SimpleNamespace(data_dir=tmp_path / 'data', get_session=lambda _: detail, prepare_replay=lambda _: command)
|
|
count = []
|
|
def export(source, dest):
|
|
count.append(1); dest.write_text(json.dumps({'poses': [], 'path_m': 0}))
|
|
service = PlanningSources(store, {command.plugin_id: export})
|
|
first = service.get(command.session_id)
|
|
assert service.get(command.session_id) == first and len(count) == 1
|
|
assert service.verify(command.session_id, first['generation']) == first
|
|
command.primary_artifact.path.write_bytes(b'X' * command.primary_artifact.file_byte_length)
|
|
with pytest.raises(ValueError): service.bound(command.session_id, first['generation'])
|
|
detail.summary.lab = object()
|
|
with pytest.raises(ValueError): service.get(command.session_id)
|