Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164"""The kernel half of the session layer, exercised frame by frame.
Sequence, acknowledgement, and replay rules decide whether a reconnect losesor repeats a message, and E2E cannot choose where a connection drops. Thesetests drive albedo_link directly: the outbox's coalescing and bounds, replayeddaemon frames, token checks, newest-attach-wins, and resending what the daemonnever acknowledged."""
from pathlib import Pathimport osimport shutilimport socketimport sysimport tempfileimport threadingimport unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "priv" / "python"))import albedo_link # noqa: E402
class OutboxTest(unittest.TestCase): def test_ack_drops_everything_up_to_it(self): outbox = albedo_link.Outbox() for n in range(4): outbox.add({"type": "call", "id": str(n)}) outbox.ack(2) self.assertEqual([seq for seq, _ in outbox.pending()], [3, 4])
def test_mirrors_and_traces_keep_only_their_newest_copy(self): outbox = albedo_link.Outbox() outbox.add({"type": "mirror", "handle": "a", "tail": "1"}) outbox.add({"type": "mirror", "handle": "b", "tail": "1"}) outbox.add({"type": "done", "id": "c1"}) outbox.add({"type": "mirror", "handle": "a", "tail": "2"}) frames = [data for _, data in outbox.pending()] self.assertEqual(len(frames), 3) self.assertIn(b'"tail":"2"', frames[-1]) self.assertNotIn(b'"handle":"a","tail":"1"', b"".join(frames))
def test_the_bound_sheds_coalescing_frames_before_deliveries(self): outbox = albedo_link.Outbox(frames=3) outbox.add({"type": "done", "id": "c1"}) outbox.add({"type": "mirror", "handle": "a"}) outbox.add({"type": "call", "id": "x"}) outbox.add({"type": "job", "id": "j"}) kinds = [data for _, data in outbox.pending()] self.assertEqual(len(kinds), 3) self.assertFalse(any(b"mirror" in data for data in kinds)) self.assertEqual(outbox.dropped, 0) outbox.add({"type": "done", "id": "c2"}) self.assertEqual(outbox.dropped, 1) self.assertNotIn(b'"c1"', b"".join(data for _, data in outbox.pending()))
def test_the_bound_sheds_the_oldest_coalescing_frame_first(self): outbox = albedo_link.Outbox(frames=3) outbox.add({"type": "mirror", "handle": "a"}) outbox.add({"type": "mirror", "handle": "b"}) outbox.add({"type": "done", "id": "c1"}) outbox.add({"type": "mirror", "handle": "c"}) frames = b"".join(data for _, data in outbox.pending()) self.assertNotIn(b'"handle":"a"', frames) self.assertIn(b'"handle":"b"', frames) self.assertIn(b'"handle":"c"', frames) self.assertEqual(outbox.dropped, 0)
def test_a_replayed_daemon_frame_is_not_applied_twice(self): inbound = albedo_link.Inbound() self.assertEqual( [inbound.accept(seq) for seq in (1, 2, 2, 1, 3)], [True, True, False, False, True], )
def test_job_book_follows_started_and_proven_gone_jobs(self): book = albedo_link.JobBook() book.observe({"type": "job_start", "id": "a", "pgid": 4242}) book.observe({"type": "job_start", "id": "b", "pgid": 4243}) book.observe({"type": "job", "id": "a", "cleanup": {"gone": False}}) book.observe({"type": "job", "id": "b", "cleanup": {"gone": True}}) book.observe({"type": "jobs", "live": 2}) self.assertEqual(sorted(book.started), ["a"]) self.assertEqual(book.live(), 3)
class SocketLinkTest(unittest.TestCase): def setUp(self): self.run_dir = tempfile.mkdtemp(prefix="link-") self.received = [] self.link = albedo_link.SocketLink( self.run_dir, "secret", lambda: {"pid": os.getpid()}, bundle="bundle", grace=60, ) threading.Thread(target=self.serve, daemon=True).start()
def serve(self): try: self.link.serve(self.received.append) except OSError: pass # the listener closed in tearDown
def tearDown(self): self.link.listener.close() shutil.rmtree(self.run_dir, ignore_errors=True)
def attach(self, token="secret", ack=0): connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) connection.settimeout(5) connection.connect(os.path.join(self.run_dir, albedo_link.SOCKET)) self.addCleanup(connection.close) attach = {"attach": {"token": token, "ack": ack, "grace": 30}} albedo_link.write_frame(connection.send, albedo_link.encode(attach)) return connection, albedo_link.read_frame(connection.recv)
def frames(self, connection, count): return [albedo_link.read_frame(connection.recv) for _ in range(count)]
def test_a_wrong_token_is_refused(self): _, answer = self.attach(token="guess") self.assertEqual(answer, {"refused": "wrong token"}) self.assertFalse(self.link.attached())
def test_reattach_resends_only_what_was_not_acknowledged(self): first, hello = self.attach() self.assertEqual((hello["hello"]["epoch"], hello["hello"]["ack"]), (1, 0)) self.assertEqual(self.link.grace, 30) for n in range(3): self.link.send({"type": "call", "id": str(n)}) self.assertEqual([f["seq"] for f in self.frames(first, 3)], [1, 2, 3]) albedo_link.write_frame( first.send, albedo_link.encode( {"seq": 1, "ack": 1, "frame": {"type": "release", "handle": "h"}} ), ) self.assertEqual(albedo_link.read_frame(first.recv), {"ack": 1}) second, hello = self.attach(ack=2) # Newest attach wins: the first connection is cut off. self.assertEqual(first.recv(1), b"") self.assertEqual((hello["hello"]["epoch"], hello["hello"]["ack"]), (2, 1)) (resent,) = self.frames(second, 1) self.assertEqual((resent["seq"], resent["frame"]["id"]), (3, "2")) # The daemon replays seq 1 after its restart: it is not delivered again. replay = {"seq": 1, "ack": 3, "frame": {"type": "release", "handle": "h"}} albedo_link.write_frame(second.send, albedo_link.encode(replay)) self.assertEqual(albedo_link.read_frame(second.recv), {"ack": 1}) self.assertEqual(len(self.received), 1) self.assertEqual(self.link.outbox.pending(), [])
def test_frames_sent_while_detached_wait_for_the_next_attach(self): self.link.send({"type": "done", "id": "c1"}) self.assertGreater(self.link.idle_for(), 0) connection, _ = self.attach() (done,) = self.frames(connection, 1) self.assertEqual(done["frame"], {"type": "done", "id": "c1"}) self.assertEqual(self.link.idle_for(), 0)
if __name__ == "__main__": unittest.main()