matrix/discord-bridge/test_lotus_presence.py

import asyncio
import unittest

import lotus_presence as lp


class MapPresence(unittest.TestCase):
    def test_statuses(self):
        self.assertEqual(lp.map_presence("online", []), ("online", ""))
        self.assertEqual(lp.map_presence("idle", []), ("unavailable", ""))
        self.assertEqual(lp.map_presence("dnd", []), (lp.BUSY, ""))
        self.assertEqual(lp.map_presence("offline", []), ("offline", ""))
        self.assertEqual(lp.map_presence("invisible", None), ("offline", ""))

    def test_offline_never_carries_activity(self):
        game = [{"type": 0, "name": "Valorant"}]
        self.assertEqual(lp.map_presence("offline", game), ("offline", ""))

    def test_activities(self):
        for activity, expected in (
            ({"type": 0, "name": "Valorant"}, "Playing Valorant"),
            ({"type": 1, "name": "Twitch", "details": "Ranked"}, "Streaming Ranked"),
            ({"type": 2, "name": "Spotify", "details": "Song", "state": "Band"},
             "Listening to Song by Band"),
            ({"type": 2, "name": "Spotify"}, "Listening to Spotify"),
            ({"type": 3, "name": "YouTube"}, "Watching YouTube"),
            ({"type": 5, "name": "Arena"}, "Competing in Arena"),
        ):
            self.assertEqual(lp.map_presence("online", [activity])[1], expected)

    def test_custom_status_comes_first_with_one_activity(self):
        activities = [
            {"type": 0, "name": "Valorant"},
            {"type": 0, "name": "Second game"},
            {"type": 4, "name": "Custom Status", "state": "afk", "emoji": {"name": "🍕"}},
        ]
        self.assertEqual(lp.map_presence("online", activities)[1], "🍕 afk · Playing Valorant")

    def test_custom_server_emoji_is_dropped(self):
        custom = {"type": 4, "state": "hi", "emoji": {"name": "partyblob", "id": "123"}}
        self.assertEqual(lp.map_presence("online", [custom])[1], "hi")

    def test_long_status_is_truncated(self):
        _, message = lp.map_presence("online", [{"type": 0, "name": "x" * 500}])
        self.assertLessEqual(len(message), lp.MAX_STATUS)
        self.assertTrue(message.endswith("…"))


class BridgeConfig(unittest.TestCase):
    def test_reads_the_database_uri(self):
        env = lp.bridge_db_env(
            "appservice:\n  database:\n    type: postgres\n"
            '    uri: "postgres://u:p%40ss@10.0.0.1:5433/db?sslmode=disable"\n'
        )
        self.assertEqual(env["PGHOST"], "10.0.0.1")
        self.assertEqual(env["PGPORT"], "5433")
        self.assertEqual(env["PGPASSWORD"], "p@ss")
        self.assertEqual(env["PGDATABASE"], "db")
        self.assertEqual(env["PGSSLMODE"], "disable")

    def test_missing_uri_is_an_error(self):
        with self.assertRaises(RuntimeError):
            lp.bridge_db_env("uri: sqlite:///x")


class FakeGhosts(lp.GhostPresence):
    def __init__(self, reject=()):
        super().__init__(None, "http://hs", "t", "s", clock=lambda: self.now)
        self.now = 0
        self.sent = []
        self.reject = set(reject)

    async def _put(self, discord_id, presence, message):
        self.sent.append((discord_id, presence, message))
        if discord_id in self.reject:
            self.unknown_until[discord_id] = self.now + lp.UNKNOWN_RETRY_S


class Ghosts(unittest.TestCase):
    def test_only_changes_are_sent(self):
        g = FakeGhosts()
        asyncio.run(g.update("1", "online", ""))
        asyncio.run(g.update("1", "online", ""))
        asyncio.run(g.update("1", "online", "Playing X"))
        self.assertEqual(len(g.sent), 2)

    def test_offline_is_sent_once_and_only_after_being_live(self):
        g = FakeGhosts()
        asyncio.run(g.update("1", "offline", ""))
        self.assertEqual(g.sent, [])
        asyncio.run(g.update("1", "online", ""))
        asyncio.run(g.update("1", "offline", ""))
        asyncio.run(g.update("1", "offline", ""))
        self.assertEqual(g.sent[-1], ("1", "offline", ""))
        self.assertEqual(len(g.sent), 2)

    def test_resync_turns_off_whoever_left(self):
        g = FakeGhosts()
        asyncio.run(g.update("1", "online", ""))
        asyncio.run(g.resync({"2": ("online", "")}))
        self.assertIn(("1", "offline", ""), g.sent)
        self.assertEqual(set(g.live), {"2"})

    def test_refresh_resends_live_states(self):
        g = FakeGhosts()
        asyncio.run(g.update("1", "unavailable", "x"))
        g.sent.clear()
        asyncio.run(g.refresh())
        self.assertEqual(g.sent, [("1", "unavailable", "x")])

    def test_missing_ghost_is_retried_later_not_every_refresh(self):
        g = FakeGhosts(reject={"1"})
        asyncio.run(g.update("1", "online", ""))
        g.sent.clear()
        asyncio.run(g.refresh())
        self.assertEqual(g.sent, [])
        g.now += lp.UNKNOWN_RETRY_S + 1
        asyncio.run(g.refresh())
        self.assertEqual(g.sent, [("1", "online", "")])


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