"""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 move that names a session that is not pinned; only a daemon with its storage shows both. """ import http.client import json import socket import threading import unittest import urllib.error import 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()