"""partial cascade contract (problems.md #11) — manifest allow_partial_downstream.

emitter (dep step) 가 partial 결과 emit 시 본 step 진행 여부를 dep 의 manifest
``allow_partial_downstream`` 에 따라 결정. default True (backward-compat).
False 명시 시 cascade stop.

검증 대상:
  - StepRunner.check_gate (gate-level)
  - analysis_dispatch_service.run_steps_batch (batch-level)
"""
from __future__ import annotations

import pytest


def _step_run(status: str, completed: int = 0, applicable: int = 0) -> dict:
    """Block B T0 (plan v2.1.3): _get_step_run dict shape helper.

    check_gate 는 ``dep_run["status"]`` 만 보지만 schema 일관성을 위해
    7 필수 키 모두 포함시킨다.
    """
    return {
        "status": status,
        "run_id": "r-test",
        "started_at": None,
        "completed_count": completed,
        "applicable_count": applicable,
        "recovery_count": 0,
        "updated_at": None,
    }


# ---------------------------------------------------------------------------
# StepRunner.check_gate — partial dep + dep contract
# ---------------------------------------------------------------------------


def test_check_gate_passes_partial_dep_default_backward_compat(monkeypatch):
    """dep status='partial' + dep manifest allow_partial_downstream 미지정 → 통과 (backward-compat)."""
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream_step"
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on",
        lambda sid: ["upstream_step"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {"label": "upstream"},  # allow_partial_downstream 미지정 → default True
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("partial", 0, 0)
    )
    # blocked AppError 없으면 OK.
    runner.check_gate()


def test_check_gate_blocks_partial_dep_when_strict(monkeypatch):
    """dep partial + dep allow_partial_downstream=False → AppError(gate.blocked)."""
    from app.core.errors import AppError
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream_step"
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on",
        lambda sid: ["upstream_strict"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {"label": "strict-upstream", "allow_partial_downstream": False},
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("partial", 1, 5)
    )

    with pytest.raises(AppError) as excinfo:
        runner.check_gate()
    assert excinfo.value.code == "gate.blocked"
    assert "strict" in excinfo.value.message or "strict-upstream" in excinfo.value.message


def test_check_gate_passes_completed_dep_regardless_of_contract(monkeypatch):
    """completed dep 는 contract 무관 통과 (status가 partial 이 아니므로)."""
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on", lambda sid: ["up"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {"label": "up", "allow_partial_downstream": False},
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("completed", 5, 5),
    )
    runner.check_gate()


def test_check_gate_blocks_failed_dep_unconditionally(monkeypatch):
    """status='failed' 는 contract 무관 항상 blocked (기존 동작)."""
    from app.core.errors import AppError
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on", lambda sid: ["up"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {"label": "up", "allow_partial_downstream": True},
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("failed", 0, 5),
    )

    with pytest.raises(AppError) as excinfo:
        runner.check_gate()
    assert excinfo.value.code == "gate.blocked"


def test_check_gate_passes_not_applicable_dep(monkeypatch):
    """status='not_applicable' 은 통과 (기존 동작 보존 회귀 가드)."""
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    monkeypatch.setattr("app.core.step_runner.get_depends_on", lambda sid: ["up"])
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict", lambda sid: {"label": "up"},
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("not_applicable", 0, 0),
    )
    runner.check_gate()  # no raise


def test_check_gate_blocks_unknown_status(monkeypatch):
    """알 수 없는 status (예: 'cancelled', 'running') 는 blocked."""
    from app.core.errors import AppError
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    monkeypatch.setattr("app.core.step_runner.get_depends_on", lambda sid: ["up"])
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict", lambda sid: {"label": "up"},
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("cancelled", 0, 0),
    )
    with pytest.raises(AppError) as excinfo:
        runner.check_gate()
    assert excinfo.value.code == "gate.blocked"


# ---------------------------------------------------------------------------
# Operator override — partial_override_config_key (G4.6 Phase 3 fix iter 2)
# ---------------------------------------------------------------------------


