87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
"""A real local aiohttp server plays the lamp; no Home Assistant instance involved."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import AsyncGenerator, Awaitable, Callable
|
|
from dataclasses import dataclass, field
|
|
import json
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
from aiohttp import web
|
|
from aiohttp.test_utils import TestServer
|
|
import pytest
|
|
|
|
|
|
@dataclass
|
|
class Call:
|
|
"""One request the fake lamp received."""
|
|
|
|
method: str
|
|
path: str
|
|
headers: dict[str, str]
|
|
body: Any
|
|
|
|
|
|
@dataclass
|
|
class FakeLamp:
|
|
"""Canned responses per (METHOD, path); records every call."""
|
|
|
|
host: str
|
|
port: int
|
|
routes: dict[tuple[str, str], tuple[int, str, float]] = field(default_factory=dict)
|
|
calls: list[Call] = field(default_factory=list)
|
|
|
|
def reply(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
status: int = 200,
|
|
payload: Any = None,
|
|
*,
|
|
raw: str | None = None,
|
|
delay: float = 0.0,
|
|
) -> None:
|
|
body = raw if raw is not None else json.dumps(payload)
|
|
self.routes[(method.upper(), path)] = (status, body, delay)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _enable_sockets(socket_enabled: None) -> None:
|
|
"""pytest-homeassistant-custom-component blocks sockets; these tests need localhost."""
|
|
|
|
|
|
@pytest.fixture
|
|
async def lamp(aiohttp_server: Callable[..., Awaitable[TestServer]]) -> FakeLamp:
|
|
holder: list[FakeLamp] = []
|
|
|
|
async def handler(request: web.Request) -> web.Response:
|
|
fake = holder[0]
|
|
text = await request.text()
|
|
try:
|
|
body: Any = json.loads(text) if text else None
|
|
except ValueError:
|
|
body = text
|
|
fake.calls.append(Call(request.method, request.path, dict(request.headers), body))
|
|
route = fake.routes.get((request.method, request.path))
|
|
if route is None:
|
|
return web.Response(status=404, text="no route")
|
|
status, text_out, delay = route
|
|
if delay:
|
|
await asyncio.sleep(delay)
|
|
return web.Response(status=status, text=text_out)
|
|
|
|
app = web.Application()
|
|
app.router.add_route("*", "/{tail:.*}", handler)
|
|
server = await aiohttp_server(app)
|
|
fake = FakeLamp(server.host, server.port)
|
|
holder.append(fake)
|
|
return fake
|
|
|
|
|
|
@pytest.fixture
|
|
async def session() -> AsyncGenerator[aiohttp.ClientSession]:
|
|
async with aiohttp.ClientSession() as client_session:
|
|
yield client_session
|