228 lines
9.4 KiB
Python
228 lines
9.4 KiB
Python
"""Focused service, HTTP, frontend, and security tests for miner control v1."""
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.identity import get_current_user, required_api_permission
|
|
from app.miner_control.dependencies import get_miner_control_service
|
|
from app.miner_control.exceptions import MinerControlConflict, MinerControlUnavailable
|
|
from app.miner_control.models import MinerControlAccepted
|
|
from app.miner_control.router import router
|
|
from app.miner_control.service import MinerControlService
|
|
|
|
|
|
NOW = datetime(2026, 8, 24, tzinfo=timezone.utc)
|
|
MINER_ID = UUID("00000000-0000-4000-8000-000000001501")
|
|
USER_ID = UUID("00000000-0000-4000-8000-000000001502")
|
|
APP_ROOT = Path(__file__).parents[1] / "app"
|
|
|
|
|
|
def context(**overrides):
|
|
result = {
|
|
"id": MINER_ID,
|
|
"site_id": uuid4(),
|
|
"current_ip": "192.0.2.10",
|
|
"runtime_state": "online",
|
|
"configured_performance_mode": "normal",
|
|
"api3_supported": True,
|
|
"api3_api_switch": "on",
|
|
"agent_id": "test-agent",
|
|
"last_checkin_at": NOW,
|
|
"agent_version": "1.0.0-alpha",
|
|
"last_payload": {"control_available": True},
|
|
}
|
|
result.update(overrides)
|
|
return result
|
|
|
|
|
|
class FakeRepository:
|
|
def __init__(self, miner_context=None, history=None, conflict=False):
|
|
self.miner_context = context() if miner_context is None else miner_context
|
|
self.history = history or []
|
|
self.conflict = conflict
|
|
self.enqueue_calls = []
|
|
self.limits = []
|
|
|
|
def get_control_context(self, miner_id):
|
|
self.requested_miner = miner_id
|
|
return self.miner_context
|
|
|
|
def list_recent(self, miner_id, limit=5):
|
|
self.limits.append(limit)
|
|
return self.history
|
|
|
|
def enqueue_with_audit(self, **kwargs):
|
|
self.enqueue_calls.append(kwargs)
|
|
if self.conflict:
|
|
raise MinerControlConflict("live command exists")
|
|
return {
|
|
"operation_id": kwargs["operation_id"],
|
|
"status": "pending",
|
|
"created_at": NOW,
|
|
}
|
|
|
|
|
|
class MinerControlServiceTests(unittest.TestCase):
|
|
def test_supported_projection_is_factual_and_bounded(self):
|
|
repository = FakeRepository()
|
|
result = MinerControlService(repository).get_controls(MINER_ID)
|
|
self.assertEqual(result.service_control.state, "supported")
|
|
self.assertEqual(result.performance_mode, "normal")
|
|
self.assertEqual(result.available_service_action, "suspend")
|
|
self.assertEqual(repository.limits, [5])
|
|
|
|
def test_capability_missing_facts_is_not_optimistic(self):
|
|
cases = (
|
|
({"site_id": None}, "unavailable", "site_unassigned"),
|
|
({"agent_id": None}, "unavailable", "agent_unassigned"),
|
|
({"current_ip": None}, "unavailable", "current_ip_missing"),
|
|
({"api3_supported": False}, "unavailable", "api3_unsupported"),
|
|
({"api3_supported": None}, "unknown", "api3_support_unknown"),
|
|
({"last_payload": {}}, "unknown", "agent_capability_unknown"),
|
|
(
|
|
{"last_payload": {"control_available": False,
|
|
"control_unavailable_reason": "credentials_missing"}},
|
|
"unavailable", "credentials_missing",
|
|
),
|
|
)
|
|
for changes, state, reason in cases:
|
|
with self.subTest(reason=reason):
|
|
result = MinerControlService(FakeRepository(context(**changes))).get_controls(MINER_ID)
|
|
self.assertEqual(result.service_control.state, state)
|
|
self.assertEqual(result.service_control.reason, reason)
|
|
|
|
def test_resume_requires_known_stopped_state(self):
|
|
service = MinerControlService(FakeRepository(context(runtime_state="offline")))
|
|
with self.assertRaisesRegex(MinerControlUnavailable, "factually_stopped"):
|
|
service.queue_control(
|
|
MINER_ID, "resume", requested_value=None,
|
|
actor_user_id=USER_ID, actor_username="operator",
|
|
)
|
|
|
|
def test_queue_contains_routing_but_audit_and_response_do_not(self):
|
|
repository = FakeRepository()
|
|
result = MinerControlService(repository).queue_control(
|
|
MINER_ID, "performance_mode", requested_value="high",
|
|
actor_user_id=USER_ID, actor_username="operator",
|
|
)
|
|
call = repository.enqueue_calls[0]
|
|
self.assertEqual(call["request_payload"]["ip_address"], "192.0.2.10")
|
|
self.assertNotIn("ip_address", call["audit_payload"])
|
|
self.assertNotIn("agent_id", call["audit_payload"])
|
|
self.assertNotIn("password", str(call).lower())
|
|
self.assertNotIn("192.0.2.10", result.model_dump_json())
|
|
|
|
def test_no_projection_changes_when_work_is_queued(self):
|
|
repository = FakeRepository()
|
|
MinerControlService(repository).queue_control(
|
|
MINER_ID, "suspend", requested_value=None,
|
|
actor_user_id=USER_ID, actor_username="operator",
|
|
)
|
|
self.assertFalse(any("update" in key for key in repository.enqueue_calls[0]))
|
|
|
|
|
|
class MinerControlRouterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.app = FastAPI()
|
|
self.app.include_router(router)
|
|
self.repository = FakeRepository()
|
|
self.service = MinerControlService(self.repository)
|
|
self.app.dependency_overrides[get_miner_control_service] = lambda: self.service
|
|
self.client = TestClient(self.app)
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
|
|
def authenticate(self, permissions):
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": str(USER_ID), "username": "operator", "role": "Custom",
|
|
"permissions": list(permissions),
|
|
}
|
|
|
|
def test_get_requires_authentication_and_miners_view(self):
|
|
path = f"/api/miners/{MINER_ID}/controls"
|
|
self.assertEqual(self.client.get(path).status_code, 401)
|
|
self.authenticate(("miners.control",))
|
|
self.assertEqual(self.client.get(path).status_code, 403)
|
|
self.authenticate(("miners.view",))
|
|
response = self.client.get(path)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"]["miner_id"], str(MINER_ID))
|
|
|
|
def test_posts_require_control_and_return_202_bounded_envelope(self):
|
|
path = f"/api/miners/{MINER_ID}/controls/suspend"
|
|
self.authenticate(("miners.view",))
|
|
self.assertEqual(self.client.post(path, json={}).status_code, 403)
|
|
self.authenticate(("miners.control",))
|
|
response = self.client.post(path, json={})
|
|
self.assertEqual(response.status_code, 202)
|
|
self.assertEqual(set(response.json()["data"]), {
|
|
"operation_id", "miner_id", "action", "requested_value", "status", "created_at"
|
|
})
|
|
self.assertNotIn("ip", response.text.lower())
|
|
self.assertNotIn("agent", response.text.lower())
|
|
|
|
def test_strict_bodies_reject_routing_commands_and_credentials(self):
|
|
self.authenticate(("miners.control",))
|
|
base = f"/api/miners/{MINER_ID}/controls"
|
|
for body in (
|
|
{"ip": "192.0.2.1"}, {"agent_id": "other"}, {"dry_run": True},
|
|
{"raw_command": "reboot"}, {"api_password": "secret"},
|
|
):
|
|
with self.subTest(body=body):
|
|
self.assertEqual(self.client.post(base + "/suspend", json=body).status_code, 422)
|
|
self.assertEqual(
|
|
self.client.post(base + "/performance-mode", json={"mode": "turbo"}).status_code,
|
|
422,
|
|
)
|
|
|
|
def test_cross_site_rejected_and_same_origin_allowed(self):
|
|
self.authenticate(("miners.control",))
|
|
path = f"/api/miners/{MINER_ID}/controls/suspend"
|
|
self.assertEqual(
|
|
self.client.post(path, json={}, headers={"Sec-Fetch-Site": "cross-site"}).status_code,
|
|
403,
|
|
)
|
|
self.assertEqual(
|
|
self.client.post(path, json={}, headers={"Origin": "http://testserver"}).status_code,
|
|
202,
|
|
)
|
|
|
|
def test_identity_permission_mapping_is_specific(self):
|
|
base = f"/api/miners/{MINER_ID}/controls"
|
|
self.assertEqual(required_api_permission("GET", base), "miners.view")
|
|
for action in ("suspend", "resume", "performance-mode"):
|
|
self.assertEqual(
|
|
required_api_permission("POST", f"{base}/{action}"),
|
|
"miners.control",
|
|
)
|
|
self.assertNotEqual(
|
|
required_api_permission("POST", base), "miners.control"
|
|
)
|
|
|
|
|
|
class MinerControlFrontendTests(unittest.TestCase):
|
|
def test_controls_are_safe_confirmation_gated_and_non_polling(self):
|
|
source = (APP_ROOT / "frontend" / "static" / "miner-detail.js").read_text()
|
|
view = (APP_ROOT / "frontend" / "miner_detail_view.py").read_text()
|
|
self.assertIn('window.confirm("Confirm: "', source)
|
|
self.assertIn("await loadControls();", source)
|
|
self.assertNotIn("setInterval", source)
|
|
self.assertNotIn("innerHTML", source)
|
|
self.assertIn("minerControlsTitle", view)
|
|
self.assertIn("Your account can inspect control capability", view)
|
|
|
|
def test_new_ui_does_not_use_legacy_control_routes(self):
|
|
source = (APP_ROOT / "frontend" / "static" / "miner-detail.js").read_text()
|
|
for legacy in ("agent-control", "/commands/", "api3", "change-password", "reboot"):
|
|
self.assertNotIn(legacy, source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|