import json from dataclasses import dataclass, field import httpx import jwt from app.core.cache import TTLCache from app.integrations.helixtahr.adapter import ALLOWED_PATH, HelixtaHRAdapter from app.livekit.adapter import InvalidWebhookError, LiveKitUnavailableError, RoomEvent from tests.tokens import HELIXTAHR_ISSUER, HELIXTAHR_SECRET # What the fake expects in the webhook's Authorization header ("signed by LiveKit"). FAKE_WEBHOOK_SIGNATURE = "fake-livekit-signature" @dataclass(frozen=True) class FakePass: room: str identity: str name: str sources: list[str] connect_within_seconds: int attributes: dict[str, str] = field(default_factory=dict) @dataclass(frozen=True) class SentData: room: str # None: sent to everyone in the room. identities: list[str] | None topic: str message: dict[str, object] class FakeLiveKitAdapter: """Stands in for app.livekit.adapter.LiveKitAdapter; no LiveKit server needed.""" def __init__(self) -> None: self.healthy = True self.closed = False # Room name -> the settings it was created with. self.rooms: dict[str, dict[str, int]] = {} self.passes: list[FakePass] = [] self.sent: list[SentData] = [] # (room, identity) -> attributes set after connecting. self.attributes: dict[tuple[str, str], dict[str, str]] = {} # (room, identity) of devices removed. self.removed: list[tuple[str, str]] = [] # Room -> identities LiveKit says are connected (set by tests). self.in_room: dict[str, set[str]] = {} self.deleted_rooms: list[str] = [] # (room, identity) -> sources set after connecting (screen sharing). self.sources: dict[tuple[str, str], list[str]] = {} # (room, identity) of devices whose microphone is sending (set by tests). self.mics_on: set[tuple[str, str]] = set() # (room, identity) of microphones the server muted. self.muted: list[tuple[str, str]] = [] def _check_up(self) -> None: if not self.healthy: raise LiveKitUnavailableError("fake LiveKit is down") async def ping(self) -> None: self._check_up() async def create_room( self, name: str, *, max_participants: int, empty_timeout: int, departure_timeout: int ) -> None: self._check_up() self.rooms.setdefault( name, { "max_participants": max_participants, "empty_timeout": empty_timeout, "departure_timeout": departure_timeout, }, ) def make_pass( self, *, room: str, identity: str, name: str, sources: list[str], connect_within_seconds: int, attributes: dict[str, str] | None = None, ) -> str: self.passes.append( FakePass(room, identity, name, sources, connect_within_seconds, attributes or {}) ) return f"fake-pass-{len(self.passes)}" async def set_attributes(self, room: str, identity: str, attributes: dict[str, str]) -> None: self._check_up() self.attributes.setdefault((room, identity), {}).update(attributes) async def send_data( self, room: str, identities: list[str] | None, topic: str, message: dict[str, object] ) -> None: self._check_up() if identities is None: self.sent.append(SentData(room, None, topic, message)) elif identities: self.sent.append(SentData(room, list(identities), topic, message)) async def set_sources(self, room: str, identity: str, sources: list[str]) -> None: self._check_up() self.sources[(room, identity)] = list(sources) async def delete_room(self, room: str) -> None: self._check_up() self.deleted_rooms.append(room) self.in_room.pop(room, None) async def remove_participant(self, room: str, identity: str) -> None: self._check_up() self.removed.append((room, identity)) self.in_room.get(room, set()).discard(identity) async def mute_microphone(self, room: str, identity: str) -> bool: self._check_up() if (room, identity) not in self.mics_on: return False self.mics_on.discard((room, identity)) self.muted.append((room, identity)) return True async def connected_identities(self, room: str) -> set[str]: self._check_up() return set(self.in_room.get(room, set())) def read_webhook(self, body: str, authorization: str) -> RoomEvent: if authorization != FAKE_WEBHOOK_SIGNATURE: raise InvalidWebhookError("not signed by the fake LiveKit") return RoomEvent(**json.loads(body)) async def aclose(self) -> None: self.closed = True MISSING_ROLE = object() class FakeHelixtaHR: """Answers HelixtaHR's API in-process. Change the fields to change its answers.""" def __init__(self) -> None: self.meetings_allowed: object = True # The user's current role; MISSING_ROLE leaves it out of the answer. self.role_name: object = "manager" self.status_code = 200 self.unreachable = False # Set to return a non-JSON body. self.raw_body: bytes | None = None self.requests: list[httpx.Request] = [] # Users (token `sub`) whose token HelixtaHR now refuses: deactivated, left. self.revoked_users: set[str] = set() def handle(self, request: httpx.Request) -> httpx.Response: self.requests.append(request) if self.unreachable: raise httpx.ConnectError("fake HelixtaHR is down", request=request) if request.url.path != ALLOWED_PATH: return httpx.Response(404) if self._caller(request) in self.revoked_users: return httpx.Response(401, json={"message": "Unauthenticated."}) if self.raw_body is not None: return httpx.Response(self.status_code, content=self.raw_body) body: dict[str, object] = {"meetings_allowed": self.meetings_allowed} if self.role_name is not MISSING_ROLE: body["role_name"] = self.role_name return httpx.Response(self.status_code, json=body) @staticmethod def _caller(request: httpx.Request) -> str | None: token = request.headers.get("Authorization", "").removeprefix("Bearer ") try: return jwt.decode(token, options={"verify_signature": False}).get("sub") except jwt.PyJWTError: return None @property def allowed_calls(self) -> int: return sum(request.url.path == ALLOWED_PATH for request in self.requests) class FakeClock: def __init__(self) -> None: self.now = 1000.0 def __call__(self) -> float: return self.now def make_helixtahr_adapter( fake: FakeHelixtaHR | None = None, clock: FakeClock | None = None ) -> HelixtaHRAdapter: """The real HelixtaHR adapter, with test secrets and a fake HelixtaHR behind it.""" fake = fake or FakeHelixtaHR() http = httpx.AsyncClient( transport=httpx.MockTransport(fake.handle), base_url="https://helixtahr.test" ) cache = TTLCache[bool](60, clock=clock) if clock else None return HelixtaHRAdapter(HELIXTAHR_ISSUER, HELIXTAHR_SECRET, http, cache)