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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195"""Pinned sessions keep an order a client can change, and archiving leaves it alone.
The daemon renumbers every pin in the commit that moves one, and refuses a movethat names a session that is not pinned; only a daemon with its storage shows both."""
import http.clientimport jsonimport socketimport threadingimport unittestimport urllib.errorimport urllib.parse
from harness import Albedo, Provider, exclusive, text, wait_until
_pin_lock = threading.Lock()
class SessionPinsTests(unittest.TestCase): def setUp(self): _pin_lock.acquire() self.addCleanup(_pin_lock.release) provider = Provider(lambda _request: text("done")) self.addCleanup(provider.close) self.app = Albedo(provider).__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.sessions = [self.app.session() for _ in range(3)]
def patch(self, session, preferences, etag=None): route = f"/sessions/{session}?view=configuration" if etag is None: with self.app.api(route) as response: response.read() etag = response.headers["ETag"] with self.app.api( route, {"preferences": preferences}, method="PATCH", headers={"If-Match": etag, "Content-Type": "application/merge-patch+json"}, ) as response: return json.load(response)
def rejected(self, session, preferences): with self.assertRaises(urllib.error.HTTPError) as caught: self.patch(session, preferences) return caught.exception.code, json.load(caught.exception)["code"]
def listen(self): """The events the daemon-wide sessions stream carries from now on.""" heard = [] ready, stopped = threading.Event(), threading.Event() address = urllib.parse.urlsplit(self.app.base) connection = http.client.HTTPConnection( address.hostname, address.port, timeout=30 )
def consume(): try: connection.request( "GET", "/sessions?scope=roots", headers={ "Authorization": "Bearer " + self.app.connection["token"], "Accept": "text/event-stream", }, ) with connection.getresponse() as response: for raw in response: if raw.startswith(b"data: "): heard.extend(json.loads(raw[6:])["events"]) ready.set() except (OSError, ValueError): if not stopped.is_set(): raise
thread = threading.Thread(target=consume, daemon=True) thread.start() self.assertTrue(ready.wait(timeout=5), "sessions stream did not send ready")
def close(): stopped.set() if connection.sock is not None: connection.sock.shutdown(socket.SHUT_RDWR) thread.join(timeout=2) connection.close()
self.addCleanup(close) return heard
def pinned(self): """Our sessions, pinned, in the order the daemon lists them.""" with self.app.api("/sessions?scope=roots&limit=200") as response: items = json.load(response)["items"] ours = [item for item in items if item["id"] in self.sessions] ranked = sorted( (item for item in ours if item["preferences"]["pinned"]), key=lambda item: item["preferences"]["pin_order"], ) return [item["id"] for item in ranked]
def test_pins_move_among_each_other(self): first, second, third = self.sessions for session in self.sessions: self.patch(session, {"pinned": True}) self.assertEqual(self.pinned(), [first, second, third])
self.patch(third, {"pin_before": first}) self.assertEqual(self.pinned(), [third, first, second])
self.patch(third, {"pin_before": None}) self.assertEqual(self.pinned(), [first, second, third])
self.patch(first, {"archived": True}) self.assertEqual(self.pinned(), [first, second, third])
def test_a_move_is_refused_unless_both_sessions_are_pinned(self): first, second, _ = self.sessions self.assertEqual( self.rejected(first, {"pin_before": second}), (400, "pin_move_unpinned"), ) self.patch(first, {"pinned": True}) self.assertEqual( self.rejected(first, {"pin_before": second}), (409, "pin_target_invalid"), ) self.assertEqual(self.pinned(), [first])
def test_a_pin_and_its_place_arrive_together(self): first, second, _ = self.sessions self.patch(first, {"pinned": True}) self.patch(second, {"pinned": True, "pin_before": first}) self.assertEqual(self.pinned(), [second, first])
def test_a_move_leaves_the_pins_outside_it_alone(self): first, second, third = self.sessions gap = self.app.session() for session in (first, gap, second, third): self.patch(session, {"pinned": True}) self.patch(gap, {"pinned": False}) route = f"/sessions/{third}?view=configuration" with self.app.api(route) as response: response.read() stale = response.headers["ETag"] self.patch(second, {"pin_before": first}) self.assertEqual(self.pinned(), [second, first, third]) # The third pin kept its number, so an edit against its read still lands. self.patch(third, {"archived": True}, etag=stale)
# Pin numbers are daemon-wide: on the shared daemon another test's pins can # leave a free number before ours, and the move would need no renumbering. @exclusive def test_a_move_announces_the_sessions_it_renumbered(self): first, second, third = self.sessions for session in self.sessions: self.patch(session, {"pinned": True}) heard = self.listen() self.patch(second, {"pin_before": first})
def announced(): return { session for event in list(heard) if event["type"] == "invalidate" for session in event["data"]["session_ids"] }
wait_until( lambda: {first, second} <= announced(), 5, "the sessions a move renumbered were not announced", ) # The third pin kept its number, so there is nothing to tell its readers. self.assertNotIn(third, announced())
@exclusive def test_a_move_renumbers_the_pins_other_clients_read(self): first, second, _ = self.sessions self.patch(first, {"pinned": True}) self.patch(second, {"pinned": True}) route = f"/sessions/{first}?view=configuration" with self.app.api(route) as response: response.read() stale = response.headers["ETag"] self.patch(second, {"pin_before": first}) with self.assertRaises(urllib.error.HTTPError) as caught: self.patch(first, {"archived": True}, etag=stale) self.assertEqual(caught.exception.code, 412)
if __name__ == "__main__": unittest.main()