from uuid import uuid4

import pytest
from sqlalchemy import event, select

from modelforge_api.persistence.models import (
    CapabilityContract,
    ProductionExecutionApproval,
    Project,
    ProjectBinding,
    SelectedModelComparison,
)
from modelforge_api.services.lifecycle import LifecycleError
from modelforge_api.services.selected_replacement import SelectedReplacementService
from tests.test_lifecycle_serving import serving_pair

__all__ = ["serving_pair"]


def _preview(pair):
    return SelectedReplacementService(pair[0]).preview(next(iter(pair[4].comparison_ids.values())))


def test_preview_derives_exact_selection_without_writes_or_satisfied_claim(serving_pair):
    session, _, current, candidate, selection, _ = serving_pair
    statements = []

    def observe(_conn, _cursor, statement, _parameters, _context, _many):
        statements.append(statement.lstrip().split(None, 1)[0].upper())

    event.listen(session.get_bind(), "before_cursor_execute", observe)
    try:
        result = _preview(serving_pair)
    finally:
        event.remove(session.get_bind(), "before_cursor_execute", observe)
    assert result.selection == selection
    assert result.route_generation == session.get(CapabilityContract, current.capability_contract_id).serving_generation
    assert result.same_embedding_space and result.affected_bindings[0].eligible
    assert result.production_approvals[0].id == selection.production_approval_id
    assert [item.code for item in result.blockers] == ["PRODUCTION_VALIDATION_REVIEW_REQUIRED"]
    assert result.production_promotion_performed is False
    assert result.model_dump().get("production_validation") is None
    assert not set(statements) & {"INSERT", "UPDATE", "DELETE", "TRUNCATE"}
    assert current.status == "stable" and candidate.production is False
    assert not session.new and not session.dirty and not session.deleted


def test_preview_lists_other_affected_project_as_blocker(serving_pair):
    session, _, current, _, _, _ = serving_pair
    project = Project(key="unreviewed", name="Unreviewed", description="test")
    session.add(project)
    session.flush()
    binding = ProjectBinding(project_id=project.id, capability_contract_id=current.capability_contract_id,
        channel="stable", priority="production")
    session.add(binding)
    session.commit()
    result = _preview(serving_pair)
    assert len(result.affected_bindings) == 2
    assert result.selection is None
    assert any(item.code == "BINDING_COMPARISON_REQUIRED" and item.project_binding_id == binding.id for item in result.blockers)


@pytest.mark.parametrize("mutation,expected", [
    ("incomplete", "COMPARISON_NOT_COMPLETED"),
    ("different-space", "SERVING_REINDEX_REQUIRED"),
    ("revoked", "PRODUCTION_APPROVAL_REQUIRED"),
    ("changed-route", "SERVING_SELECTION_INVALID"),
])
def test_preview_blocks_missing_execution_identity_or_authority(serving_pair, mutation, expected):
    session, serving, current, candidate, selection, _ = serving_pair
    if mutation == "incomplete":
        session.get(SelectedModelComparison, next(iter(selection.comparison_ids.values()))).status = "interrupted"
    elif mutation == "different-space":
        candidate.embedding_space_id = None
    elif mutation == "revoked":
        serving.repo.approval(selection.production_approval_id).status = "revoked"
    else:
        current.status = "draining"
    session.commit()
    result = _preview(serving_pair)
    assert result.selection is None
    assert expected in {item.code for item in result.blockers}


def test_missing_comparison_is_not_found(serving_pair):
    with pytest.raises(LifecycleError) as error:
        SelectedReplacementService(serving_pair[0]).preview(uuid4())
    assert error.value.status_code == 404


def test_latest_negative_project_fit_remains_visible_blocker(serving_pair):
    from tests.test_lifecycle_serving import _fit

    _fit(serving_pair, recommendation="KEEP_LAB", production_validation="DEFERRED_EXTERNAL_VALIDATION")
    result = _preview(serving_pair)
    assert result.selection is None
    assert any(item.code == "SERVING_PROJECT_FIT_BLOCKED" for item in result.blockers)
    assert serving_pair[0].scalar(select(CapabilityContract.serving_generation)) == result.route_generation


def test_stale_historical_approval_does_not_hide_valid_choice(serving_pair):
    session, _, _, _, selection, _ = serving_pair
    valid = session.get(ProductionExecutionApproval, selection.production_approval_id)
    stale = ProductionExecutionApproval(**{
        column.name: getattr(valid, column.name)
        for column in ProductionExecutionApproval.__table__.columns
        if column.name not in {"id", "evidence_fingerprint"}
    }, evidence_fingerprint="f" * 64)
    session.add(stale)
    session.commit()
    result = _preview(serving_pair)
    assert result.selection == selection
    assert [item.id for item in result.production_approvals] == [valid.id]
    assert [item.code for item in result.blockers] == ["PRODUCTION_VALIDATION_REVIEW_REQUIRED"]


def test_actual_app_preview_requires_operator_auth_and_disables_cache(serving_pair, monkeypatch):
    from fastapi.testclient import TestClient
    from pydantic import SecretStr

    from modelforge_api.db import get_session
    from modelforge_api.main import app
    from modelforge_api.settings import Settings, get_settings

    token = "preview-offline-operator-test"  # noqa: S105 - isolated test credential
    configured = Settings(_env_file=None, operator_api_key=SecretStr(token))
    previous = dict(app.dependency_overrides)
    # SQLite fixture connections are thread-bound; qualify the real service above,
    # then exercise middleware, dependencies and response serialization in TestClient.
    result = _preview(serving_pair)
    monkeypatch.setattr(SelectedReplacementService, "preview", lambda self, comparison_id: result)
    app.dependency_overrides[get_session] = lambda: serving_pair[0]
    app.dependency_overrides[get_settings] = lambda: configured
    path = f"/api/v1/selected-model-comparisons/{next(iter(serving_pair[4].comparison_ids.values()))}/replacement-preview"
    try:
        client = TestClient(app)
        assert client.get(path).status_code == 401
        response = client.get(path, headers={"X-ModelForge-Admin-Token": token})
        assert response.status_code == 200, response.text
        assert response.headers["Cache-Control"] == "no-store"
        assert response.json()["selection"]["candidate_deployment_id"] == str(serving_pair[3].id)
        assert response.json()["production_promotion_performed"] is False
    finally:
        app.dependency_overrides.clear()
        app.dependency_overrides.update(previous)