106 lines
4.1 KiB
Python
106 lines
4.1 KiB
Python
"""WP-100 Unraid connector contract tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
from connectors.unraid import FakeBrokerClient, RegisteredContainer, UnraidConnector
|
|
from connectors.unraid.client import BrokerClientError, FileBrokerTokenProvider
|
|
from packages.connector_kit import ConnectorContext, PageRequest
|
|
from services.docker_read_broker import ServiceTokenVerifier
|
|
|
|
PROJECT_ID = UUID("33333333-3333-3333-3333-333333333333")
|
|
|
|
|
|
def make_connector(fail: bool = False) -> UnraidConnector:
|
|
return UnraidConnector(
|
|
FakeBrokerClient(fail=fail), (RegisteredContainer("ludarium-web", PROJECT_ID, "prod"),)
|
|
)
|
|
|
|
|
|
def make_context(project_id: UUID = PROJECT_ID, environment: str = "prod") -> ConnectorContext:
|
|
return ConnectorContext(
|
|
workspace_id=UUID("44444444-4444-4444-4444-444444444444"),
|
|
project_id=project_id,
|
|
project_slug="ludarium",
|
|
environment=environment,
|
|
)
|
|
|
|
|
|
class UnraidConnectorTests(unittest.TestCase):
|
|
def test_file_provider_issues_short_lived_project_scoped_broker_token(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "broker-signing-key"
|
|
secret = "broker-signing-key-with-at-least-thirty-two-bytes" # noqa: S105
|
|
path.write_text(secret, encoding="utf-8")
|
|
os.chmod(path, 0o600)
|
|
|
|
token = FileBrokerTokenProvider(str(path)).get_token(make_context())
|
|
|
|
self.assertNotEqual(token, secret)
|
|
self.assertEqual(ServiceTokenVerifier(secret).verify(token), ("ludarium", "prod"))
|
|
|
|
def test_broker_backed_read_tools_and_source_metadata(self) -> None:
|
|
connector = make_connector()
|
|
context = make_context()
|
|
self.assertEqual(
|
|
connector.host_status(context)["meta"]["source"], "unraid:docker-read-broker"
|
|
)
|
|
self.assertEqual(
|
|
connector.container_list(context)["data"]["items"][0]["containerRef"], "ludarium-web"
|
|
)
|
|
self.assertEqual(
|
|
connector.container_inspect(context, "ludarium-web")["data"]["health"], "healthy"
|
|
)
|
|
self.assertEqual(
|
|
connector.container_health(context, "ludarium-web")["data"]["status"], "healthy"
|
|
)
|
|
self.assertEqual(
|
|
connector.container_logs(context, "ludarium-web")["data"]["lines"], ["ready"]
|
|
)
|
|
self.assertEqual(
|
|
connector.network_list(context)["data"]["items"][0]["name"], "hub-connectors"
|
|
)
|
|
self.assertEqual(
|
|
connector.storage_list(context, page=PageRequest(limit=1))["data"]["items"][0]["name"],
|
|
"appdata",
|
|
)
|
|
summary = connector.project_summary(context)["data"]
|
|
self.assertEqual(summary["status"], "healthy")
|
|
self.assertEqual(summary["expectedContainers"], 1)
|
|
self.assertEqual(summary["runningContainers"], 1)
|
|
|
|
def test_cross_project_and_unknown_container_are_denied(self) -> None:
|
|
connector = make_connector()
|
|
with self.assertRaises(BrokerClientError):
|
|
connector.container_inspect(
|
|
make_context(UUID("55555555-5555-5555-5555-555555555555")), "ludarium-web"
|
|
)
|
|
with self.assertRaises(BrokerClientError):
|
|
connector.container_logs(make_context(), "unknown")
|
|
|
|
def test_log_limits_and_publication_are_read_only(self) -> None:
|
|
connector = make_connector()
|
|
with self.assertRaises(ValueError):
|
|
connector.container_logs(make_context(), "ludarium-web", tail=501)
|
|
self.assertEqual(len(connector.tools()), 8)
|
|
self.assertTrue(all(tool.read_only for tool in connector.tools()))
|
|
self.assertEqual(
|
|
connector.resources()[0].uri_template,
|
|
"itworx-unraid://{project}/{environment}/container/{containerRef}",
|
|
)
|
|
|
|
def test_broker_failure_is_scoped_to_unraid(self) -> None:
|
|
connector = make_connector(fail=True)
|
|
with self.assertRaises(RuntimeError):
|
|
connector.host_status(make_context())
|
|
self.assertEqual(connector.readiness()["status"], "ready")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|