80 lines
3.5 KiB
Python
80 lines
3.5 KiB
Python
"""Apparatus metadata is independent of its powered/offline board."""
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from test_pairing import setup, create, cert, heartbeat
|
|
from k1link.fleet.registry import FleetRegistry
|
|
from k1link.fleet.trust import PairingError
|
|
from k1link.web.fleet_api import router
|
|
|
|
|
|
def test_offline_rename_persists_without_touching_board_or_authority(setup):
|
|
fleet, _, _, calls = setup
|
|
public, _ = create(setup)
|
|
fleet.advance(public['id'])
|
|
before = fleet.find(public['id'])
|
|
network_calls = len(calls)
|
|
result = fleet.rename(before['id'], ' New rover ', before['revision'])
|
|
expected = {**before, 'name': 'New rover', 'revision': before['revision']+1}
|
|
assert fleet.find(before['id']) == expected
|
|
assert result['connectivity'] == 'offline'
|
|
assert len(calls) == network_calls
|
|
assert fleet.rename(before['id'], 'New rover', before['revision']) == result
|
|
reopened = FleetRegistry(fleet.root)
|
|
try:
|
|
assert reopened.find(before['id']) == expected
|
|
finally:
|
|
reopened.close()
|
|
# The existing board certificate and heartbeat continue to work unchanged.
|
|
assert fleet.receive(cert(before), '/v1/node/heartbeat', heartbeat(before))[0] == 200
|
|
assert fleet.find(before['id'])['name'] == 'New rover'
|
|
|
|
|
|
def test_stale_editor_cannot_overwrite_newer_name(setup):
|
|
fleet = setup[0]
|
|
row, _ = create(setup)
|
|
fleet.rename(row['id'], 'Other editor', row['revision'])
|
|
with pytest.raises(PairingError, match='изменился'):
|
|
fleet.rename(row['id'], 'Stale edit', row['revision'])
|
|
assert fleet.find(row['id'])['name'] == 'Other editor'
|
|
|
|
|
|
@pytest.mark.parametrize('name', ['', ' ', 'a'*81, 'bad\nname', 'bad\x7fname'])
|
|
def test_invalid_name_does_not_change_saved_apparatus(setup, name):
|
|
fleet = setup[0]
|
|
row, _ = create(setup)
|
|
before = fleet.find(row['id'])
|
|
with pytest.raises(PairingError):
|
|
fleet.rename(row['id'], name, row['revision'])
|
|
assert fleet.find(row['id']) == before
|
|
|
|
|
|
def test_database_failure_keeps_previous_name(setup):
|
|
fleet = setup[0]
|
|
row, _ = create(setup)
|
|
fleet.db.execute("CREATE TRIGGER fail_rename BEFORE UPDATE ON vehicles BEGIN SELECT RAISE(ABORT, 'synthetic failure'); END")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
fleet.rename(row['id'], 'Unsaved', row['revision'])
|
|
assert fleet.find(row['id'])['name'] == row['name']
|
|
|
|
|
|
def test_rename_endpoint_is_local_scoped_and_returns_public_metadata(setup):
|
|
fleet = setup[0]
|
|
row, _ = create(setup)
|
|
app = FastAPI(); app.state.fleet = fleet; app.include_router(router)
|
|
path = '/api/v1/fleet/'+row['id']
|
|
body = {'name':'Updated rover','expected_revision':row['revision']}
|
|
with TestClient(app, base_url='http://127.0.0.1:8000', client=('127.0.0.1',55555)) as client:
|
|
assert client.patch(path, json=body, headers={'Origin':'https://attacker.test'}).status_code == 403
|
|
assert client.patch(path, json={**body,'node_id':'injected'}).status_code == 422
|
|
assert client.patch(path, json={**body,'expected_revision':True}).status_code == 422
|
|
response = client.patch(path, json=body)
|
|
assert response.status_code == 200
|
|
assert response.headers['cache-control'] == 'no-store'
|
|
assert response.json()['name'] == 'Updated rover'
|
|
assert fleet.find(row['id'])['binding']['binding_id'] not in response.text
|
|
assert client.patch(path, json={**body,'name':'Stale'}).status_code == 409
|