96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
"""Deterministic broker client fake for Unraid connector tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from packages.connector_kit import ConnectorContext
|
|
|
|
|
|
class FakeBrokerClient:
|
|
def __init__(self, *, fail: bool = False) -> None:
|
|
self.fail = fail
|
|
self.calls: list[str] = []
|
|
|
|
def _check(self, operation: str) -> None:
|
|
self.calls.append(operation)
|
|
if self.fail:
|
|
raise RuntimeError("broker unavailable")
|
|
|
|
def host(self, context: ConnectorContext) -> dict[str, Any]:
|
|
self._check("host")
|
|
return {
|
|
"hostRef": "unraid",
|
|
"hostname": "unraid-test",
|
|
"uptimeSeconds": 3600,
|
|
"cpuPercent": 12.5,
|
|
"memoryUsedBytes": 100,
|
|
"memoryTotalBytes": 1000,
|
|
"dockerVersion": "27.0.0",
|
|
}
|
|
|
|
def storage(self, context: ConnectorContext) -> dict[str, Any]:
|
|
self._check("storage")
|
|
return {
|
|
"items": [
|
|
{"name": "appdata", "driver": "local", "usageBytes": 1024, "capacityBytes": 4096}
|
|
]
|
|
}
|
|
|
|
def containers(self, context: ConnectorContext, state: str | None = None) -> dict[str, Any]:
|
|
self._check("containers")
|
|
return {
|
|
"items": [
|
|
{
|
|
"containerRef": "ludarium-web",
|
|
"name": "ludarium-web",
|
|
"image": "web:1",
|
|
"state": state or "running",
|
|
"health": "healthy",
|
|
"startedAt": "2026-08-01T10:00:00Z",
|
|
"restartCount": 2,
|
|
"ports": [],
|
|
"networks": ["hub-connectors"],
|
|
"labels": {"com.docker.compose.project": "ludarium"},
|
|
}
|
|
]
|
|
}
|
|
|
|
def inspect(self, context: ConnectorContext, registered_id: str) -> dict[str, Any]:
|
|
self._check("inspect")
|
|
return dict(self.containers(context)["items"][0]) | {"containerRef": registered_id}
|
|
|
|
def health(self, context: ConnectorContext, registered_id: str) -> dict[str, Any]:
|
|
self._check("health")
|
|
return {"status": "healthy", "running": True, "restartCount": 2}
|
|
|
|
def logs(
|
|
self,
|
|
context: ConnectorContext,
|
|
registered_id: str,
|
|
*,
|
|
tail: int,
|
|
since_seconds: int | None,
|
|
contains: str | None,
|
|
) -> dict[str, Any]:
|
|
self._check("logs")
|
|
return {
|
|
"containerRef": registered_id,
|
|
"lines": ["ready"],
|
|
"truncated": False,
|
|
"redactionsApplied": 0,
|
|
}
|
|
|
|
def networks(self, context: ConnectorContext) -> dict[str, Any]:
|
|
self._check("networks")
|
|
return {
|
|
"items": [
|
|
{
|
|
"name": "hub-connectors",
|
|
"networkRef": "network-1",
|
|
"driver": "bridge",
|
|
"scope": "local",
|
|
}
|
|
]
|
|
}
|