ive harnessed the harness
Something went wrong. Try again.
10 kB · 201 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202from __future__ import annotations
import asyncioimport inspectimport jsonimport sqlite3import structimport tempfileimport unittestfrom pathlib import Path
from klbr.hooks import AttentionDecision, DeliveryDecision, describefrom klbr_runtime.release import fingerprintfrom klbr_runtime.wire import MAX_FRAME, ProtocolError, encode, read_framefrom peer import ROOT, behavior_copy
class FrameTests(unittest.IsolatedAsyncioTestCase): async def read(self,data): reader=asyncio.StreamReader();reader.feed_data(data);reader.feed_eof() return await read_frame(reader)
async def test_length_prefix_roundtrip_unicode(self): value={"test":"hello 🌙"} frame=encode(value) self.assertEqual(struct.unpack('>I',frame[:4])[0],len(frame)-4) self.assertEqual(await self.read(frame),value)
async def test_truncated_frames_fail(self): for frame in [b'\x00\x00',struct.pack('>I',20)+b'{}']: with self.assertRaises(asyncio.IncompleteReadError): await self.read(frame)
async def test_oversize_rejected_before_payload_read(self): with self.assertRaises(ProtocolError): await self.read(struct.pack('>I',MAX_FRAME+1)) with self.assertRaises(ProtocolError): encode({"large":"x"*MAX_FRAME})
async def test_duplicate_keys_and_nonfinite_values_rejected(self): for raw in [b'{"a":1,"a":2}',b'{"x":NaN}',b'{"x":Infinity}',b'\xff',b'{']: with self.subTest(raw=raw),self.assertRaises(ProtocolError): await self.read(struct.pack('>I',len(raw))+raw)
async def test_shared_golden_frames(self): fixtures=json.loads((ROOT/"tests/runtime-fixtures/envelopes.json").read_text()) for fixture in fixtures: with self.subTest(kind=fixture["message"]["kind"]): self.assertEqual(await self.read(encode(fixture)),fixture)
class ValueTests(unittest.TestCase): def test_decision_shapes_are_explicit(self): # A mistyped public SDK slot must fail locally, not kill a workbench by # sending an invalid typed wire variant to the host. with self.assertRaises(ValueError): asyncio.run(describe("misspelled.slot")) self.assertEqual(DeliveryDecision.allow().to_wire(),{"action":"allow"}) self.assertEqual(AttentionDecision.defer(8).to_wire(),{"action":"defer","seconds":8}) for decision in [DeliveryDecision("allow","invalid"),DeliveryDecision.hold(""), AttentionDecision.defer(float("nan")),AttentionDecision.defer(-1)]: with self.subTest(decision=decision),self.assertRaises(ValueError): decision.to_wire()
def test_behavior_identity_changes_with_source(self): with tempfile.TemporaryDirectory() as directory: root=behavior_copy(Path(directory)/"candidate") original=fingerprint(root) self.assertEqual(original,fingerprint(ROOT/"behavior")) with (root/"klbr_hooks/defaults.py").open("a") as file: file.write("\n# a change\n") self.assertNotEqual(original,fingerprint(root))
def test_behavior_symlinks_rejected(self): with tempfile.TemporaryDirectory() as directory: root=behavior_copy(Path(directory)/"candidate") (root/"outside.py").symlink_to('/etc/passwd') with self.assertRaises(ValueError): fingerprint(root)
def test_behavior_root_symlinks_rejected(self): with tempfile.TemporaryDirectory() as directory: root=Path(directory)/"alias" root.symlink_to(ROOT/"behavior",target_is_directory=True) with self.assertRaises(ValueError): fingerprint(root)
def test_oversized_behavior_file_rejected(self): with tempfile.TemporaryDirectory() as directory: root=behavior_copy(Path(directory)/"candidate") with (root/"oversized.py").open("wb") as file: file.truncate(2_097_153) with self.assertRaises(ValueError): fingerprint(root)
def test_shared_behavior_digest(self): fixture=json.loads((ROOT/"tests/runtime-fixtures/behavior.json").read_text()) self.assertEqual(fingerprint(ROOT/"behavior"),fixture["sha256"])
class SchemaTests(unittest.TestCase): """Executes the exact shipped migration, not Rust Store methods.
Python's linked SQLite is only used for these serial schema tests. The Rust runtime separately enforces the documented WAL-reset fixed-version floor. """ def setUp(self): self.db=sqlite3.connect(':memory:') self.db.execute('PRAGMA foreign_keys=ON') self.db.executescript((ROOT/'klbr-runtime/migrations/001_runtime.sql').read_text()) self.db.execute("INSERT INTO sessions(id) VALUES('root')") self.db.execute("INSERT INTO sessions(id) VALUES('child')") self.db.execute("INSERT INTO releases(id,path) VALUES('revision','/test')") self.db.commit()
def tearDown(self): self.db.close()
def event(self,key=None,session='root',kind='input'): return self.db.execute("INSERT INTO events(session_id,kind,source_key,payload,created_ms) VALUES(?,?,?,?,0)", (session,kind,key,'{}')).lastrowid
def execution(self,identity='cell',session='root'): event=self.event(session=session,kind='execution.started') self.db.execute("INSERT INTO executions(id,session_id,generation,hook_revision,state,started_event) VALUES(?,?,'g','revision','running',?)", (identity,session,event))
def test_schema_identity(self): self.assertEqual(self.db.execute('PRAGMA user_version').fetchone()[0],1) self.assertEqual(self.db.execute('PRAGMA application_id').fetchone()[0],1263288882)
def test_originals_cannot_be_updated_or_deleted(self): identity=self.event() for sql in ['UPDATE events SET created_ms=1 WHERE seq=?','DELETE FROM events WHERE seq=?']: with self.assertRaises(sqlite3.IntegrityError): self.db.execute(sql,(identity,))
def test_source_identity_is_unique_per_session(self): self.event('source-1') with self.assertRaises(sqlite3.IntegrityError): self.event('source-1') self.event('source-1',session='child')
def test_burst_has_one_inbox_row_per_input(self): for i in range(500): event=self.event(f'input-{i}') self.db.execute('INSERT INTO inbox(event_id) VALUES(?)',(event,)) self.assertEqual(self.db.execute("SELECT count(*) FROM inbox WHERE status='pending'").fetchone()[0],500)
def test_foreign_keys_prevent_orphaned_inbox(self): with self.assertRaises(sqlite3.IntegrityError): self.db.execute('INSERT INTO inbox(event_id) VALUES(999)')
def test_unfinished_execution_unique_with_independent_sessions(self): self.execution() with self.assertRaises(sqlite3.IntegrityError): self.execution('second') self.execution('child-cell',session='child')
def test_yield_does_not_open_a_second_foreground_slot(self): self.execution() self.db.execute("UPDATE executions SET state='yielded' WHERE id='cell'") with self.assertRaises(sqlite3.IntegrityError): self.execution('second')
def test_confirmed_effect_requires_receipt(self): self.execution() with self.assertRaises(sqlite3.IntegrityError): self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES('e','cell','r','hello','confirmed')")
def test_publication_transaction_rolls_back_receipt_and_event_together(self): self.execution();self.db.commit() before=self.db.execute('SELECT count(*) FROM events').fetchone()[0] self.db.execute('BEGIN IMMEDIATE') event=self.event(kind='local.sent') self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status,event_id) VALUES('e','cell','r','hello','confirmed',?)",(event,)) self.db.rollback() self.assertEqual(self.db.execute('SELECT count(*) FROM effects').fetchone()[0],0) self.assertEqual(self.db.execute('SELECT count(*) FROM events').fetchone()[0],before)
def test_dedup_is_by_request_not_message_text(self): self.execution() for identity in ['a','b']: self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES(?,'cell',?,'same','proposed')",(identity,identity)) with self.assertRaises(sqlite3.IntegrityError): self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES('c','cell','a','other','proposed')") self.assertEqual(self.db.execute('SELECT count(*) FROM effects').fetchone()[0],2)
def test_impossible_ready_with_deadline_is_rejected(self): with self.assertRaises(sqlite3.IntegrityError): self.db.execute("UPDATE sessions SET state='ready',wake_at_ms=100 WHERE id='root'")
class PeerSurfaceTests(unittest.TestCase): """The peer surface names its subject once; the duplicated functional layer stays gone."""
def test_no_callable_repeats_peer_id(self): import klbr.peer as peer for name in ["send", "get", "stop"]: signature = inspect.signature(getattr(peer.PeerSession, name)) self.assertNotIn("peer_id", signature.parameters, name) for name in ["result", "cancel"]: signature = inspect.signature(getattr(peer.PeerRequestHandle, name)) self.assertNotIn("peer_id", signature.parameters, name)
def test_subjectless_verbs_are_the_whole_module_surface(self): import klbr.peer as peer self.assertEqual(sorted(peer.__all__), ["PeerRequestHandle", "PeerSession", "answer", "open", "peers"]) for gone in ["send", "result", "cancel", "get", "stop", "list_peers", "submit_result"]: self.assertFalse(hasattr(peer, gone), f"peer.{gone} repeats a subject the handle already carries")
def test_the_agents_alias_vocabulary_is_gone(self): with self.assertRaises(ModuleNotFoundError): import klbr.agents # noqa: F401