feat(api): HostRepairAgent 三條執行路徑 + known_hosts + Ansible 白名單 (Sprint 3 T3)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/api/tests/test_host_repair_agent.py

@@ -209,3 +209,65 @@ class TestValidateShellSafety:
         from src.services.host_repair_agent import validate_shell_safety
         with pytest.raises(ValueError, match="too long"):
             validate_shell_safety("a" * 513)
+
+
+import os
+from unittest.mock import patch, AsyncMock
+
+
+class TestAnsibleWhitelist:
+    def test_allowed_playbook_passes(self):
+        from src.services.host_repair_agent import validate_ansible_playbook
+        with patch.dict(os.environ, {"ANSIBLE_PLAYBOOK_WHITELIST": "vacuum_postgres.yml,clear_redis_cache.yml"}):
+            validate_ansible_playbook("vacuum_postgres.yml")  # must not raise
+
+    def test_disallowed_playbook_raises(self):
+        from src.services.host_repair_agent import validate_ansible_playbook
+        with patch.dict(os.environ, {"ANSIBLE_PLAYBOOK_WHITELIST": "vacuum_postgres.yml"}):
+            with pytest.raises(ValueError, match="not in allowed whitelist"):
+                validate_ansible_playbook("evil_script.sh")
+
+    def test_path_traversal_blocked(self):
+        from src.services.host_repair_agent import validate_ansible_playbook
+        with patch.dict(os.environ, {"ANSIBLE_PLAYBOOK_WHITELIST": "vacuum_postgres.yml"}):
+            with pytest.raises(ValueError, match="not in allowed whitelist"):
+                validate_ansible_playbook("../../../etc/passwd")
+
+
+class TestRepairByUri:
+    @pytest.mark.asyncio
+    async def test_openclaw_scheme_calls_repair(self):
+        from src.services.host_repair_agent import HostRepairAgent, HostRepairResult
+        agent = HostRepairAgent()
+        with patch.object(agent, "_execute_openclaw", new_callable=AsyncMock) as mock_oc:
+            mock_oc.return_value = HostRepairResult(success=True, layer="docker-110", component="sentry", output="REPAIR_OK:sentry")
+            result = await agent.repair_by_uri("openclaw://docker-110/sentry")
+        assert result.success is True
+        mock_oc.assert_awaited_once_with("docker-110", "sentry")
+
+    @pytest.mark.asyncio
+    async def test_ansible_scheme_calls_ansible(self):
+        from src.services.host_repair_agent import HostRepairAgent, HostRepairResult
+        agent = HostRepairAgent()
+        with patch.object(agent, "_execute_ansible", new_callable=AsyncMock) as mock_ans, \
+             patch.dict(os.environ, {"ANSIBLE_PLAYBOOK_WHITELIST": "vacuum_postgres.yml"}):
+            mock_ans.return_value = HostRepairResult(success=True, layer="ansible", component="vacuum_postgres.yml", output="REPAIR_OK:ansible")
+            result = await agent.repair_by_uri("ansible://192.168.0.188/vacuum_postgres.yml")
+        assert result.success is True
+        mock_ans.assert_awaited_once_with("192.168.0.188", "vacuum_postgres.yml")
+
+    @pytest.mark.asyncio
+    async def test_ssh_scheme_blocked_without_approval_flag(self):
+        from src.services.host_repair_agent import HostRepairAgent
+        agent = HostRepairAgent()
+        result = await agent.repair_by_uri("ssh://wooo@192.168.0.110/docker ps")
+        assert result.success is False
+        assert "requires_approval" in result.error
+
+    @pytest.mark.asyncio
+    async def test_invalid_uri_returns_failure(self):
+        from src.services.host_repair_agent import HostRepairAgent
+        agent = HostRepairAgent()
+        result = await agent.repair_by_uri("bad-format")
+        assert result.success is False
+        assert "Unsupported scheme" in result.error