"""Unit tests for the roster. No sockets, no async — pure policy.""" from __future__ import annotations import pytest from mirror.registry import Registry, RegistryError def by_target(outbox): """Group emitted envelopes by their destination peer id.""" grouped: dict[str, list[dict]] = {} for peer_id, message in outbox: grouped.setdefault(peer_id, []).append(message) return grouped def types_for(outbox, peer_id): return [m["type"] for m in by_target(outbox).get(peer_id, [])] def message_for(outbox, peer_id, kind): for message in by_target(outbox).get(peer_id, []): if message["type"] == kind: return message raise AssertionError(f"no {kind!r} for {peer_id!r} in {by_target(outbox)}") @pytest.fixture def mesh(registry): """One sharer, two viewers, and the sharer already mirroring.""" sharer, _ = registry.join(role="sharer", name="Linux Box") v1, _ = registry.join(role="viewer", name="Phone") v2, _ = registry.join(role="viewer", name="Tablet") registry.publish(sharer, label="Terminal", width=1280, height=720) return sharer, v1, v2, registry class TestJoin: def test_viewer_gets_an_empty_roster(self, registry): peer_id, outbox = registry.join(role="viewer", name="Phone") welcome = message_for(outbox, peer_id, "welcome") assert welcome["peerId"] == peer_id assert welcome["role"] == "viewer" assert welcome["streams"] == [] def test_ids_are_unique_and_sequential_when_injected(self, registry): first, _ = registry.join(role="viewer") second, _ = registry.join(role="viewer") assert (first, second) == ("p1", "p2") def test_generated_ids_do_not_collide(self): registry = Registry(id_factory=lambda: "same") registry.join(role="viewer") with pytest.raises(RegistryError, match="out of peer ids"): registry.join(role="viewer") def test_default_names_are_useful(self, registry): sharer, _ = registry.join(role="sharer") viewer, _ = registry.join(role="viewer") assert registry.peers[sharer].name == "This device" assert registry.peers[viewer].name == "Viewer" def test_names_are_sanitised_and_capped(self, registry): peer_id, _ = registry.join(role="sharer", name=" a\n\t b " + "x" * 80) name = registry.peers[peer_id].name assert name.startswith("a b ") assert len(name) == 48 assert "\n" not in name and " " not in name def test_unknown_role_is_fatal(self, registry): with pytest.raises(RegistryError) as excinfo: registry.join(role="spectator") assert excinfo.value.code == "bad_role" assert excinfo.value.fatal is True def test_a_late_viewer_sees_existing_mirrors(self, mesh): sharer, _v1, _v2, registry = mesh late, outbox = registry.join(role="viewer", name="Late") roster = message_for(outbox, late, "welcome")["streams"] assert [s["id"] for s in roster] == [sharer] assert roster[0]["label"] == "Terminal" assert roster[0]["viewers"] == 0 class TestPublish: def test_viewers_are_told_and_other_sharers_are_not(self, registry): sharer, _ = registry.join(role="sharer", name="A") other_sharer, _ = registry.join(role="sharer", name="B") viewer, _ = registry.join(role="viewer", name="V") outbox = registry.publish(sharer, label="Terminal", width=800, height=600) assert types_for(outbox, viewer) == ["stream_added"] assert by_target(outbox).get(other_sharer) is None added = message_for(outbox, viewer, "stream_added")["stream"] assert added == { "id": sharer, "sharerId": sharer, "name": "A", "label": "Terminal", "displaySurface": "", "width": 800, "height": 600, "audio": False, "viewers": 0, } def test_a_viewer_cannot_publish(self, registry): viewer, _ = registry.join(role="viewer") with pytest.raises(RegistryError) as excinfo: registry.publish(viewer) assert excinfo.value.code == "not_sharer" assert excinfo.value.fatal is True def test_republish_updates_in_place(self, registry): sharer, _ = registry.join(role="sharer", name="A") registry.join(role="viewer") registry.publish(sharer, label="First", width=800, height=600) registry.publish(sharer, label="Second", width=1920, height=1080, audio=True) assert len(registry.streams) == 1 assert registry.roster()[0]["label"] == "Second" assert registry.roster()[0]["audio"] is True def test_unchanged_republish_does_not_re_offer_to_viewers(self, mesh): sharer, v1, _v2, registry = mesh registry.subscribe(v1, sharer) outbox = registry.publish(sharer, label="Terminal", width=1280, height=720) # The viewer already has a peer connection; nothing to renegotiate. assert types_for(outbox, v1) == ["stream_updated"] assert types_for(outbox, sharer) == ["stream_updated"] def test_changed_republish_asks_for_a_new_offer_per_viewer(self, mesh): sharer, v1, v2, registry = mesh registry.subscribe(v1, sharer) registry.subscribe(v2, sharer) outbox = registry.publish(sharer, label="Terminal", width=1920, height=1080) assert types_for(outbox, v1).count("peer_joined") == 1 assert types_for(outbox, v2).count("peer_joined") == 1 assert message_for(outbox, v1, "peer_joined")["streamId"] == sharer def test_unpublish_clears_the_tile_for_every_viewer(self, mesh): """A tile is drawn on `stream_added`, so every viewer must be told it is gone — including one that had not subscribed yet.""" sharer, v1, v2, registry = mesh registry.subscribe(v1, sharer) outbox = registry.unpublish(sharer) assert types_for(outbox, v1) == ["stream_removed"] assert types_for(outbox, v2) == ["stream_removed"] assert registry.streams == {} assert registry.watching(v1) == set() def test_dropping_a_mirror_leaves_other_mirrors_alone(self, registry): first, _ = registry.join(role="sharer", name="A") second, _ = registry.join(role="sharer", name="B") viewer, _ = registry.join(role="viewer") registry.publish(first, label="one") registry.publish(second, label="two") registry.subscribe(viewer, second) outbox = registry.unpublish(first) assert [m["streamId"] for m in by_target(outbox)[viewer]] == [first] assert [s["label"] for s in registry.roster()] == ["two"] assert registry.watching(viewer) == {second} def test_unpublish_without_a_stream_is_a_no_op(self, registry): sharer, _ = registry.join(role="sharer") assert registry.unpublish(sharer) == [] def test_capacity_blocks_a_new_mirror_but_not_an_existing_one(self): registry = Registry(max_streams=1) first, _ = registry.join(role="sharer", name="A") second, _ = registry.join(role="sharer", name="B") viewer, _ = registry.join(role="viewer") registry.publish(first, label="one") with pytest.raises(RegistryError) as excinfo: registry.publish(second, label="two") assert excinfo.value.code == "capacity" assert excinfo.value.fatal is True # The incumbent may keep refreshing its own mirror. registry.publish(first, label="one", width=1920, height=1080) assert [s["label"] for s in registry.roster()] == ["one"] # Capacity counts mirrors; an audience never uses it up. assert registry.stats()["viewers"] == 1 assert registry.watching(viewer) == set() class TestSubscribe: def test_sharer_is_told_to_negotiate(self, mesh): sharer, v1, _v2, registry = mesh outbox = registry.subscribe(v1, sharer) joined = message_for(outbox, sharer, "peer_joined") assert joined["viewerId"] == v1 assert joined["name"] == "Phone" assert message_for(outbox, v1, "stream_updated")["stream"]["viewers"] == 1 def test_subscribing_twice_is_idempotent(self, mesh): sharer, v1, _v2, registry = mesh registry.subscribe(v1, sharer) assert registry.subscribe(v1, sharer) == [] assert registry.streams[sharer].viewers == {v1} def test_unknown_stream(self, registry): viewer, _ = registry.join(role="viewer") with pytest.raises(RegistryError) as excinfo: registry.subscribe(viewer, "nope") assert excinfo.value.code == "unknown_stream" def test_a_sharer_cannot_even_watch_itself(self, mesh): sharer, _v1, _v2, registry = mesh with pytest.raises(RegistryError) as excinfo: registry.subscribe(sharer, sharer) assert excinfo.value.code == "not_viewer" def test_a_sharer_cannot_subscribe_like_a_viewer(self, mesh): sharer, _v1, _v2, registry = mesh other, _ = registry.join(role="sharer", name="B") registry.publish(other, label="B's screen") with pytest.raises(RegistryError) as excinfo: registry.subscribe(sharer, other) assert excinfo.value.code == "not_viewer" def test_unsubscribe_closes_the_leg(self, mesh): sharer, v1, v2, registry = mesh registry.subscribe(v1, sharer) registry.subscribe(v2, sharer) outbox = registry.unsubscribe(v1, sharer) assert message_for(outbox, sharer, "peer_left")["viewerId"] == v1 assert message_for(outbox, v2, "stream_updated")["stream"]["viewers"] == 1 assert registry.unsubscribe(v1, sharer) == [] def test_unsubscribing_from_an_unknown_stream_is_quiet(self, registry): viewer, _ = registry.join(role="viewer") assert registry.unsubscribe(viewer, "ghost") == [] class TestSignalAuthorization: def test_sharer_may_reach_a_subscribed_viewer(self, mesh): sharer, v1, _v2, registry = mesh registry.subscribe(v1, sharer) outbox = registry.signal(sharer, v1, {"sdp": {"type": "offer"}}) relay = message_for(outbox, v1, "signal") assert relay["from"] == sharer assert relay["payload"] == {"sdp": {"type": "offer"}} def test_viewer_may_answer_the_sharer_it_watches(self, mesh): sharer, v1, _v2, registry = mesh registry.subscribe(v1, sharer) assert registry.signal(v1, sharer, {"candidate": {"candidate": "x"}}) == [ ( sharer, { "type": "signal", "from": v1, "to": sharer, "payload": {"candidate": {"candidate": "x"}}, }, ) ] def test_sharer_may_not_blindly_push_to_a_stranger(self, mesh): sharer, _v1, v2, registry = mesh with pytest.raises(RegistryError) as excinfo: registry.signal(sharer, v2, {"sdp": {"type": "offer"}}) assert excinfo.value.code == "not_subscribed" def test_viewer_may_not_reach_a_sharer_it_never_subscribed_to(self, mesh): _sharer, v1, _v2, registry = mesh other, _ = registry.join(role="sharer", name="C") registry.publish(other, label="private") with pytest.raises(RegistryError) as excinfo: registry.signal(v1, other, {}) assert excinfo.value.code == "not_subscribed" def test_peers_may_not_talk_to_each_other_directly(self, mesh): _sharer, v1, v2, registry = mesh with pytest.raises(RegistryError) as excinfo: registry.signal(v1, v2, {}) assert excinfo.value.code == "peer_to_peer" def test_unknown_endpoints_are_refused(self, mesh): sharer, _v1, _v2, registry = mesh with pytest.raises(RegistryError) as excinfo: registry.signal(sharer, "ghost", {}) assert excinfo.value.code == "unknown_peer" class TestTeardown: def test_sharer_disconnect_removes_the_mirror(self, mesh): sharer, v1, v2, registry = mesh registry.subscribe(v1, sharer) registry.subscribe(v2, sharer) outbox = registry.leave(sharer) assert types_for(outbox, v1) == ["stream_removed"] assert types_for(outbox, v2) == ["stream_removed"] assert registry.streams == {} assert registry.watching(v1) == set() def test_viewer_disconnect_releases_the_sharer(self, mesh): sharer, v1, v2, registry = mesh registry.subscribe(v1, sharer) registry.subscribe(v2, sharer) outbox = registry.leave(v1) assert message_for(outbox, sharer, "peer_left")["viewerId"] == v1 assert registry.streams[sharer].viewers == {v2} def test_leaving_twice_is_quiet(self, mesh): _sharer, v1, _v2, registry = mesh registry.leave(v1) assert registry.leave(v1) == [] def test_sharer_disconnect_after_a_crash_still_cleans_subscriptions(self, mesh): sharer, v1, _v2, registry = mesh registry.subscribe(v1, sharer) registry.leave(sharer) # A stale viewer must not be able to signal the dead sharer. with pytest.raises(RegistryError): registry.signal(v1, sharer, {}) class TestLayout: def test_the_default_is_the_wall_not_fullscreen_with_zero_followers(self, registry): registry.join(role="viewer") assert registry.layout.staged is None assert registry.layout.fullscreen is False assert registry.layout_dto() == { "type": "layout", "staged": None, "fullscreen": False, "followers": 0, } def test_staging_broadcasts_to_viewers_and_the_sender_without_doubling(self, mesh): sharer, v1, v2, registry = mesh outbox = registry.set_layout(sharer, staged=sharer) assert {t for t, _ in outbox} == {sharer, v1, v2} for peer_id in (sharer, v1, v2): assert types_for(outbox, peer_id) == ["layout"] expected = {"type": "layout", "staged": sharer, "fullscreen": False, "followers": 0} assert all(m == expected for _t, m in outbox) assert len(outbox) == 3 # one frame per target; nobody receives it twice def test_a_partial_update_touches_only_the_named_fields(self, mesh): sharer, _v1, _v2, registry = mesh registry.set_layout(sharer, staged=sharer) outbox = registry.set_layout(sharer, fullscreen=True) dto = message_for(outbox, sharer, "layout") assert registry.layout.staged == sharer # absent field left alone assert registry.layout.fullscreen is True assert dto == {"type": "layout", "staged": sharer, "fullscreen": True, "followers": 0} def test_an_explicit_null_clears_and_an_opinionless_arrange_is_ack_only(self, mesh): sharer, v1, v2, registry = mesh registry.set_layout(sharer, staged=sharer, fullscreen=True) outbox = registry.set_layout(v1, staged=None) assert registry.layout.staged is None assert registry.layout.fullscreen is True # absent field left alone # The clearing is a real change: viewers plus the sender (deduped) are told. assert {t for t, _ in outbox} == {v1, v2} # No opinion at all: the sender is acked and nobody else hears anything. outbox = registry.set_layout(v2) assert outbox == [ (v2, {"type": "layout", "staged": None, "fullscreen": True, "followers": 0}) ] def test_an_unknown_staged_mirror_is_refused_and_changes_nothing(self, mesh): sharer, _v1, _v2, registry = mesh registry.set_layout(sharer, staged=sharer, fullscreen=True) with pytest.raises(RegistryError) as excinfo: registry.set_layout(sharer, staged="ghost", fullscreen=False) assert excinfo.value.code == "unknown_stream" assert registry.layout.staged == sharer assert registry.layout.fullscreen is True def test_repeating_the_same_state_replies_with_the_ack_only(self, mesh): sharer, _v1, _v2, registry = mesh registry.set_layout(sharer, staged=sharer) outbox = registry.set_layout(sharer, staged=sharer, fullscreen=False) assert outbox == [ (sharer, {"type": "layout", "staged": sharer, "fullscreen": False, "followers": 0}) ] def test_unpublishing_the_staged_mirror_falls_back_to_the_wall(self, mesh): sharer, v1, v2, registry = mesh registry.set_layout(sharer, staged=sharer) outbox = registry.unpublish(sharer) for peer_id in (v1, v2): # Exactly one layout frame, riding the same envelope set as the removal. assert types_for(outbox, peer_id) == ["stream_removed", "layout"] assert message_for(outbox, peer_id, "layout")["staged"] is None assert registry.layout.staged is None def test_a_sharer_disconnecting_falls_back_to_the_wall(self, mesh): sharer, v1, v2, registry = mesh registry.set_layout(sharer, staged=sharer) outbox = registry.leave(sharer) for peer_id in (v1, v2): assert types_for(outbox, peer_id) == ["stream_removed", "layout"] assert message_for(outbox, peer_id, "layout")["staged"] is None assert registry.layout.staged is None def test_a_replaced_stream_leaves_the_stage_cleared(self, mesh): sharer, v1, _v2, registry = mesh registry.set_layout(sharer, staged=sharer) registry.leave(sharer) # the old mirror goes with its socket fresh, _ = registry.join(role="sharer", name="Linux Box") outbox = registry.publish(fresh, label="Terminal", width=1280, height=720) # The replacement has a new id; the stage must not still point at the # old one, and nothing auto-stages the newcomer. assert fresh != sharer assert registry.layout.staged is None assert "layout" not in types_for(outbox, v1) assert [s["id"] for s in registry.roster()] == [fresh] def test_the_stage_is_said_once_when_a_kiosk_sharer_leaves(self, registry): kiosk_sharer, _ = registry.join(role="sharer", kiosk=True) viewer, _ = registry.join(role="viewer") registry.publish(kiosk_sharer, label="Terminal") registry.set_layout(kiosk_sharer, staged=kiosk_sharer) outbox = registry.leave(kiosk_sharer) # The fallback and the follower change must not double the layout frame. assert types_for(outbox, viewer) == ["stream_removed", "layout"] def test_followers_counts_only_kiosk_peers(self, registry): registry.join(role="viewer") registry.join(role="viewer") registry.join(role="viewer", kiosk=True) registry.join(role="sharer") assert registry.layout_dto()["followers"] == 1 def test_a_kiosk_joining_pushes_the_new_follower_count(self, mesh): _sharer, v1, v2, registry = mesh _id, outbox = registry.join(role="viewer", kiosk=True) # Told to the other viewers; the joiner got the state in its welcome, # not as a layout frame. assert {t for t, m in outbox if m["type"] == "layout"} == {v1, v2} assert types_for(outbox, _id) == ["welcome"] assert message_for(outbox, v1, "layout")["followers"] == 1 assert message_for(outbox, v1, "layout") == message_for(outbox, v2, "layout") assert registry.layout_dto()["followers"] == 1 def test_a_non_kiosk_join_says_nothing_about_the_layout(self, mesh): _sharer, v1, _v2, registry = mesh _id, outbox = registry.join(role="viewer") assert "layout" not in types_for(outbox, v1) def test_a_kiosk_leaving_pushes_the_new_follower_count(self, mesh): _sharer, v1, v2, registry = mesh kiosk, _ = registry.join(role="viewer", kiosk=True) assert registry.layout_dto()["followers"] == 1 outbox = registry.leave(kiosk) assert {t for t, _ in outbox} == {v1, v2} assert message_for(outbox, v1, "layout")["followers"] == 0 assert registry.layout_dto()["followers"] == 0 def test_a_peer_joined_after_an_arrange_arrives_already_arranged(self, mesh): sharer, _v1, _v2, registry = mesh registry.set_layout(sharer, staged=sharer, fullscreen=True) late, outbox = registry.join(role="viewer", kiosk=True) welcome = message_for(outbox, late, "welcome") assert welcome["layout"] == { "type": "layout", "staged": sharer, "fullscreen": True, "followers": 1, } def test_a_dead_peer_cannot_arrange_the_wall(self, mesh): _sharer, _v1, _v2, registry = mesh with pytest.raises(RegistryError) as excinfo: registry.set_layout("ghost", staged="ghost") assert excinfo.value.code == "unknown_peer" class TestRenameAndStats: def test_rename_refreshes_the_mirrors_that_show_it(self, mesh): sharer, v1, _v2, registry = mesh outbox = registry.rename(sharer, name="Meeting Room") assert message_for(outbox, v1, "stream_updated")["stream"]["name"] == "Meeting Room" assert registry.roster()[0]["name"] == "Meeting Room" def test_rename_to_the_same_name_is_quiet(self, mesh): sharer, _v1, _v2, registry = mesh assert registry.rename(sharer, name="Linux Box") == [] def test_stats_count_every_role(self, mesh): sharer, v1, _v2, registry = mesh registry.subscribe(v1, sharer) assert registry.stats() == { "mirrors": 1, "sharers": 1, "viewers": 2, "connections": 3, "watching": 1, } def test_roster_is_oldest_first(self, registry): first, _ = registry.join(role="sharer", name="first") registry.publish(first, label="first") second, _ = registry.join(role="sharer", name="second") registry.publish(second, label="second") assert [s["label"] for s in registry.roster()] == ["first", "second"]