tl/scenes/tests.py

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 and stamps identity; the party itself
    is worked out in the browsers, so there is nothing here to store 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):
        comm = WebsocketCommunicator(application, f"/ws/projects/{self.project.pk}/")
        comm.scope["user"] = user or AnonymousUser()
        connected, _ = await comm.connect()
        self.assertTrue(connected)
        return comm, await comm.receive_json_from()

    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_presence_is_stamped_with_the_server_side_identity(self):
        a, a_hello = await self.open(self.alice)
        await a.receive_json_from()                       # our own join, echoed back
        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.receive_json_from()                       # join
        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)
        await a.receive_json_from()                       # join
        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()