straywild-discovery-server/tests/test_social_registry.py

from __future__ import annotations

import hashlib
import unittest

from straywild_discovery.registry import RoomRegistry, ValidationError
from straywild_discovery.social_registry import (
    FriendOfflineError,
    INBOX_REACHABLE_SECONDS,
    PRESENCE_TTL_SECONDS,
    SocialRegistry,
)


GAME_VERSION = "0.20.3-alpha"
PROTOCOL_VERSION = 12
VALID_ROOM = {
    "room_name": "Pond Friends",
    "port": 7777,
    "current_players": 1,
    "max_players": 8,
    "game_version": GAME_VERSION,
    "protocol_version": PROTOCOL_VERSION,
}


class FakeClock:
    def __init__(self) -> None:
        self.now = 1_000.0

    def __call__(self) -> float:
        return self.now


def capability_id(domain: str, token: str) -> str:
    return hashlib.sha256(f"{domain}{token}".encode()).hexdigest()


class SocialRegistryTests(unittest.TestCase):
    def setUp(self) -> None:
        self.clock = FakeClock()
        self.rooms = RoomRegistry(ttl_seconds=300.0, clock=self.clock)
        room, _lease, verification = self.rooms.create(
            "203.0.113.10", VALID_ROOM
        )
        self.room = self.rooms.verify_endpoint(
            room.room_id, verification, "203.0.113.10", 7777
        )
        self.social = SocialRegistry(clock=self.clock)

    def test_presence_is_capability_addressed_and_expires(self) -> None:
        write_token = "a" * 64
        channel = capability_id("straywild_PRESENCE_V1:", write_token)
        self.social.publish_presence(
            "203.0.113.10",
            {
                "write_tokens": [write_token],
                "online": True,
                "display_name": "Voyager",
                "room_id": self.room.room_id,
                "game_version": GAME_VERSION,
                "protocol_version": PROTOCOL_VERSION,
            },
        )
        presence = self.social.query_presence(
            [channel], GAME_VERSION, PROTOCOL_VERSION, self.rooms
        )
        self.assertEqual(len(presence), 1)
        self.assertEqual(presence[0]["channel"], channel)
        self.assertEqual(presence[0]["room"]["room_id"], self.room.room_id)

        self.clock.now += PRESENCE_TTL_SECONDS
        self.assertEqual(
            self.social.query_presence(
                [channel], GAME_VERSION, PROTOCOL_VERSION, self.rooms
            ),
            [],
        )

    def test_presence_keeps_pre_rebrand_friend_channels_reachable(self) -> None:
        write_token = "c" * 64
        legacy_channel = capability_id("NETFISHING_PRESENCE_V1:", write_token)
        channels = self.social.publish_presence(
            "203.0.113.10",
            {
                "write_tokens": [write_token],
                "online": True,
                "display_name": "Voyager",
                "room_id": self.room.room_id,
                "game_version": GAME_VERSION,
                "protocol_version": PROTOCOL_VERSION,
            },
        )
        self.assertIn(legacy_channel, channels)
        presence = self.social.query_presence(
            [legacy_channel], GAME_VERSION, PROTOCOL_VERSION, self.rooms
        )
        self.assertEqual(len(presence), 1)
        self.assertEqual(presence[0]["channel"], legacy_channel)

    def test_invitation_requires_a_current_live_poll(self) -> None:
        inbox_token = "b" * 64
        with self.assertRaisesRegex(
            FriendOfflineError,
            "This person needs to be online to do this",
        ):
            self.social.send_invitation(
                inbox_token,
                self.room.room_id,
                GAME_VERSION,
                PROTOCOL_VERSION,
                self.rooms,
            )

        self.assertEqual(
            self.social.poll_invitations(
                [inbox_token], GAME_VERSION, PROTOCOL_VERSION, self.rooms
            ),
            [],
        )
        sent = self.social.send_invitation(
            inbox_token,
            self.room.room_id,
            GAME_VERSION,
            PROTOCOL_VERSION,
            self.rooms,
        )
        received = self.social.poll_invitations(
            [inbox_token], GAME_VERSION, PROTOCOL_VERSION, self.rooms
        )
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]["invite_id"], sent.invite_id)
        self.assertEqual(received[0]["room"]["room_id"], self.room.room_id)

        self.clock.now += INBOX_REACHABLE_SECONDS + 0.01
        with self.assertRaises(FriendOfflineError):
            self.social.send_invitation(
                inbox_token,
                self.room.room_id,
                GAME_VERSION,
                PROTOCOL_VERSION,
                self.rooms,
            )

    def test_invalid_capabilities_are_rejected(self) -> None:
        with self.assertRaises(ValidationError):
            self.social.publish_presence(
                "203.0.113.10",
                {
                    "write_tokens": ["not-a-capability"],
                    "online": False,
                },
            )


if __name__ == "__main__":
    unittest.main()