kodi
140 lines
5.2 KiB
Python
140 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
|
|
|
|
from backend.app.dependencies import get_remote_client_service
|
|
from backend.app.db.remote_client_repository import RemoteClientRepository
|
|
from backend.app.main import app
|
|
from backend.app.services.remote_client_service import RemoteClientService
|
|
|
|
|
|
class _Clock:
|
|
def __init__(self, current: datetime):
|
|
self.current = current
|
|
|
|
def now(self) -> datetime:
|
|
return self.current
|
|
|
|
def advance(self, *, seconds: int) -> None:
|
|
self.current += timedelta(seconds=seconds)
|
|
|
|
|
|
class RemoteClientsApiGoldenTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.clock = _Clock(datetime(2026, 3, 26, 12, 0, 0, tzinfo=timezone.utc))
|
|
repository = RemoteClientRepository(str(Path(self.temp_dir.name) / "remote-clients.db"))
|
|
service = RemoteClientService(
|
|
repository=repository,
|
|
registration_token="secret-token",
|
|
offline_timeout_seconds=60,
|
|
now=self.clock.now,
|
|
)
|
|
|
|
async def _override_remote_client_service() -> RemoteClientService:
|
|
return service
|
|
|
|
app.dependency_overrides[get_remote_client_service] = _override_remote_client_service
|
|
|
|
def tearDown(self) -> None:
|
|
app.dependency_overrides.clear()
|
|
self.temp_dir.cleanup()
|
|
|
|
def _request(self, method: str, url: str, payload: dict | None = None, token: str | None = None) -> httpx.Response:
|
|
async def _run() -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
headers = {}
|
|
if token is not None:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
if method == "GET":
|
|
return await client.get(url, headers=headers)
|
|
return await client.post(url, json=payload, headers=headers)
|
|
|
|
return asyncio.run(_run())
|
|
|
|
@staticmethod
|
|
def _register_payload() -> dict:
|
|
return {
|
|
"client_id": "client-123",
|
|
"display_name": "Jan MacBook",
|
|
"platform": "macos",
|
|
"agent_version": "1.1.0",
|
|
"endpoint": "http://192.168.1.25:8765",
|
|
"shares": [{"key": "downloads", "label": "Downloads"}],
|
|
}
|
|
|
|
def test_list_is_empty_by_default(self) -> None:
|
|
response = self._request("GET", "/api/clients")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json(), {"items": []})
|
|
|
|
def test_register_then_list_then_heartbeat_and_status_timeout(self) -> None:
|
|
register_response = self._request(
|
|
"POST",
|
|
"/api/clients/register",
|
|
self._register_payload(),
|
|
token="secret-token",
|
|
)
|
|
|
|
self.assertEqual(register_response.status_code, 200)
|
|
register_body = register_response.json()
|
|
self.assertEqual(register_body["client_id"], "client-123")
|
|
self.assertEqual(register_body["display_name"], "Jan MacBook")
|
|
self.assertEqual(register_body["status"], "online")
|
|
self.assertEqual(register_body["last_seen"], "2026-03-26T12:00:00Z")
|
|
self.assertIsNone(register_body["last_error"])
|
|
self.assertIsNone(register_body["reachable_at"])
|
|
|
|
list_response = self._request("GET", "/api/clients")
|
|
self.assertEqual(list_response.status_code, 200)
|
|
self.assertEqual(len(list_response.json()["items"]), 1)
|
|
self.assertEqual(list_response.json()["items"][0]["status"], "online")
|
|
|
|
self.clock.advance(seconds=30)
|
|
heartbeat_response = self._request(
|
|
"POST",
|
|
"/api/clients/heartbeat",
|
|
{"client_id": "client-123", "agent_version": "1.1.1"},
|
|
token="secret-token",
|
|
)
|
|
self.assertEqual(heartbeat_response.status_code, 200)
|
|
heartbeat_body = heartbeat_response.json()
|
|
self.assertEqual(heartbeat_body["agent_version"], "1.1.1")
|
|
self.assertEqual(heartbeat_body["last_seen"], "2026-03-26T12:00:30Z")
|
|
self.assertEqual(heartbeat_body["status"], "online")
|
|
|
|
self.clock.advance(seconds=61)
|
|
timed_out_list = self._request("GET", "/api/clients")
|
|
self.assertEqual(timed_out_list.status_code, 200)
|
|
timed_out_item = timed_out_list.json()["items"][0]
|
|
self.assertEqual(timed_out_item["status"], "offline")
|
|
self.assertEqual(timed_out_item["last_seen"], "2026-03-26T12:00:30Z")
|
|
self.assertIsNone(timed_out_item["last_error"])
|
|
self.assertIsNone(timed_out_item["reachable_at"])
|
|
|
|
def test_register_rejects_invalid_token(self) -> None:
|
|
response = self._request(
|
|
"POST",
|
|
"/api/clients/register",
|
|
self._register_payload(),
|
|
token="wrong-token",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertEqual(response.json()["error"]["code"], "forbidden")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|