"""Focused regression tests for retired legacy miner-write surfaces.""" import asyncio import inspect import unittest from pathlib import Path from unittest.mock import patch from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from app import main from app.identity import get_current_user, require_permission, required_api_permission APP_ROOT = Path(__file__).parents[1] / "app" class LegacyControlRetirementTests(unittest.TestCase): @staticmethod def assert_retired(callable_, *args, **kwargs): try: result = callable_(*args, **kwargs) if inspect.isawaitable(result): asyncio.run(result) except HTTPException as exc: if exc.status_code != 410: raise AssertionError(f"Expected HTTP 410, got {exc.status_code}") from exc return raise AssertionError("Expected a fixed HTTP 410 retirement response") def test_all_legacy_http_write_routes_are_fixed_inert_responses(self): routes = ( (main.run_dashboard_m30_command_test_lab, {"role": "Admin"}), (main.retired_direct_reboot_miner_command, "miner-1", {"role": "Admin"}), (main.real_power_off_miner_command, "miner-1", {"role": "Admin"}), (main.real_power_on_miner_command, "miner-1", {"role": "Admin"}), (main.real_suspend_miner_command, "miner-1", {"role": "Admin"}), (main.real_resume_miner_command, "miner-1", {"role": "Admin"}), (main.retired_direct_api3_password_change, "miner-1", {"role": "Admin"}), (main.retired_legacy_agent_miner_control, "miner-1", {"role": "Admin"}), (main.process_due_dashboard_scheduled_bulk_actions, {"role": "Admin"}), (main.create_dashboard_scheduled_bulk_action, {"role": "Admin"}), (main.whatsminer_read_test, {"role": "Admin"}), ) with ( patch.object(main, "get_db_connection") as sync_database, patch.object(main, "minerops_command_db_connect") as async_database, patch.object(main, "insert_miner_command_event") as audit_write, patch.object(main, "minerops_send_whatsminer_reboot") as reboot_write, patch.object(main, "minerops_power_command") as power_write, patch.object(main, "minerops_api3_change_password") as password_write, ): for route in routes: with self.subTest(route=route[0].__name__): self.assert_retired(route[0], *route[1:]) for forbidden_call in ( sync_database, async_database, audit_write, reboot_write, power_write, password_write, ): self.assertFalse(forbidden_call.called) def test_m30_requires_control_permission_before_inert_response(self): self.assertEqual( required_api_permission( "POST", "/api/dashboard/m30-command-test-lab/run" ), "miners.control", ) dependency = require_permission("miners.control") with self.assertRaises(HTTPException) as denied: dependency({"role": "Read Only", "permissions": []}) self.assertEqual(denied.exception.status_code, 403) route_source = inspect.getsource(main.run_dashboard_m30_command_test_lab) self.assertIn('Depends(require_permission("miners.control"))', route_source) self.assertIn("status_code=410", route_source) self.assertNotIn("minerops_agent_command_queue", route_source) self.assertNotIn("plexus-windows-agent", route_source) self.assertNotIn("dry_run", route_source) app = FastAPI() app.add_api_route( "/api/dashboard/m30-command-test-lab/run", main.run_dashboard_m30_command_test_lab, methods=["POST"], ) with TestClient(app) as client: app.dependency_overrides[get_current_user] = lambda: { "role": "Read Only", "permissions": [], } self.assertEqual( client.post("/api/dashboard/m30-command-test-lab/run").status_code, 403, ) app.dependency_overrides[get_current_user] = lambda: { "role": "Custom", "permissions": ["miners.control"], } self.assertEqual( client.post("/api/dashboard/m30-command-test-lab/run").status_code, 410, ) def test_registered_retired_routes_have_no_direct_write_implementation(self): retired_paths = { "/api/dashboard/m30-command-test-lab/run", "/api/dashboard/scheduled-bulk-actions/process-due", "/api/dashboard/scheduled-bulk-actions", "/api/tools/whatsminer-read-test", "/api/miners/{miner_id}/commands/reboot", "/api/miners/{miner_id}/commands/power-off", "/api/miners/{miner_id}/commands/power-on", "/api/miners/{miner_id}/commands/suspend", "/api/miners/{miner_id}/commands/resume", "/api/miners/{miner_id}/api3/change-password", "/api/miners/{miner_id}/agent-control", } registered = [ (route.path, route.endpoint) for route in main.app.routes if getattr(route, "path", None) in retired_paths and "POST" in getattr(route, "methods", set()) ] self.assertEqual({path for path, _ in registered}, retired_paths) self.assertEqual(len(registered), len(retired_paths)) forbidden = ( "socket", "minerops_send_", "minerops_power_command", "minerops_api3_change_password", "insert_miner_command_event", "minerops_agent_command_queue", "minerops_command_db_connect", "get_db_connection", "request.json", ) for path, endpoint in registered: source = inspect.getsource(endpoint) with self.subTest(path=path): self.assertIn("status_code=410", source) for unsafe_fragment in forbidden: self.assertNotIn(unsafe_fragment, source) def test_legacy_scheduler_is_noop_and_not_registered_before_claim(self): with patch.object(main, "get_db_connection") as database: self.assertEqual(main.minerops_process_due_scheduled_bulk_actions(), []) self.assertFalse(database.called) self.assertFalse(hasattr(main.app.state, "agent_command_before_claim")) source = inspect.getsource(main.minerops_process_due_scheduled_bulk_actions) self.assertNotIn("minerops_agent_command_queue", source) self.assertNotIn("get_db_connection", source) def test_every_retired_http_route_denies_unprivileged_users_and_never_writes(self): paths = { "/api/dashboard/m30-command-test-lab/run", "/api/dashboard/scheduled-bulk-actions", "/api/dashboard/scheduled-bulk-actions/process-due", "/api/tools/whatsminer-read-test", "/api/miners/{miner_id}/commands/reboot", "/api/miners/{miner_id}/commands/power-off", "/api/miners/{miner_id}/commands/power-on", "/api/miners/{miner_id}/commands/suspend", "/api/miners/{miner_id}/commands/resume", "/api/miners/{miner_id}/api3/change-password", "/api/miners/{miner_id}/agent-control", } app = FastAPI() for route in main.app.routes: if getattr(route, "path", None) in paths and "POST" in route.methods: app.add_api_route(route.path, route.endpoint, methods=["POST"]) with TestClient(app) as client, patch.object(main.socket, "create_connection") as transport, patch.object(main, "get_db_connection") as database: for permissions, expected in (([], 403), (["miners.control", "credentials.manage"], 410)): app.dependency_overrides[get_current_user] = lambda: { "role": "Custom", "permissions": permissions, } for path in paths: concrete = path.replace("{miner_id}", "miner-1") self.assertEqual(client.post(concrete, json={"dry_run": False}).status_code, expected, concrete) transport.assert_not_called() database.assert_not_called() def test_diagnostic_write_disguises_overrides_and_api3_are_retired(self): app = FastAPI() app.add_api_route("/api/tools/whatsminer-read-test", main.whatsminer_read_test, methods=["POST"]) app.dependency_overrides[get_current_user] = lambda: {"role": "Custom", "permissions": ["miners.control"]} attacks = ( {"command": "reboot"}, {"command": "read_then_reboot"}, {"command": "summary", "params": {"command": "reboot"}}, {"command": "summary", "params": {"cmd": "set.miner.service", "param": "stop"}}, {"port": 4433, "command": "set.miner.power_mode", "params": {"param": "high"}}, {"command": "summary\n{\"command\":\"reboot\"}"}, {"command": "summary"}, ) with TestClient(app) as client, patch.object(main.socket, "create_connection") as transport, patch.object(main, "send_whatsminer_read_command") as sender: for attack in attacks: self.assertEqual(client.post("/api/tools/whatsminer-read-test", json={"ip_address": "192.0.2.1", **attack}).status_code, 410) transport.assert_not_called() sender.assert_not_called() def test_internal_discovery_helper_cannot_become_a_generic_command_gateway(self): with patch.object(main.socket, "create_connection") as transport: for command, port, params in ( ("reboot", 4028, None), ("read_reboot", 4028, None), ("summary", 4433, None), ("set.miner.service", 4433, {"param": "stop"}), ("summary", 4028, {"command": "reboot"}), ("summary", 4028, {"cmd": "set.miner.service"}), ("get_miner_info", 4028, {"info": "arbitrary"}), ): self.assert_retired(main.send_whatsminer_read_command, "192.0.2.1", port, command, 500, params) transport.assert_not_called() def test_old_agent_aliases_are_inert_even_without_authentication(self): app = FastAPI() app.add_api_route("/api/agent/commands/pending-disabled-old", main.minerops_agent_commands_pending_retired, methods=["GET"]) app.add_api_route("/api/agent/commands/result-disabled-old", main.minerops_agent_commands_result_retired, methods=["POST"]) with TestClient(app) as client, patch.object(main, "minerops_command_db_connect") as database: self.assertEqual(client.get("/api/agent/commands/pending-disabled-old").status_code, 410) self.assertEqual(client.post("/api/agent/commands/result-disabled-old", json={"status": "completed"}).status_code, 410) database.assert_not_called() def test_canonical_control_still_derives_agent_from_miner_zone(self): repository_source = ( APP_ROOT / "miner_control" / "repository.py" ).read_text() self.assertIn("miners.agent_zone_id", repository_source) self.assertIn( "agents.agent_zone_id = miners.agent_zone_id", repository_source, ) self.assertNotIn("agents.site_id = miners.site_id", repository_source) if __name__ == "__main__": unittest.main()