awoooi/apps/api/tests/test_km_playbook_feedback_loop.py

"""
KM → Playbook 互饋回路單元測試
================================
W2 PR-L1: 飛輪斷鏈 C3 + C4 修復測試

測試範圍：
  1. test_playbook_promotion_writes_km_entry
     — _promote_playbook 觸發後，KMWriter 被呼叫寫 playbook_evolution 條目
  2. test_playbook_demotion_writes_km_entry
     — _demote_playbook 觸發後，KMWriter 被呼叫寫 playbook_evolution 條目
  3. test_km_accumulation_triggers_playbook_review
     — 同 symptoms_hash 累積 5 條 → UPDATE playbooks.review_required=true
  4. test_km_accumulation_below_threshold_no_update
     — KM 條目 < threshold → 不執行 UPDATE
  5. test_playbook_deprecated_demotes_alert_rule_confidence
     — DEPRECATED Playbook → alert_rule_catalog.confidence *= 0.5
  6. test_feature_flag_disabled
     — ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP=false → 三條邏輯全部跳過，不呼叫 DB

設計原則：
  - 外部服務（DB / KMWriter / PlaybookRepository）以 AsyncMock 替換
  - 每個 test 只測一條主路徑（單一職責）
  - Feature flag 透過 patch 'src.core.config.settings' 控制
  - get_db_context patch 路徑：src.db.base.get_db_context（local import 的來源模組）
  - get_playbook_repository patch 路徑：
      src.repositories.playbook_repository.get_playbook_repository

建立：2026-04-28 (台北時區) ogt + Claude Sonnet 4.6
"""

from __future__ import annotations

from contextlib import asynccontextmanager
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest


# =============================================================================
# Helpers
# =============================================================================

def _make_playbook(
    playbook_id: str = "PB-20260428-AAAAAA",
    name: str = "TestPlaybook",
    trust_score: float = 0.5,
    success_count: int = 3,
    failure_count: int = 1,
    status: str = "approved",
    alert_names: list[str] | None = None,
) -> SimpleNamespace:
    """
    建立一個最小可用的 Playbook mock 物件。

    使用 SimpleNamespace 讓屬性存取與 Pydantic model 相同，
    但不引入真實 ORM / Pydantic 依賴（防止 DB 連線）。
    symptom_pattern.compute_hash() 返回固定 'abc123' 供測試使用。
    """
    symptom = SimpleNamespace(
        alert_names=alert_names or ["HighCpuUsage"],
        affected_services=["api"],
        label_patterns={},
        compute_hash=lambda: "abc123",
    )

    from src.models.playbook import PlaybookStatus
    status_enum = PlaybookStatus(status)

    return SimpleNamespace(
        playbook_id=playbook_id,
        name=name,
        trust_score=trust_score,
        success_count=success_count,
        failure_count=failure_count,
        status=status_enum,
        symptom_pattern=symptom,
    )


def _make_learning_service():
    """
    建立 LearningService 實例，所有外部依賴 mock 掉。
    repository 和 trust_repository 均使用 AsyncMock 防止 Redis 連線。
    """
    from src.services.learning_service import LearningService

    mock_repo = AsyncMock()
    mock_trust_repo = AsyncMock()
    mock_trust_mgr = MagicMock()
    mock_trust_mgr.get_trust_record.return_value = None

    svc = LearningService(
        repository=mock_repo,
        trust_repository=mock_trust_repo,
    )
    svc._trust_manager = mock_trust_mgr
    return svc


def _make_settings(enable_loop: bool = True, threshold: int = 5) -> MagicMock:
    """
    建立 settings mock。
    patch 路徑：src.core.config.settings（learning_service 各方法均 local import 自此模組）
    """
    m = MagicMock()
    m.ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP = enable_loop
    m.KM_PLAYBOOK_REVIEW_THRESHOLD = threshold
    m.KM_WRITE_AWAIT = True
    m.KM_WRITE_TIMEOUT_SECONDS = 5.0
    return m


def _make_db_context_factory(mock_db):
    """
    返回一個可多次呼叫的 async context manager factory。

    每次呼叫 factory() 返回新的 async context manager 實例，
    防止同一 cm 物件被複用（async generator 只能迭代一次）。
    """
    def factory():
        @asynccontextmanager
        async def _ctx():
            yield mock_db
        return _ctx()
    return factory


# =============================================================================
# 1. Promote 觸發 → 寫 KM 演化條目
# =============================================================================

