"""The kernel half of the session layer, exercised frame by frame.
Sequence, acknowledgement, and replay rules decide whether a reconnect loses
or repeats a message, and E2E cannot choose where a connection drops. These
tests drive albedo_link directly: the outbox's coalescing and bounds, replayed
daemon frames, token checks, newest-attach-wins, and resending what the daemon
never acknowledged.
"""
from pathlib import Path
import os
import shutil
import socket
import sys
import tempfile
import threading
import 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_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()