fix(cd): isolate learning-chain async tests
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 3m47s
CD Pipeline / build-and-deploy (push) Successful in 5m7s
CD Pipeline / post-deploy-checks (push) Has been cancelled

This commit is contained in:
ogt
2026-07-10 12:03:51 +08:00
parent 836d614149
commit 964859292f
2 changed files with 71 additions and 1 deletions

View File

@@ -113,6 +113,26 @@ class StubPlaybookService:
return playbook is not None
class StubRunbookGenerator:
"""No-network/no-DB runbook generator for auto-repair fire-and-forget tasks."""
async def generate_runbook(self, *args, **kwargs) -> None:
return None
async def generate_anti_pattern(self, *args, **kwargs) -> None:
return None
class StubAutoRepairExecutionRepository:
async def create(self, *args, **kwargs) -> dict:
return {"stored": False}
class StubAnomalyCounter:
async def record_disposition(self, *args, **kwargs) -> bool:
return True
class StubRecommendation:
def __init__(self, playbook: Playbook, similarity_score: float = 0.9) -> None:
self.playbook = playbook
@@ -179,6 +199,40 @@ async def _no_cooldown(*args, **kwargs) -> tuple[bool, str]:
return True, "允許修復 (test bypass)"
@pytest.fixture(autouse=True)
def isolate_auto_repair_side_effects(monkeypatch):
"""Keep this E2E chain in-process only: no DB, network, KM, or runbook writes."""
async def _record_global_repair_action() -> None:
return None
import src.services.auto_repair_service as _ars_mod
monkeypatch.setattr(_ars_mod, "record_global_repair_action", _record_global_repair_action)
import src.services.runbook_generator as _rg_mod
monkeypatch.setattr(_rg_mod, "get_runbook_generator", lambda: StubRunbookGenerator())
import src.repositories.audit_log_repository as _audit_mod
monkeypatch.setattr(
_audit_mod,
"get_auto_repair_execution_repository",
lambda: StubAutoRepairExecutionRepository(),
)
import src.services.anomaly_counter as _counter_mod
monkeypatch.setattr(_counter_mod, "get_anomaly_counter", lambda: StubAnomalyCounter())
async def _drain_service_tasks(service: AutoRepairService) -> None:
drain = await service.drain_pending_tasks(timeout=1.0)
assert drain.get("timeout") is False, drain
assert not service._pending_tasks
# =============================================================================
# Tests
# =============================================================================
@@ -234,6 +288,7 @@ async def test_auto_repair_success_triggers_verify_and_learn(monkeypatch):
assert call["action_taken"] == stub_verifier.calls[0]["action_taken"]
assert call["verification_result"] == "success"
assert call["matched_playbook_id"] == playbook.playbook_id
await _drain_service_tasks(service)
@pytest.mark.asyncio
@@ -273,6 +328,7 @@ async def test_auto_repair_can_delegate_post_verification(monkeypatch):
assert stub_verifier.calls == []
assert stub_learning.verification_calls == []
await _drain_service_tasks(service)
@pytest.mark.asyncio
@@ -321,6 +377,7 @@ async def test_auto_repair_failure_does_not_call_verifier(monkeypatch):
# 失敗路徑不進入 verify-and-learn 塊
assert len(stub_verifier.calls) == 0, "執行失敗時不應呼叫 verifier"
assert len(stub_learning.verification_calls) == 0, "執行失敗時不應呼叫 record_verification_result"
await _drain_service_tasks(service)
@pytest.mark.asyncio
@@ -355,6 +412,7 @@ async def test_auto_repair_failed_step_string_marks_execution_failure(monkeypatc
assert "simulated executor failure" in (result.error or "")
assert len(stub_verifier.calls) == 0
assert len(stub_learning.verification_calls) == 0
await _drain_service_tasks(service)
@pytest.mark.asyncio
@@ -447,3 +505,4 @@ async def test_verifier_exception_does_not_block_repair(monkeypatch):
assert len(stub_verifier.calls) == 1
# learning 不應被呼叫(因為 verifier raise 中斷了 _verify_and_learn
assert len(stub_learning.verification_calls) == 0, "verifier 拋例外後 learning 不應被呼叫"
await _drain_service_tasks(service)