zyrarbot/tests/test_api_surface.py

"""Przypina powierzchnię API - panel jest osobnym klientem, więc kontrakt musi być stabilny."""

from __future__ import annotations

import asyncio
import json
import tempfile
import unittest
from unittest.mock import MagicMock

from zyrarbot.application import Application
from zyrarbot.settings import Settings

EXPECTED_ENDPOINTS = {
    ("get", "/api/status"),
    ("get", "/api/health"),
    ("post", "/api/restart"),
    ("post", "/api/play"),
    ("post", "/api/play_file"),
    ("post", "/api/pause"),
    ("post", "/api/resume"),
    ("post", "/api/seek"),
    ("post", "/api/skip"),
    ("post", "/api/stop"),
    ("post", "/api/volume"),
    ("post", "/api/effects/toggle"),
    ("post", "/api/loop"),
    ("post", "/api/loop_playlist"),
    ("post", "/api/loop_cycle"),
    ("post", "/api/jump"),
    ("post", "/api/remove"),
    ("post", "/api/reorder"),
    ("post", "/api/queue/restore"),
    ("post", "/api/queue/plays"),
    ("get", "/api/uploads"),
    ("post", "/api/upload"),
    ("delete", "/api/uploads/{filename}"),
    ("post", "/api/connect"),
    ("post", "/api/disconnect"),
    ("post", "/api/notification_channel"),
    ("post", "/api/default_plays"),
}


class ApiSurfaceTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls._workdir = tempfile.TemporaryDirectory()
        settings = Settings(
            discord_token="test",
            config_dir=f"{cls._workdir.name}/config",
            uploads_dir=f"{cls._workdir.name}/uploads",
            static_dir=f"{cls._workdir.name}/static",
        )
        persistence = MagicMock()
        persistence.config = MagicMock()
        persistence.history = MagicMock()
        persistence.queues = MagicMock()
        cls.app = Application(settings, persistence=persistence).web
        cls.spec = cls.app.openapi()

    @classmethod
    def tearDownClass(cls) -> None:
        cls._workdir.cleanup()

    def _declared(self) -> set[tuple[str, str]]:
        return {
            (method, path)
            for path, operations in self.spec["paths"].items()
            for method in operations
        }

    def test_all_expected_endpoints_are_declared(self):
        self.assertEqual(set(), EXPECTED_ENDPOINTS - self._declared())

    def test_no_endpoint_added_without_updating_this_test(self):
        # Panel jest odrębnym klientem - nowa trasa powinna trafić tu świadomie.
        self.assertEqual(set(), self._declared() - EXPECTED_ENDPOINTS)

    def test_openapi_document_is_well_formed(self):
        self.assertTrue(self.spec["openapi"].startswith("3."))
        self.assertIn("ŻyrarBOT", self.spec["info"]["title"])

    def test_health_endpoint_returns_api_response_without_database(self):
        messages = []
        received = False

        async def receive():
            nonlocal received
            if received:
                return {"type": "http.disconnect"}
            received = True
            return {"type": "http.request", "body": b"", "more_body": False}

        async def send(message):
            messages.append(message)

        scope = {
            "type": "http",
            "asgi": {"version": "3.0", "spec_version": "2.3"},
            "http_version": "1.1",
            "method": "GET",
            "scheme": "http",
            "path": "/api/health",
            "raw_path": b"/api/health",
            "query_string": b"",
            "root_path": "",
            "headers": [],
            "client": ("testclient", 50000),
            "server": ("testserver", 80),
        }
        asyncio.run(self.app(scope, receive, send))

        start = next(message for message in messages if message["type"] == "http.response.start")
        body = next(message for message in messages if message["type"] == "http.response.body")
        self.assertEqual(start["status"], 200)
        self.assertEqual(json.loads(body["body"]), {"status": "ok", "ready": False})

    def test_restart_endpoint_calls_restart_state(self):
        import asyncio
        from unittest.mock import MagicMock

        from zyrarbot.ui.api.routers.status_router import restart_bot

        mock_request = MagicMock()
        mock_restart = MagicMock()
        mock_request.app.state.restart = mock_restart

        result = asyncio.run(restart_bot(mock_request))
        self.assertEqual(result, {"status": "restarting"})
        mock_restart.assert_called_once()

    def test_voice_channels_payload_filters_unconnectable_channels(self):
        from unittest.mock import MagicMock

        from zyrarbot.ui.api.status_service import StatusService

        mock_guild = MagicMock()
        mock_guild.id = 123
        mock_guild.name = "Test Guild"

        ch1 = MagicMock()
        ch1.id = 1
        ch1.name = "Allowed Voice"
        ch1.permissions_for.return_value.connect = True

        ch2 = MagicMock()
        ch2.id = 2
        ch2.name = "Forbidden Voice"
        ch2.permissions_for.return_value.connect = False

        mock_guild.voice_channels = [ch1, ch2]
        mock_guild.me = MagicMock()

        mock_guilds_resolver = MagicMock()
        mock_guilds_resolver.guilds.return_value = [mock_guild]

        service = StatusService(MagicMock(), mock_guilds_resolver)
        channels = service._voice_channels_payload(mock_guild)

        self.assertEqual(len(channels), 1)
        self.assertEqual(channels[0]["channel_id"], "1")
        self.assertEqual(channels[0]["channel_name"], "Allowed Voice")


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