@pytest.mark.asyncio
async def test_playbook_promotion_writes_km_entry():
    """
    _promote_playbook 觸發後，若 ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP=True，
    km_write_with_flag 應被呼叫一次，path_type 含 'playbook_evolution'。
    """
    svc = _make_learning_service()
    playbook = _make_playbook(trust_score=0.5, status="approved")

    km_calls: list = []

    async def _mock_km_write(payload, *, timeout=None):
        km_calls.append(payload)
        from src.services.km_writer import KMWriteResult
        return KMWriteResult.SUCCESS

    mock_pb_repo = AsyncMock()
    mock_pb_repo.find_by_source_incident = AsyncMock(return_value=[playbook])
    mock_pb_repo.adjust_confidence = AsyncMock(return_value=True)

    mock_settings = _make_settings(enable_loop=True)

    with (
        patch("src.core.config.settings", mock_settings),
        patch("src.services.km_writer.km_write_with_flag", side_effect=_mock_km_write),
        patch(
            "src.repositories.playbook_repository.get_playbook_repository",
            return_value=mock_pb_repo,
        ),
    ):
        result = await svc._promote_playbook("INC-TEST-001")

    assert result is True
    assert len(km_calls) == 1, "KMWriter 應被呼叫一次（一個 Playbook promote）"
    assert "playbook_evolution" in km_calls[0].path_type
    assert km_calls[0].metadata["evolution_type"] == "promote"
    assert km_calls[0].metadata["playbook_id"] == playbook.playbook_id
    assert km_calls[0].metadata["previous_trust"] == 0.5


# =============================================================================
# 2. Demote 觸發 → 寫 KM 演化條目
# =============================================================================

@pytest.mark.asyncio
async def test_playbook_demotion_writes_km_entry():
    """
    _demote_playbook 觸發後，若 ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP=True，
    km_write_with_flag 應被呼叫一次，evolution_type='demote'。

    status='approved'（非 DEPRECATED）→ 邏輯 3 不觸發，保持單一職責。
    """
    svc = _make_learning_service()
    playbook = _make_playbook(trust_score=0.4, status="approved")

    km_calls: list = []

    async def _mock_km_write(payload, *, timeout=None):
        km_calls.append(payload)
        from src.services.km_writer import KMWriteResult
        return KMWriteResult.SUCCESS

    mock_pb_repo = AsyncMock()
    mock_pb_repo.find_by_source_incident = AsyncMock(return_value=[playbook])
    mock_pb_repo.adjust_confidence = AsyncMock(return_value=True)

    mock_settings = _make_settings(enable_loop=True)

    with (
        patch("src.core.config.settings", mock_settings),
        patch("src.services.km_writer.km_write_with_flag", side_effect=_mock_km_write),
        patch(
            "src.repositories.playbook_repository.get_playbook_repository",
            return_value=mock_pb_repo,
        ),
    ):
        result = await svc._demote_playbook("INC-TEST-002")

    assert result is True
    assert len(km_calls) == 1, "KMWriter 應被呼叫一次（一個 Playbook demote）"
    assert "playbook_evolution" in km_calls[0].path_type
    assert km_calls[0].metadata["evolution_type"] == "demote"


# =============================================================================
# 3. KM 累積 N=5 → review_required=True
# =============================================================================

@pytest.mark.asyncio
async def test_km_accumulation_triggers_playbook_review():
    """
    同 symptoms_hash 的 KM 條目達到 threshold（預設 5）時，
    _check_and_mark_playbook_review 應執行 COUNT + UPDATE，並 commit。
    """
    svc = _make_learning_service()
    symptoms_hash = "abc123"

    mock_db = AsyncMock()
    execute_call_count = {"n": 0}

    mock_count_result = MagicMock()
    mock_count_result.scalar.return_value = 5

    mock_update_result = MagicMock()
    mock_update_result.fetchall.return_value = [("PB-20260428-AAAAAA",)]

    async def _multi_execute(stmt, params=None):
        execute_call_count["n"] += 1
        if execute_call_count["n"] == 1:
            return mock_count_result
        return mock_update_result

    mock_db.execute = _multi_execute
    mock_db.commit = AsyncMock()

    mock_settings = _make_settings(enable_loop=True, threshold=5)

    with (
        patch("src.core.config.settings", mock_settings),
        patch(
            "src.db.base.get_db_context",
            side_effect=_make_db_context_factory(mock_db),
        ),
    ):
        await svc._check_and_mark_playbook_review(symptoms_hash)

    assert execute_call_count["n"] == 2, "應執行兩次 SQL（COUNT + UPDATE）"
    mock_db.commit.assert_called_once()


