102 lines
8.7 KiB
Python
102 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
from yovision_brain.integration.machine_identity.token import KeyRecord, Registry, Signer, Verifier
|
|
from yovision_brain.integration.sense_control.connector import ConnectorResponse, SourceConfigEndpoint, StatusSender
|
|
from yovision_brain.integration.sense_control.consumer import SourceConfigConsumer, SourceConfigError
|
|
from yovision_brain.integration.sense_control.replay import SQLiteReplayStore
|
|
from yovision_brain.integration.sense_control.status import RuntimeStatusPublisher
|
|
|
|
|
|
def source_document(revision: int = 1, *, state: str = "active", effective: int = 0) -> bytes:
|
|
now=datetime(2026,8,31,tzinfo=timezone.utc)
|
|
document={"schema_version":"yovision.source-config/v1","config_id":"gate-primary","revision":revision,"published_at":now.isoformat().replace("+00:00","Z"),"effective_at":(now+timedelta(seconds=effective)).isoformat().replace("+00:00","Z"),"site":{"id":"site-east"},"logical_device":{"id":"camera-1"},"profile":{"id":"main","width":1920,"height":1080,"encoding":"H264","frame_rate":25},"media":{"ref":"media:site-east/camera-1/main","transport":"rtsp"},"rule_set":{"version":f"rules-{revision}","state":state,"profile_binding":{"profile_id":"main","width":1920,"height":1080},"areas":[{"id":"danger","version":1,"kind":"danger_area","enabled":True,"points":[{"x":.1,"y":.1},{"x":.8,"y":.1},{"x":.5,"y":.8}]}],"directional_lines":[]}}
|
|
digest=hashlib.sha256(json.dumps(document,separators=(",",":"),sort_keys=True).encode()).hexdigest();document["integrity"]={"algorithm":"sha256","value":digest}
|
|
return json.dumps(document,separators=(",",":"),sort_keys=True).encode()
|
|
|
|
|
|
def with_extension(body: bytes, extensions: object) -> bytes:
|
|
document=json.loads(body);document["extensions"]=extensions;unsigned=dict(document);unsigned.pop("integrity");document["integrity"]["value"]=hashlib.sha256(json.dumps(unsigned,separators=(",",":"),sort_keys=True).encode()).hexdigest();return json.dumps(document,separators=(",",":"),sort_keys=True).encode()
|
|
|
|
|
|
def identity(tmp_path, now: int):
|
|
private=Ed25519PrivateKey.generate();signer=Signer("yv:sense:east","sense-key-01",private,clock=lambda:now)
|
|
registry=Registry([KeyRecord("yv:sense:east","sense-key-01",private.public_key(),"yovision-brain",frozenset({"source-config:write"}))])
|
|
replay=SQLiteReplayStore(tmp_path/"replay.sqlite")
|
|
return signer,Verifier(registry,replay,clock=lambda:now)
|
|
|
|
|
|
def test_authenticated_apply_is_idempotent_and_replay_survives_restart(tmp_path):
|
|
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now);endpoint=SourceConfigEndpoint(consumer,verifier)
|
|
token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
|
|
result=endpoint.receive("Bearer "+token,body,"corr-request-0001");assert result.state=="applied" and result.config.rules is not None
|
|
with pytest.raises(ValueError,match="machine_token_replayed"):endpoint.receive("Bearer "+token,body,"corr-request-0001")
|
|
restarted=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),Verifier(verifier._registry,SQLiteReplayStore(tmp_path/"replay.sqlite"),clock=lambda:now))
|
|
with pytest.raises(ValueError,match="machine_token_replayed"):restarted.receive("Bearer "+token,body,"corr-request-0002")
|
|
new_token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body);assert restarted.receive("Bearer "+new_token,body,"corr-request-0003").state=="idempotent"
|
|
|
|
|
|
def test_atomic_replay_accepts_once_under_concurrency(tmp_path):
|
|
store=SQLiteReplayStore(tmp_path/"atomic.sqlite");results=[]
|
|
threads=[threading.Thread(target=lambda:results.append(store.consume("yv:sense:east","token-id",200,100))) for _ in range(12)]
|
|
for thread in threads:thread.start()
|
|
for thread in threads:thread.join()
|
|
assert results.count(True)==1
|
|
|
|
|
|
def test_stale_unknown_profile_and_recalibration_are_safe(tmp_path):
|
|
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
|
|
assert consumer.apply(source_document(2)).config.rules is not None
|
|
with pytest.raises(SourceConfigError,match="STALE_REVISION"):consumer.apply(source_document(1))
|
|
invalid=json.loads(source_document(3));invalid["profile"]["width"]=1280;invalid["integrity"]["value"]="0"*64
|
|
with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(json.dumps(invalid).encode())
|
|
unknown=json.loads(source_document(3));unknown["schema_version"]="yovision.source-config/v2"
|
|
with pytest.raises(SourceConfigError,match="UNSUPPORTED_SCHEMA_VERSION"):consumer.apply(json.dumps(unknown).encode())
|
|
safe=consumer.apply(source_document(3,state="recalibration_required"));assert safe.config.rule_state=="recalibration_required" and safe.config.rules is None
|
|
|
|
|
|
def test_future_effective_snapshot_activates_atomically_after_restart(tmp_path):
|
|
base=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());current=[base];path=tmp_path/"state.sqlite";consumer=SourceConfigConsumer(path,clock=lambda:current[0])
|
|
assert consumer.apply(source_document(1)).state=="applied";scheduled=consumer.apply(source_document(2,effective=60));assert scheduled.state=="scheduled" and scheduled.config.revision==1
|
|
current[0]+=61;restarted=SourceConfigConsumer(path,clock=lambda:current[0]);activated=restarted.activate_due();assert activated[0].revision==2 and restarted.get_active("gate-primary").revision==2
|
|
|
|
|
|
def test_unknown_valid_extension_namespace_is_ignored(tmp_path):
|
|
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
|
|
result=consumer.apply(with_extension(source_document(),{"vendor.example":{"feature":"safe"}}));assert result.state=="applied" and result.config.revision==1
|
|
|
|
|
|
@pytest.mark.parametrize("extensions", [[], {"1invalid":{}}, {"vendor_ok":{}}, {"vendor.example":"not-an-object"}])
|
|
def test_invalid_extensions_are_rejected(tmp_path, extensions):
|
|
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
|
|
with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(with_extension(source_document(),extensions))
|
|
|
|
|
|
def test_status_sequence_restart_retry_timeout_and_disable(tmp_path):
|
|
path=tmp_path/"status.sqlite";publisher=RuntimeStatusPublisher(path,"brain-east-01");health={"overall":"healthy","error_codes":[],"metrics":{"load_percent":1.0,"queue_depth":0,"latency_ms":2.0}}
|
|
one=json.loads(publisher.build(runtime_state="running",runtime_version="1.0.0",started_at=datetime.now(timezone.utc),model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));two=json.loads(RuntimeStatusPublisher(path,"brain-east-01").build(runtime_state="running",runtime_version="1.0.0",started_at=None,model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));assert (one["sequence"],two["sequence"])==(0,1)
|
|
private=Ed25519PrivateKey.generate();signer=Signer("yv:brain:east","brain-key-01",private,clock=lambda:1_787_000_000);attempts=[]
|
|
def send(_auth,_body,_corr,timeout):attempts.append(timeout);raise TimeoutError
|
|
sender=StatusSender(signer,send,max_attempts=3,sleeper=lambda _:None)
|
|
with pytest.raises(RuntimeError,match="STATUS_DELIVERY_EXHAUSTED"):sender.publish(b"{}","corr-request-0001")
|
|
assert attempts==[5.0,5.0,5.0]
|
|
disabled=StatusSender(signer,lambda *_:ConnectorResponse(204,b"","corr-request-0001"),enabled=False)
|
|
with pytest.raises(RuntimeError,match="CONNECTOR_DISABLED"):disabled.publish(b"{}","corr-request-0001")
|
|
|
|
|
|
@pytest.mark.parametrize("request_id", ["short", "0123456789abcde\n", "0123456789abcde!", "a"*129])
|
|
def test_connector_rejects_unsafe_request_ids(tmp_path, request_id):
|
|
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();endpoint=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),verifier)
|
|
token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
|
|
with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):endpoint.receive("Bearer "+token,body,request_id)
|
|
status_signer=Signer("yv:brain:east","brain-key-01",Ed25519PrivateKey.generate(),clock=lambda:now)
|
|
sender=StatusSender(status_signer,lambda *_:ConnectorResponse(204,b"",request_id))
|
|
with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):sender.publish(b"{}",request_id)
|