def test_check_gate_passes_partial_strict_dep_when_operator_override(monkeypatch, caplog):
    """dep manifest 의 partial_override_config_key 가 project_config 에서 truthy → 통과."""
    import logging
    caplog.set_level(logging.WARNING)
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    runner.project_config = {"allow_failed_validator": True}  # operator override
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on", lambda sid: ["upstream_strict"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {
            "label": "strict-up",
            "allow_partial_downstream": False,
            "partial_override_config_key": "allow_failed_validator",
        },
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("partial", 1, 5),
    )

    runner.check_gate()  # 통과 — AppError 없음
    assert any(
        "explicit operator override" in rec.message
        for rec in caplog.records
    )


def test_check_gate_blocks_partial_strict_dep_when_override_key_missing(monkeypatch):
    """partial_override_config_key 명시되어도 project_config 에 키 없으면 차단."""
    from app.core.errors import AppError
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    runner.project_config = {}  # 키 미설정
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on", lambda sid: ["upstream_strict"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {
            "label": "strict-up",
            "allow_partial_downstream": False,
            "partial_override_config_key": "allow_failed_validator",
        },
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("partial", 1, 5),
    )

    with pytest.raises(AppError) as excinfo:
        runner.check_gate()
    assert excinfo.value.code == "gate.blocked"


def test_check_gate_blocks_partial_strict_dep_when_override_key_false(monkeypatch):
    """override key 가 project_config 에 있어도 falsy 값이면 차단."""
    from app.core.errors import AppError
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    runner.project_config = {"allow_failed_validator": False}
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on", lambda sid: ["upstream_strict"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {
            "label": "strict-up",
            "allow_partial_downstream": False,
            "partial_override_config_key": "allow_failed_validator",
        },
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("partial", 1, 5),
    )

    with pytest.raises(AppError) as excinfo:
        runner.check_gate()
    assert excinfo.value.code == "gate.blocked"


def test_check_gate_strict_partial_dep_rejects_truthy_string_override(monkeypatch):
    """Codex iter 3 I1 — 'true' / '1' / 1 같은 truthy 표현은 명시 bool override X → 차단.

    오직 ``project_config[key] is True`` 만 통과. 운영자 명시 결정 기록.
    """
    from app.core.errors import AppError
    from app.core.step_runner import StepRunner

    for non_strict in ["true", "True", "1", 1, 1.0, "yes", "on"]:
        runner = StepRunner.__new__(StepRunner)
        runner.step_id = "downstream"
        runner.project_config = {"allow_failed_validator": non_strict}
        monkeypatch.setattr(
            "app.core.step_runner.get_depends_on", lambda sid: ["upstream_strict"],
        )
        monkeypatch.setattr(
            "app.core.step_runner.get_manifest_dict",
            lambda sid: {
                "label": "strict-up",
                "allow_partial_downstream": False,
                "partial_override_config_key": "allow_failed_validator",
            },
        )
        monkeypatch.setattr(
            runner.__class__, "_get_step_run", lambda self, dep: _step_run("partial", 1, 5),
        )

        with pytest.raises(AppError) as excinfo:
            runner.check_gate()
        assert excinfo.value.code == "gate.blocked", (
            f"non-strict value {non_strict!r} should not unlock override"
        )


def test_check_gate_strict_partial_dep_without_override_key_still_blocks(monkeypatch):
    """partial_override_config_key 미명시인 strict dep — project_config 무관 차단 (회귀 가드)."""
    from app.core.errors import AppError
    from app.core.step_runner import StepRunner

    runner = StepRunner.__new__(StepRunner)
    runner.step_id = "downstream"
    runner.project_config = {"allow_failed_validator": True}
    monkeypatch.setattr(
        "app.core.step_runner.get_depends_on", lambda sid: ["upstream_strict"],
    )
    monkeypatch.setattr(
        "app.core.step_runner.get_manifest_dict",
        lambda sid: {
            "label": "strict-up",
            "allow_partial_downstream": False,
            # partial_override_config_key 미명시
        },
    )
    monkeypatch.setattr(
        runner.__class__, "_get_step_run", lambda self, dep: _step_run("partial", 1, 5),
    )

    with pytest.raises(AppError) as excinfo:
        runner.check_gate()
    assert excinfo.value.code == "gate.blocked"