minerops/backend/tests/test_miner_control.py

"""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(),
        "agent_zone_id": uuid4(),
        "agent_zone_exists": True,
        "agent_zone_enabled": True,
        "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_disabled_projection_preserves_facts_and_bounded_history(self):
        repository = FakeRepository()
        result = MinerControlService(repository).get_controls(MINER_ID)
        self.assertEqual(result.service_control.state, "unavailable")
        self.assertEqual(result.service_control.reason, "physical_control_disabled_stage_a")
        self.assertEqual(result.performance_mode, "normal")
        self.assertIsNone(result.available_service_action)
        self.assertEqual(repository.limits, [5])

    def test_capability_cannot_be_enabled_by_agent_claims(self):
        for facts in ({}, {"control_available": True}, {"control_available": False}):
            repository = FakeRepository(context(last_payload=facts))
            result = MinerControlService(repository).get_controls(MINER_ID)
            self.assertEqual(result.service_control.state, "unavailable")
            self.assertEqual(result.service_control.reason, "physical_control_disabled_stage_a")

    def test_resume_is_disabled_even_for_known_stopped_state(self):
        service = MinerControlService(FakeRepository(context(runtime_state="stopped")))
        with self.assertRaisesRegex(MinerControlUnavailable, "physical_control_disabled_stage_a"):
            service.queue_control(MINER_ID, "resume", requested_value=None,
                                  actor_user_id=USER_ID, actor_username="operator")

    def test_all_current_actions_are_rejected_before_routing_or_queue_write(self):
        for action in ("suspend", "resume", "performance_mode"):
            repository = FakeRepository()
            with self.assertRaisesRegex(MinerControlUnavailable, "physical_control_disabled_stage_a"):
                MinerControlService(repository).queue_control(
                    MINER_ID, action, requested_value="high",
                    actor_user_id=USER_ID, actor_username="operator")
            self.assertEqual(repository.enqueue_calls, [])
            self.assertFalse(hasattr(repository, "requested_miner"))

    def test_rejected_request_leaves_context_unchanged(self):
        repository = FakeRepository()
        before = repository.miner_context.copy()
        with self.assertRaises(MinerControlUnavailable):
            MinerControlService(repository).queue_control(
                MINER_ID, "suspend", requested_value=None,
                actor_user_id=USER_ID, actor_username="operator")
        self.assertEqual(repository.miner_context, before)
        self.assertEqual(repository.enqueue_calls, [])


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_then_fail_closed(self):
        for action, body in (("suspend", {}), ("resume", {}), ("performance-mode", {"mode": "high"})):
            path = f"/api/miners/{MINER_ID}/controls/{action}"
            self.authenticate(("miners.view",))
            self.assertEqual(self.client.post(path, json=body).status_code, 403)
            self.authenticate(("miners.control",))
            response = self.client.post(path, json=body)
            self.assertEqual(response.status_code, 409)
            self.assertIn("physical_control_disabled_stage_a", response.text)
        self.assertEqual(self.repository.enqueue_calls, [])

    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,
            409,
        )

    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()