m4
5a581c78a2
Build / build (push) Has been cancelled
Docker / build (push) Has been cancelled
Lint / ruff (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.11) (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.12) (push) Has been cancelled
Test / pytest (windows-latest, 3.11) (push) Has been cancelled
Test / pytest (windows-latest, 3.12) (push) Has been cancelled
Introduce provider, model, and invocation contracts with encrypted configuration persistence. Add web runtime fencing, route fallback, recovery middleware, workspace scoping, and comprehensive tests.
65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
|
|
from EvoScientist.llm.contracts import EvoRuntimeError, HmacGrantAuthority
|
|
from EvoScientist.llm.crypto import canonical_json_v1
|
|
from EvoScientist.sessions import FencedPruningCheckpointer
|
|
from tests.v3_fixtures import RUNTIME_KEY_ID, RUNTIME_SECRET
|
|
|
|
|
|
def test_canonical_json_normalizes_unicode_and_orders_keys():
|
|
assert canonical_json_v1({"z": "e\u0301", "a": 1}) == canonical_json_v1(
|
|
{"a": 1, "z": "\u00e9"}
|
|
)
|
|
|
|
|
|
def test_previous_runtime_key_verifies_during_rotation():
|
|
old_secret = "old-runtime-secret-for-tests-000000000000000000000"
|
|
old = HmacGrantAuthority(old_secret, "old-key")
|
|
grant = old.sign_subject(
|
|
subject_id="user", plan="starter", roles=("user",), ttl_ms=60_000
|
|
)
|
|
rotated = HmacGrantAuthority(
|
|
RUNTIME_SECRET,
|
|
RUNTIME_KEY_ID,
|
|
previous_secret=old_secret,
|
|
previous_key_id="old-key",
|
|
)
|
|
assert rotated.verify_subject(grant)
|
|
with pytest.raises(EvoRuntimeError, match="CONTRACT_SIGNATURE_INVALID"):
|
|
rotated.require_admin(replace(grant, audience="evoscientist-runtime"))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stale_turn_token_cannot_write_or_release_new_lease(tmp_path):
|
|
database = tmp_path / "sessions.sqlite"
|
|
async with FencedPruningCheckpointer.from_conn_string_with_keep(
|
|
str(database), keep_per_ns=10
|
|
) as saver:
|
|
first = await saver.acquire_turn_lease(
|
|
"web:user:thread", "owner-1", ttl_seconds=30
|
|
)
|
|
assert await saver.release_turn_lease(first)
|
|
current = await saver.acquire_turn_lease(
|
|
"web:user:thread", "owner-2", ttl_seconds=30
|
|
)
|
|
assert current.fencing_token == first.fencing_token + 1
|
|
assert not await saver.release_turn_lease(first)
|
|
with pytest.raises(RuntimeError, match="TURN_FENCED"):
|
|
await saver.aput_writes(
|
|
{
|
|
"configurable": {
|
|
"thread_id": first.thread_id,
|
|
"checkpoint_id": "checkpoint",
|
|
"turn_lease_owner": first.owner_id,
|
|
"turn_fencing_token": first.fencing_token,
|
|
}
|
|
},
|
|
[("channel", "value")],
|
|
"task",
|
|
)
|
|
assert await saver.release_turn_lease(current)
|