@pytest.mark.asyncio
async def test_km_accumulation_below_threshold_no_update():
    """
    KM 條目數 < threshold → 不執行 UPDATE，不 commit。
    """
    svc = _make_learning_service()
    symptoms_hash = "abc123"

    mock_db = AsyncMock()
    execute_call_count = {"n": 0}

    mock_count_result = MagicMock()
    mock_count_result.scalar.return_value = 3  # < 5

    async def _single_execute(stmt, params=None):
        execute_call_count["n"] += 1
        return mock_count_result

    mock_db.execute = _single_execute
    mock_db.commit = AsyncMock()

    mock_settings = _make_settings(enable_loop=True, threshold=5)

    with (
        patch("src.core.config.settings", mock_settings),
        patch(
            "src.db.base.get_db_context",
            side_effect=_make_db_context_factory(mock_db),
        ),
    ):
        await svc._check_and_mark_playbook_review(symptoms_hash)

    assert execute_call_count["n"] == 1, "只執行 COUNT，不執行 UPDATE"
    mock_db.commit.assert_not_called()


# =============================================================================
# 4. DEPRECATED → alert_rule_catalog.confidence *= 0.5
# =============================================================================

@pytest.mark.asyncio
async def test_playbook_deprecated_demotes_alert_rule_confidence():
    """
    DEPRECATED Playbook 的 _demote_alert_rule_catalog_confidence 執行後，
    每個 alert_name 執行一次 UPDATE，最後 commit 一次。
    """
    svc = _make_learning_service()

    from src.models.playbook import PlaybookStatus
    playbook = _make_playbook(
        status="deprecated",
        alert_names=["HighCpuUsage", "PodCrashLooping"],
    )
    playbook.status = PlaybookStatus.DEPRECATED

    mock_db = AsyncMock()
    execute_call_count = {"n": 0}

    async def _track_execute(stmt, params=None):
        execute_call_count["n"] += 1
        m = MagicMock()
        m.rowcount = 1
        return m

    mock_db.execute = _track_execute
    mock_db.commit = AsyncMock()

    mock_settings = _make_settings(enable_loop=True)

    with (
        patch("src.core.config.settings", mock_settings),
        patch(
            "src.db.base.get_db_context",
            side_effect=_make_db_context_factory(mock_db),
        ),
    ):
        await svc._demote_alert_rule_catalog_confidence(playbook)

    assert execute_call_count["n"] == 2, "2 條 alert_names → 2 次 UPDATE"
    mock_db.commit.assert_called_once()


# =============================================================================
# 5. Feature flag disabled → 所有邏輯跳過
# =============================================================================

@pytest.mark.asyncio
async def test_feature_flag_disabled():
    """
    ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP=False 時，
    _write_playbook_evolution_km / _check_and_mark_playbook_review /
    _demote_alert_rule_catalog_confidence 均不應呼叫任何 DB 或 KMWriter。
    """
    svc = _make_learning_service()
    from src.models.playbook import PlaybookStatus
    playbook = _make_playbook(trust_score=0.3, status="deprecated")
    playbook.status = PlaybookStatus.DEPRECATED

    km_write_calls: list = []
    db_execute_calls: list = []

    async def _mock_km_write(payload, *, timeout=None):
        km_write_calls.append(payload)
        from src.services.km_writer import KMWriteResult
        return KMWriteResult.SUCCESS

    mock_db = AsyncMock()

    async def _track_execute(stmt, params=None):
        db_execute_calls.append(stmt)
        return MagicMock()

    mock_db.execute = _track_execute
    mock_db.commit = AsyncMock()

    mock_settings = _make_settings(enable_loop=False)

    with (
        patch("src.core.config.settings", mock_settings),
        patch("src.services.km_writer.km_write_with_flag", side_effect=_mock_km_write),
        patch(
            "src.db.base.get_db_context",
            side_effect=_make_db_context_factory(mock_db),
        ),
    ):
        # 邏輯 1
        await svc._write_playbook_evolution_km(
            playbook=playbook,
            previous_trust=0.5,
            evolution_type="promote",
            incident_id="INC-TEST-FLAG",
        )
        # 邏輯 2
        await svc._check_and_mark_playbook_review("abc123")
        # 邏輯 3
        await svc._demote_alert_rule_catalog_confidence(playbook)

    assert len(km_write_calls) == 0, "KMWriter 不應被呼叫（flag=False）"
    assert len(db_execute_calls) == 0, "DB execute 不應被呼叫（flag=False）"
    mock_db.commit.assert_not_called()