import json from channels.testing import WebsocketCommunicator from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser from django.test import Client, TestCase, TransactionTestCase from scenes.models import Project, Revision from server.asgi import application User = get_user_model() def ann(name, content="", **extra): return {"type": "annotation", "in": ["root"], "name": name, "content": content, "marks": [], **extra} class SceneApiTestCase(TestCase): def setUp(self): self.alice = User.objects.create_user("alice", password="pw") self.bob = User.objects.create_user("bob", password="pw") self.project = Project.objects.create(owner=self.alice, name="p") self.project.collaborators.add(self.bob) self.a = Client(); self.a.login(username="alice", password="pw") self.b = Client(); self.b.login(username="bob", password="pw") def put(self, client, **body): return client.put(f"/api/projects/{self.project.pk}/scene/", json.dumps(body), content_type="application/json") def groups(self): self.project.refresh_from_db() return self.project.scene.get("groups", {}) class DeltaMergeTests(SceneApiTestCase): def test_different_annotations_merge_without_clobber(self): self.put(self.a, changed={"a1": ann("opening")}) self.put(self.b, changed={"a2": ann("beat")}) self.assertEqual(set(self.groups()), {"a1", "a2"}) def test_same_annotation_same_field_is_last_write_wins(self): self.put(self.a, changed={"a1": ann("x", "first")}) self.put(self.b, changed={"a1": {"content": "second"}}) self.assertEqual(self.groups()["a1"]["content"], "second") def test_same_annotation_different_fields_merge(self): self.put(self.a, changed={"a1": ann("x", "first", marks=[{"id": "m0"}])}) self.put(self.a, changed={"a1": {"marks": [{"id": "m1"}]}}) self.put(self.b, changed={"a1": {"content": "second"}}) g = self.groups()["a1"] self.assertEqual(g["content"], "second") self.assertEqual(g["marks"], [{"id": "m1"}]) def test_stale_client_neither_resurrects_nor_wipes(self): # bob only ever knew a2; his delta must not erase alice's a1 self.put(self.a, changed={"a1": ann("x")}) self.put(self.b, changed={"a2": ann("y")}) self.assertEqual(set(self.groups()), {"a1", "a2"}) def test_delete_is_explicit_and_scoped(self): self.put(self.a, changed={"a1": ann("x"), "a2": ann("y")}) self.put(self.b, deleted=["a1"]) self.assertEqual(set(self.groups()), {"a2"}) class AttributionTests(SceneApiTestCase): def test_server_stamps_creator_ignoring_client(self): self.put(self.a, changed={"a1": ann("x", createdBy="hacker", editedBy="hacker")}) g = self.groups()["a1"] self.assertEqual((g["createdBy"], g["editedBy"]), ("alice", "alice")) def test_edit_preserves_creator_updates_editor(self): self.put(self.a, changed={"a1": ann("x")}) self.put(self.b, changed={"a1": ann("x", "edited")}) g = self.groups()["a1"] self.assertEqual(g["createdBy"], "alice") self.assertEqual(g["editedBy"], "bob") def test_unchanged_annotation_is_not_restamped(self): self.put(self.a, changed={"a1": ann("x")}) before = self.groups()["a1"]["editedAt"] self.put(self.b, changed={"a1": ann("x")}) # identical content after = self.groups()["a1"] self.assertEqual(after["editedAt"], before) self.assertEqual(after["editedBy"], "alice") def test_revision_per_real_change_only(self): self.put(self.a, changed={"a1": ann("x")}) self.put(self.b, changed={"a1": ann("x")}) # no-op → no revision self.put(self.a, deleted=["a1"]) revs = Revision.objects.filter(project=self.project).order_by("created") self.assertEqual([r.summary for r in revs], ["+1 ~0 −0 annotations", "+0 ~0 −1 annotations"]) self.assertEqual(revs[0].user, self.alice) class AccessTests(SceneApiTestCase): def test_write_requires_authentication(self): r = Client().put(f"/api/projects/{self.project.pk}/scene/", json.dumps({"changed": {}}), content_type="application/json") self.assertEqual(r.status_code, 401) def test_non_collaborator_cannot_edit(self): User.objects.create_user("carol", password="pw") carol = Client(); carol.login(username="carol", password="pw") self.assertEqual(self.put(carol, changed={"a1": ann("x")}).status_code, 404) class PresenceRelayTests(TransactionTestCase): """The socket hands out connection ids, stamps identity, and hands a joiner the room in one message. The party itself is worked out in the browsers, so there is nothing here to decide or trust.""" def setUp(self): self.alice = User.objects.create_user("alice", password="pw") self.project = Project.objects.create(owner=self.alice, name="p") async def open(self, user=None): """Connect and drain the handshake, returning (comm, welcome, roster).""" comm = WebsocketCommunicator(application, f"/ws/projects/{self.project.pk}/") comm.scope["user"] = user or AnonymousUser() connected, _ = await comm.connect() self.assertTrue(connected) welcome = await comm.receive_json_from() roster = await comm.receive_json_from() await comm.receive_json_from() # our own join, echoed back return comm, welcome, roster async def test_welcome_carries_an_id_and_the_signed_in_name(self): comm, welcome, _ = await self.open(self.alice) self.assertEqual(welcome["kind"], "welcome") self.assertEqual(welcome["user"], "alice") self.assertTrue(welcome["cid"]) await comm.disconnect() async def test_a_joiner_is_handed_the_whole_room_in_one_message(self): # the alternative — everyone answering a join — costs a broadcast per # member, and a room of a hundred loses most of its roster to it a, a_hello, a_roster = await self.open(self.alice) self.assertEqual(a_roster["peers"], []) # first one in await a.send_json_to({"kind": "state", "party": "P", "joinable": False}) await a.receive_json_from() b, _, b_roster = await self.open() self.assertEqual([p["cid"] for p in b_roster["peers"]], [a_hello["cid"]]) seen = b_roster["peers"][0] self.assertEqual((seen["user"], seen["party"], seen["joinable"]), ("alice", "P", False)) # including what they last said self.assertTrue(await b.receive_nothing(timeout=0.2)) # and nobody answers await a.disconnect(); await b.disconnect() async def test_a_departure_is_forgotten_not_handed_to_the_next_joiner(self): a, _, _ = await self.open(self.alice) b, b_hello, _ = await self.open() await a.receive_json_from() # b's join await b.disconnect() await a.receive_json_from() # b's leave c, _, c_roster = await self.open() self.assertNotIn(b_hello["cid"], [p["cid"] for p in c_roster["peers"]]) await a.disconnect(); await c.disconnect() async def test_presence_is_stamped_with_the_server_side_identity(self): a, a_hello, _ = await self.open(self.alice) b, _, _ = await self.open() self.assertEqual((await a.receive_json_from())["kind"], "join") await b.send_json_to({"kind": "state", "party": a_hello["cid"], "joinable": True, "joining": a_hello["cid"], "user": "alice", "cid": "forged"}) seen = await a.receive_json_from() self.assertEqual(seen["party"], a_hello["cid"]) self.assertIsNone(seen["user"]) # anonymous, not "alice" self.assertNotEqual(seen["cid"], "forged") await a.disconnect(); await b.disconnect() async def test_scene_edits_are_not_accepted_over_the_socket(self): a, _, _ = await self.open(self.alice) await a.send_json_to({"kind": "scene", "changed": {"a1": ann("x")}}) self.assertTrue(await a.receive_nothing(timeout=0.2)) await a.disconnect() async def test_leaving_tells_the_room(self): a, _, _ = await self.open(self.alice) b, b_hello, _ = await self.open() await a.receive_json_from() # b's join await b.disconnect() bye = await a.receive_json_from() self.assertEqual((bye["kind"], bye["cid"]), ("leave", b_hello["cid"])) await a.disconnect()