# #region DatasetReviewApiTests [TYPE Module] [C:3] [SEMANTICS dataset_review, api, tests, lifecycle, exports, orchestration]
# @BRIEF Verify backend US1 dataset review lifecycle, export, parsing, and dictionary-resolution contracts.
# @LAYER API
# @RELATION BINDS_TO ->[DatasetReviewApi]
# @RELATION BINDS_TO ->[DatasetReviewOrchestrator]
from datetime import UTC, datetime
import pytest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

from fastapi.testclient import TestClient

from src.api.routes.dataset_review import (
    _get_orchestrator,
    _get_repository,
)
from src.app import app
from src.core.config_models import AppConfig, Environment, GlobalSettings
from src.core.utils.superset_context_extractor import SupersetContextExtractor
from src.dependencies import get_config_manager, get_current_user, get_task_manager
from src.models.dataset_review import (
    ApprovalState,
    BusinessSummarySource,
    CandidateMatchType,
    CandidateStatus,
    ClarificationOption,
    ClarificationQuestion,
    ClarificationSession,
    ClarificationStatus,
    CompiledPreview,
    ConfidenceState,
    DatasetReviewSession,
    ExecutionMapping,
    FieldKind,
    FieldProvenance,
    FindingArea,
    FindingSeverity,
    LaunchStatus,
    MappingMethod,
    PreviewStatus,
    QuestionState,
    ReadinessState,
    RecommendedAction,
    ResolutionState,
    SemanticCandidate,
    SemanticFieldEntry,
    SemanticSource,
    SemanticSourceStatus,
    SemanticSourceType,
    SessionPhase,
    SessionStatus,
    TrustLevel,
)
from src.services.dataset_review.event_logger import SessionEventLogger
from src.services.dataset_review.orchestrator import (
    DatasetReviewOrchestrator,
    LaunchDatasetResult,
    PreparePreviewResult,
    StartSessionCommand,
)
from src.services.dataset_review.orchestrator_pkg._helpers import (
    build_execution_snapshot,
)
from src.services.dataset_review.repositories.session_repository import (
    DatasetReviewSessionVersionConflictError,
)
from src.services.dataset_review.semantic_resolver import SemanticSourceResolver

client = TestClient(app)
# #region _make_user [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
def _make_user():
    permissions = [
        SimpleNamespace(resource="dataset:session", action="READ"),
        SimpleNamespace(resource="dataset:session", action="MANAGE"),
        SimpleNamespace(resource="dataset:execution:launch", action="EXECUTE"),
    ]
    dataset_review_role = SimpleNamespace(
        name="DatasetReviewOperator", permissions=permissions
    )
    return SimpleNamespace(id="user-1", username="tester", roles=[dataset_review_role])
# #endregion _make_user
# #region _make_config_manager [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
def _make_config_manager():
    env = Environment(
        id="env-1",
        name="DEV",
        url="http://superset.local",
        username="demo",
        password="secret",
    )
    config = AppConfig(environments=[env], settings=GlobalSettings())
    manager = MagicMock()
    manager.get_environment.side_effect = (
        lambda env_id: env if env_id == "env-1" else None
    )
    manager.get_config.return_value = config
    return manager
# #endregion _make_config_manager
# #region _make_session [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
def _make_session():
    now = datetime.now(UTC)
    return DatasetReviewSession(
        session_id="sess-1",
        user_id="user-1",
        environment_id="env-1",
        source_kind="superset_link",
        source_input="http://superset.local/dashboard/10",
        dataset_ref="public.sales",
        dataset_id=42,
        dashboard_id=10,
        version=0,
        readiness_state=ReadinessState.REVIEW_READY,
        recommended_action=RecommendedAction.REVIEW_DOCUMENTATION,
        status=SessionStatus.ACTIVE,
        current_phase=SessionPhase.REVIEW,
        created_at=now,
        updated_at=now,
        last_activity_at=now,
    )
# #endregion _make_session
# #region _make_us2_session [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
def _make_us2_session():
    now = datetime.now(UTC)
    session = _make_session()
    session.readiness_state = ReadinessState.CLARIFICATION_NEEDED
    session.recommended_action = RecommendedAction.START_CLARIFICATION
    session.current_phase = SessionPhase.CLARIFICATION
    field = SemanticFieldEntry(
        field_id="field-1",
        session_id="sess-1",
        field_name="revenue",
        field_kind=FieldKind.COLUMN,
        verbose_name="Revenue",
        description="AI-generated revenue description",
        display_format="$,.2f",
        provenance=FieldProvenance.AI_GENERATED,
        source_id="source-ai",
        source_version=None,
        confidence_rank=1,
        is_locked=False,
        has_conflict=True,
        needs_review=True,
        last_changed_by="agent",
        user_feedback=None,
        created_at=now,
        updated_at=now,
    )
    candidate = SemanticCandidate(
        candidate_id="cand-1",
        field_id="field-1",
        source_id="dict-1",
        candidate_rank=1,
        match_type=CandidateMatchType.EXACT,
        confidence_score=1.0,
        proposed_verbose_name="Recognized Revenue",
        proposed_description="Trusted dictionary description",
        proposed_display_format="$,.2f",
        status=CandidateStatus.PROPOSED,
        created_at=now,
    )
    field.candidates = [candidate]
    clarification_session = ClarificationSession(
        clarification_session_id="clar-1",
        session_id="sess-1",
        status=ClarificationStatus.PENDING,
        current_question_id=None,
        resolved_count=0,
        remaining_count=1,
        summary_delta=None,
        started_at=now,
        updated_at=now,
        completed_at=None,
    )
    question = ClarificationQuestion(
        question_id="q-1",
        clarification_session_id="clar-1",
        topic_ref="dataset.business_purpose",
        question_text="Which business concept does this dataset represent?",
        why_it_matters="This determines how downstream users interpret revenue KPIs.",
        current_guess="Revenue reporting",
        priority=100,
        state=QuestionState.OPEN,
        created_at=now,
        updated_at=now,
    )
    question.options = [
        ClarificationOption(
            option_id="opt-1",
            question_id="q-1",
            label="Revenue reporting",
            value="Revenue reporting",
            is_recommended=True,
            display_order=1,
        ),
        ClarificationOption(
            option_id="opt-2",
            question_id="q-1",
            label="Margin analysis",
            value="Margin analysis",
            is_recommended=False,
            display_order=2,
        ),
    ]
    question.answer = None
    clarification_session.questions = [question]
    session.findings = []
    session.collaborators = []
    session.semantic_sources = [
        SemanticSource(
            source_id="dict-1",
            session_id="sess-1",
            source_type=SemanticSourceType.CONNECTED_DICTIONARY,
            source_ref="dict://finance",
            source_version="2026.03",
            display_name="Finance Dictionary",
            trust_level=TrustLevel.TRUSTED,
            schema_overlap_score=1.0,
            status=SemanticSourceStatus.AVAILABLE,
            created_at=now,
        )
    ]
    session.semantic_fields = [field]
    session.imported_filters = []
    session.template_variables = []
    session.execution_mappings = []
    session.clarification_sessions = [clarification_session]
    session.previews = []
    session.run_contexts = []
    return session
# #endregion _make_us2_session
# #region _make_us3_session [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
def _make_us3_session():
    """Fake session factory for US3 flow tests.
    `imported_filter` and `template_variable` are bare MagicMocks without spec;
    ORM attribute access is unchecked.
    """
    now = datetime.now(UTC)
    session = _make_session()
    session.readiness_state = ReadinessState.MAPPING_REVIEW_NEEDED
    session.recommended_action = RecommendedAction.APPROVE_MAPPING
    session.current_phase = SessionPhase.MAPPING_REVIEW
    imported_filter = MagicMock()  # @RISK: No spec= guard — enum/field contract changes are undetectable at test time.
    imported_filter.filter_id = "filter-1"
    imported_filter.session_id = "sess-1"
    imported_filter.filter_name = "country"
    imported_filter.display_name = "Country"
    imported_filter.raw_value = "DE"
    imported_filter.normalized_value = "DE"
    imported_filter.source = "superset_url"
    imported_filter.confidence_state = "imported"
    imported_filter.requires_confirmation = False
    imported_filter.recovery_status = "recovered"
    imported_filter.notes = "Recovered from URL state"
    template_variable = MagicMock()  # @RISK: No spec= guard — enum/field contract changes are undetectable at test time.
    template_variable.variable_id = "var-1"
    template_variable.session_id = "sess-1"
    template_variable.variable_name = "country"
    template_variable.expression_source = "{{ filter_values('country') }}"
    template_variable.variable_kind = "native_filter"
    template_variable.is_required = True
    template_variable.default_value = None
    template_variable.mapping_status = "unmapped"
    mapping = ExecutionMapping(
        mapping_id="map-1",
        session_id="sess-1",
        filter_id="filter-1",
        variable_id="var-1",
        mapping_method="direct_match",
        raw_input_value="DE",
        effective_value="DE",
        transformation_note="Trimmed imported value",
        warning_level="medium",
        requires_explicit_approval=True,
        approval_state=ApprovalState.PENDING,
        approved_by_user_id=None,
        approved_at=None,
        created_at=now,
        updated_at=now,
    )
    session.findings = []
    session.collaborators = []
    session.semantic_sources = []
    session.semantic_fields = []
    session.imported_filters = [imported_filter]
    session.template_variables = [template_variable]
    session.execution_mappings = [mapping]
    session.clarification_sessions = []
    session.previews = []
    session.run_contexts = []
    return session
# #endregion _make_us3_session
# #region _make_preview_ready_session [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
def _make_preview_ready_session():
    session = _make_us3_session()
    session.readiness_state = ReadinessState.COMPILED_PREVIEW_READY
    session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW
    session.current_phase = SessionPhase.PREVIEW
    return session
# #endregion _make_preview_ready_session
# #region dataset_review_api_dependencies [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
@pytest.fixture(autouse=True)
def dataset_review_api_dependencies():
    mock_user = _make_user()
    config_manager = _make_config_manager()
    task_manager = MagicMock()
    app.dependency_overrides[get_current_user] = lambda: mock_user
    app.dependency_overrides[get_config_manager] = lambda: config_manager
    app.dependency_overrides[get_task_manager] = lambda: task_manager
    yield {
        "user": mock_user,
        "config_manager": config_manager,
        "task_manager": task_manager,
    }
    app.dependency_overrides.clear()
# #endregion dataset_review_api_dependencies
# #region test_parse_superset_link_dashboard_partial_recovery [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify dashboard links recover dataset context and preserve explicit partial-recovery markers.
def test_parse_superset_link_dashboard_partial_recovery():
    env = Environment(
        id="env-1",
        name="DEV",
        url="http://superset.local",
        username="demo",
        password="secret",
    )
    fake_client = MagicMock()
    fake_client.get_dashboard_detail.return_value = {
        "id": 10,
        "datasets": [{"id": 42}, {"id": 77}],
    }
    fake_client.get_dataset_detail.return_value = {
        "table_name": "sales",
        "schema": "public",
    }
    extractor = SupersetContextExtractor(environment=env, client=fake_client)
    result = extractor.parse_superset_link(
        "http://superset.local/dashboard/10/?native_filters=%5B%7B%22name%22%3A%22country%22%2C%22value%22%3A%22DE%22%7D%5D"
    )
    assert result.dataset_id == 42
    assert result.dashboard_id == 10
    assert result.dataset_ref == "public.sales"
    assert result.partial_recovery is True
    assert "multiple_dashboard_datasets" in result.unresolved_references
    assert result.imported_filters[0]["filter_name"] == "country"
# #endregion test_parse_superset_link_dashboard_partial_recovery
# #region test_parse_superset_link_dashboard_slug_recovery [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify dashboard slug links resolve through dashboard detail endpoints and recover dataset context.
def test_parse_superset_link_dashboard_slug_recovery():
    env = Environment(
        id="env-1",
        name="DEV",
        url="http://superset.local",
        username="demo",
        password="secret",
    )
    fake_client = MagicMock()
    fake_client.get_dashboard_detail.return_value = {
        "id": 15,
        "datasets": [{"id": 42}],
    }
    fake_client.get_dataset_detail.return_value = {
        "table_name": "sales",
        "schema": "public",
    }
    extractor = SupersetContextExtractor(environment=env, client=fake_client)
    result = extractor.parse_superset_link(
        "https://ss-dev.bebesh.ru/superset/dashboard/slack/?native_filters_key=8ZLV4M-UXOM"
    )
    assert result.dataset_id == 42
    assert result.dashboard_id == 15
    assert result.dataset_ref == "public.sales"
    assert result.partial_recovery is False
    assert result.query_state["native_filters_key"] == "8ZLV4M-UXOM"
    fake_client.get_dashboard_detail.assert_called_once_with("slack")
# #endregion test_parse_superset_link_dashboard_slug_recovery
# #region test_parse_superset_link_dashboard_permalink_partial_recovery [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify dashboard permalink links no longer fail parsing and preserve permalink filter state for partial recovery.
def test_parse_superset_link_dashboard_permalink_partial_recovery():
    env = Environment(
        id="env-1",
        name="DEV",
        url="http://superset.local",
        username="demo",
        password="secret",
    )
    fake_client = MagicMock()
    fake_client.get_dashboard_permalink_state.return_value = {
        "state": {
            "dataMask": {
                "NATIVE_FILTER-1": {
                    "id": "country",
                    "filterState": {
                        "label": "Country",
                        "value": ["DE"],
                    },
                    "extraFormData": {
                        "filters": [{"col": "country", "op": "IN", "val": ["DE"]}],
                    },
                }
            }
        }
    }
    extractor = SupersetContextExtractor(environment=env, client=fake_client)
    result = extractor.parse_superset_link(
        "http://ss-dev.bebesh.ru/superset/dashboard/p/QabXy6wG30Z/"
    )
    assert result.resource_type == "dashboard"
    assert result.dataset_id is None
    assert result.dashboard_id is None
    assert result.dataset_ref == "dashboard_permalink:QabXy6wG30Z"
    assert result.partial_recovery is True
    assert (
        "dashboard_permalink_dataset_binding_unresolved" in result.unresolved_references
    )
    assert result.imported_filters[0]["filter_name"] == "country"
    assert result.imported_filters[0]["raw_value"] == ["DE"]
    fake_client.get_dashboard_permalink_state.assert_called_once_with("QabXy6wG30Z")
# #region test_parse_superset_link_dashboard_permalink_recovers_dataset_from_nested_dashboard_state [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify permalink state with nested dashboard id recovers dataset binding and keeps imported filters.
def test_parse_superset_link_dashboard_permalink_recovers_dataset_from_nested_dashboard_state():
    env = Environment(
        id="env-1",
        name="DEV",
        url="http://superset.local",
        username="demo",
        password="secret",
    )
    fake_client = MagicMock()
    fake_client.get_dashboard_permalink_state.return_value = {
        "state": {
            "form_data": {"dashboardId": 22},
            "dataMask": {
                "NATIVE_FILTER-1": {
                    "id": "country",
                    "filterState": {"label": "Country", "value": ["DE"]},
                }
            },
        }
    }
    fake_client.get_dashboard_detail.return_value = {"id": 22, "datasets": [{"id": 42}]}
    fake_client.get_dataset_detail.return_value = {
        "table_name": "sales",
        "schema": "public",
    }
    extractor = SupersetContextExtractor(environment=env, client=fake_client)
    result = extractor.parse_superset_link(
        "http://ss-dev.bebesh.ru/superset/dashboard/p/QabXy6wG30Z/"
    )
    assert result.dashboard_id == 22
    assert result.dataset_id == 42
    assert result.dataset_ref == "public.sales"
    assert (
        "dashboard_permalink_dataset_binding_unresolved"
        not in result.unresolved_references
    )
    assert result.imported_filters[0]["filter_name"] == "country"
# #endregion test_parse_superset_link_dashboard_permalink_recovers_dataset_from_nested_dashboard_state
# #endregion test_parse_superset_link_dashboard_permalink_partial_recovery
# #region test_resolve_from_dictionary_prefers_exact_match [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify trusted dictionary exact matches outrank fuzzy candidates and unresolved fields stay explicit.
def test_resolve_from_dictionary_prefers_exact_match():
    resolver = SemanticSourceResolver()
    result = resolver.resolve_from_dictionary(
        {
            "source_ref": "dict://finance",
            "rows": [
                {
                    "field_name": "revenue",
                    "verbose_name": "Revenue",
                    "description": "Recognized revenue amount",
                    "display_format": "$,.2f",
                },
                {
                    "field_name": "revnue",
                    "verbose_name": "Revenue typo",
                    "description": "Fuzzy variant",
                },
            ],
        },
        [
            {"field_name": "revenue", "is_locked": False},
            {"field_name": "margin", "is_locked": False},
        ],
    )
    resolved_exact = next(
        item for item in result.resolved_fields if item["field_name"] == "revenue"
    )
    unresolved = next(
        item for item in result.resolved_fields if item["field_name"] == "margin"
    )
    assert resolved_exact["applied_candidate"]["match_type"] == "exact"
    assert resolved_exact["provenance"] == "dictionary_exact"
    assert unresolved["status"] == "unresolved"
    assert "margin" in result.unresolved_fields
    assert result.partial_recovery is True
# #endregion test_resolve_from_dictionary_prefers_exact_match
# #region test_orchestrator_start_session_preserves_partial_recovery [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify session start persists usable recovery-required state when Superset intake is partial.
def test_orchestrator_start_session_preserves_partial_recovery(
    dataset_review_api_dependencies,
):
    repository = MagicMock()
    created_session = _make_session()
    created_session.readiness_state = ReadinessState.RECOVERY_REQUIRED
    created_session.current_phase = SessionPhase.RECOVERY
    repository.create_session.return_value = created_session
    repository.save_profile_and_findings.return_value = created_session
    repository.save_recovery_state.return_value = created_session
    repository.db = MagicMock()
    orchestrator = DatasetReviewOrchestrator(
        repository=repository,
        config_manager=dataset_review_api_dependencies["config_manager"],
        # WARNING: task_manager=None — all async task dispatch paths are bypassed.
        # Task dispatch failures are invisible to this test.
        task_manager=None,
    )
    parsed_context = SimpleNamespace(
        dataset_ref="public.sales",
        dataset_id=42,
        dashboard_id=10,
        chart_id=None,
        partial_recovery=True,
        unresolved_references=["dashboard_dataset_binding_missing"],
        imported_filters=[],
    )
    fake_extractor = MagicMock()
    fake_extractor.parse_superset_link.return_value = parsed_context
    fake_extractor.recover_imported_filters.return_value = []
    fake_extractor.client.get_dataset_detail.return_value = {
        "id": 42,
        "sql": "",
        "columns": [],
        "metrics": [],
    }
    fake_extractor.discover_template_variables.return_value = []
    with patch(
        "src.services.dataset_review.orchestrator.SupersetContextExtractor",
        side_effect=[fake_extractor, fake_extractor],
    ):
        result = orchestrator.start_session(
            StartSessionCommand(
                user=dataset_review_api_dependencies["user"],
                environment_id="env-1",
                source_kind="superset_link",
                source_input="http://superset.local/dashboard/10",
            )
        )
    assert result.session.readiness_state == ReadinessState.RECOVERY_REQUIRED
    assert result.findings
    assert result.findings[0].severity.value == "warning"
    repository.create_session.assert_called_once()
    repository.save_profile_and_findings.assert_called_once()
# #endregion test_orchestrator_start_session_preserves_partial_recovery
# #region test_orchestrator_start_session_bootstraps_recovery_state [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify session start persists recovered filters, template variables, and initial execution mappings for review workspace bootstrap.
def test_orchestrator_start_session_bootstraps_recovery_state(
    dataset_review_api_dependencies,
):
    repository = MagicMock()
    created_session = _make_session()
    created_session.readiness_state = ReadinessState.RECOVERY_REQUIRED
    created_session.current_phase = SessionPhase.RECOVERY
    repository.create_session.return_value = created_session
    repository.save_profile_and_findings.return_value = created_session
    repository.save_recovery_state.return_value = created_session
    repository.db = MagicMock()
    orchestrator = DatasetReviewOrchestrator(
        repository=repository,
        config_manager=dataset_review_api_dependencies["config_manager"],
        # WARNING: task_manager=None — all async task dispatch paths are bypassed.
        # Task dispatch failures are invisible to this test.
        task_manager=None,
    )
    parsed_context = SimpleNamespace(
        dataset_ref="public.sales",
        dataset_id=42,
        dashboard_id=10,
        chart_id=None,
        partial_recovery=True,
        unresolved_references=["dashboard_dataset_binding_missing"],
        imported_filters=[{"filter_name": "country", "raw_value": ["DE"]}],
        dataset_payload=None,
    )
    fake_extractor = MagicMock()
    fake_extractor.parse_superset_link.return_value = parsed_context
    fake_extractor.recover_imported_filters.return_value = [
        {
            "filter_name": "country",
            "display_name": "Country",
            "raw_value": ["DE"],
            "normalized_value": {
                "filter_clauses": [{"col": "country_code", "op": "IN", "val": ["DE"]}],
                "extra_form_data": {
                    "filters": [{"col": "country_code", "op": "IN", "val": ["DE"]}]
                },
                "value_origin": "extra_form_data.filters",
            },
            "source": "superset_url",
            "confidence_state": "imported",
            "requires_confirmation": False,
            "recovery_status": "recovered",
            "notes": "Recovered from permalink state",
        }
    ]
    fake_extractor.client.get_dataset_detail.return_value = {
        "id": 42,
        "sql": "select * from sales where country in {{ filter_values('country') }}",
        "columns": [],
        "metrics": [],
    }
    fake_extractor.discover_template_variables.return_value = [
        {
            "variable_name": "country",
            "expression_source": "{{ filter_values('country') }}",
            "variable_kind": "native_filter",
            "is_required": True,
            "default_value": None,
            "mapping_status": "unmapped",
        }
    ]
    with patch(
        "src.services.dataset_review.orchestrator.SupersetContextExtractor",
        side_effect=[fake_extractor, fake_extractor],
    ):
        result = orchestrator.start_session(
            StartSessionCommand(
                user=dataset_review_api_dependencies["user"],
                environment_id="env-1",
                source_kind="superset_link",
                source_input="http://superset.local/dashboard/10",
            )
        )
    assert result.session.readiness_state == ReadinessState.RECOVERY_REQUIRED
    repository.save_recovery_state.assert_called_once()
    saved_filters = repository.save_recovery_state.call_args.args[2]
    saved_variables = repository.save_recovery_state.call_args.args[3]
    saved_mappings = repository.save_recovery_state.call_args.args[4]
    assert len(saved_filters) == 1
    assert saved_filters[0].filter_name == "country"
    assert saved_filters[0].normalized_value == {
        "filter_clauses": [{"col": "country_code", "op": "IN", "val": ["DE"]}],
        "extra_form_data": {
            "filters": [{"col": "country_code", "op": "IN", "val": ["DE"]}]
        },
        "value_origin": "extra_form_data.filters",
    }
    assert len(saved_variables) == 1
    assert saved_variables[0].variable_name == "country"
    assert len(saved_mappings) == 1
    assert saved_mappings[0].raw_input_value == ["DE"]
# #endregion test_orchestrator_start_session_bootstraps_recovery_state
# #region test_start_session_endpoint_returns_created_summary [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify POST session lifecycle endpoint returns a persisted ownership-scoped summary.
def test_start_session_endpoint_returns_created_summary(
    dataset_review_api_dependencies,
):
    session = _make_session()
    orchestrator = MagicMock()
    orchestrator.start_session.return_value = SimpleNamespace(
        session=session, findings=[], parsed_context=None
    )
    app.dependency_overrides[_get_orchestrator] = lambda: orchestrator
    response = client.post(
        "/api/dataset-orchestration/sessions",
        json={
            "source_kind": "superset_link",
            "source_input": "http://superset.local/dashboard/10",
            "environment_id": "env-1",
        },
    )
    assert response.status_code == 201
    payload = response.json()
    assert payload["session_id"] == "sess-1"
    assert payload["dataset_ref"] == "public.sales"
    assert payload["environment_id"] == "env-1"
# #endregion test_start_session_endpoint_returns_created_summary
# #region test_get_session_detail_export_and_lifecycle_endpoints [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Verify lifecycle get/patch/delete plus documentation and validation exports remain ownership-scoped and usable.
def test_get_session_detail_export_and_lifecycle_endpoints(
    dataset_review_api_dependencies,
):
    now = datetime.now(UTC)
    session = MagicMock(spec=DatasetReviewSession)
    session.session_id = "sess-1"
    session.user_id = "user-1"
    session.environment_id = "env-1"
    session.source_kind = "superset_link"
    session.source_input = "http://superset.local/dashboard/10"
    session.dataset_ref = "public.sales"
    session.dataset_id = 42
    session.dashboard_id = 10
    session.readiness_state = ReadinessState.REVIEW_READY
    session.recommended_action = RecommendedAction.REVIEW_DOCUMENTATION
    session.status = SessionStatus.ACTIVE
    session.current_phase = SessionPhase.REVIEW
    session.created_at = now
    session.updated_at = now
    session.last_activity_at = now
    session.profile = SimpleNamespace(
        dataset_name="sales",
        business_summary="Summary text",
        confidence_state=ConfidenceState.MOSTLY_CONFIRMED,
        dataset_type="unknown",
        schema_name=None,
        database_name=None,
        business_summary_source=BusinessSummarySource.IMPORTED,
        description=None,
        is_sqllab_view=False,
        completeness_score=None,
        has_blocking_findings=False,
        has_warning_findings=True,
        manual_summary_locked=False,
        created_at=now,
        updated_at=now,
        profile_id="profile-1",
        session_id="sess-1",
    )
    session.findings = [
        SimpleNamespace(
            finding_id="f-1",
            session_id="sess-1",
            area=FindingArea.SOURCE_INTAKE,
            severity=FindingSeverity.WARNING,
            code="PARTIAL_SUPERSET_RECOVERY",
            title="Partial",
            message="Some filters require review",
            resolution_state=ResolutionState.OPEN,
            resolution_note=None,
            caused_by_ref=None,
            created_at=now,
            resolved_at=None,
        )
    ]
    session.collaborators = []
    session.semantic_sources = []
    session.semantic_fields = []
    session.imported_filters = []
    session.template_variables = []
    session.execution_mappings = []
    session.clarification_sessions = []
    session.previews = []
    session.run_contexts = []
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.list_sessions_for_user.return_value = [session]
    repository.require_session_version.side_effect = lambda current, expected: current
    repository.bump_session_version.side_effect = lambda current: setattr(
        current, "version", int(getattr(current, "version", 0) or 0) + 1
    ) or getattr(current, "version", 0)
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    repository.event_logger.log_for_session.return_value = SimpleNamespace(
        session_event_id="evt-0"
    )
    app.dependency_overrides[_get_repository] = lambda: repository
    detail_response = client.get("/api/dataset-orchestration/sessions/sess-1")
    assert detail_response.status_code == 200
    assert detail_response.json()["session_id"] == "sess-1"
    patch_response = client.patch(
        "/api/dataset-orchestration/sessions/sess-1",
        json={"status": "paused"},
        headers={"X-Session-Version": "0"},
    )
    assert patch_response.status_code == 200
    assert patch_response.json()["status"] == "paused"
    doc_response = client.get(
        "/api/dataset-orchestration/sessions/sess-1/exports/documentation?format=json"
    )
    assert doc_response.status_code == 200
    assert doc_response.json()["artifact_type"] == "documentation"
    validation_response = client.get(
        "/api/dataset-orchestration/sessions/sess-1/exports/validation?format=markdown"
    )
    assert validation_response.status_code == 200
    assert validation_response.json()["artifact_type"] == "validation_report"
    assert "Validation Report" in validation_response.json()["content"]["markdown"]
    delete_response = client.delete(
        "/api/dataset-orchestration/sessions/sess-1",
        headers={"X-Session-Version": "1"},
    )
    assert delete_response.status_code == 204
# #endregion test_get_session_detail_export_and_lifecycle_endpoints
# #region test_get_clarification_state_returns_empty_payload_when_session_has_no_record [TYPE Function]
# @BRIEF Clarification state endpoint should return a non-blocking empty payload when the session has no clarification aggregate yet.
def test_get_clarification_state_returns_empty_payload_when_session_has_no_record(
    dataset_review_api_dependencies,
):
    session = _make_us3_session()
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    app.dependency_overrides[_get_repository] = lambda: repository
    response = client.get("/api/dataset-orchestration/sessions/sess-1/clarification")
    assert response.status_code == 200
    assert response.json() == {
        "clarification_session": None,
        "current_question": None,
    }
# #endregion test_get_clarification_state_returns_empty_payload_when_session_has_no_record
# #region test_us2_clarification_endpoints_persist_answer_and_feedback [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Clarification endpoints should expose one current question, persist the answer before advancement, and store feedback on the answer audit record.
def test_us2_clarification_endpoints_persist_answer_and_feedback(
    dataset_review_api_dependencies,
):
    session = _make_us2_session()
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.require_session_version.side_effect = lambda current, expected: current
    repository.db = MagicMock()
    repository.db.commit.side_effect = lambda: None
    repository.db.refresh.side_effect = lambda obj: None
    def _add_side_effect(obj):
        if obj.__class__.__name__ == "ClarificationAnswer":
            session.clarification_sessions[0].questions[0].answer = obj
    repository.db.add.side_effect = _add_side_effect
    repository.db.flush.side_effect = lambda: None
    app.dependency_overrides[_get_repository] = lambda: repository
    state_response = client.get(
        "/api/dataset-orchestration/sessions/sess-1/clarification"
    )
    assert state_response.status_code == 200
    state_payload = state_response.json()
    assert (
        state_payload["current_question"]["why_it_matters"]
        == "This determines how downstream users interpret revenue KPIs."
    )
    assert state_payload["current_question"]["current_guess"] == "Revenue reporting"
    assert len(state_payload["current_question"]["options"]) == 2
    answer_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/clarification/answers",
        json={
            "question_id": "q-1",
            "answer_kind": "selected",
            "answer_value": "Revenue reporting",
        },
        headers={"X-Session-Version": "0"},
    )
    assert answer_response.status_code == 200
    answer_payload = answer_response.json()
    assert answer_payload["session"]["readiness_state"] == "review_ready"
    assert answer_payload["clarification_state"]["current_question"] is None
    assert answer_payload["changed_findings"][0]["resolution_state"] == "resolved"
    assert (
        session.clarification_sessions[0].questions[0].answer.answer_value
        == "Revenue reporting"
    )
    feedback_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/clarification/questions/q-1/feedback",
        json={"feedback": "up"},
        headers={"X-Session-Version": "0"},
    )
    assert feedback_response.status_code == 200
    assert feedback_response.json() == {"target_id": "q-1", "feedback": "up"}
    assert session.clarification_sessions[0].questions[0].answer.user_feedback == "up"
# #endregion test_us2_clarification_endpoints_persist_answer_and_feedback
# #region test_us2_field_semantic_override_lock_unlock_and_feedback [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Semantic field endpoints should apply manual overrides with lock/provenance invariants and persist feedback independently.
def test_us2_field_semantic_override_lock_unlock_and_feedback(
    dataset_review_api_dependencies,
):
    session = _make_us2_session()
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.require_session_version.side_effect = lambda current, expected: current
    repository.bump_session_version.side_effect = lambda current: setattr(
        current, "version", int(getattr(current, "version", 0) or 0) + 1
    ) or getattr(current, "version", 0)
    repository.db = MagicMock()
    repository.db.commit.side_effect = lambda: None
    repository.db.refresh.side_effect = lambda obj: None
    repository.db.add.side_effect = lambda obj: None
    repository.db.flush.side_effect = lambda: None
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    repository.event_logger.log_for_session.return_value = SimpleNamespace(
        session_event_id="evt-1"
    )
    app.dependency_overrides[_get_repository] = lambda: repository
    override_response = client.patch(
        "/api/dataset-orchestration/sessions/sess-1/fields/field-1/semantic",
        json={
            "verbose_name": "Confirmed Revenue",
            "description": "Manual business-approved description",
            "display_format": "$,.0f",
        },
        headers={"X-Session-Version": "0"},
    )
    assert override_response.status_code == 200
    override_payload = override_response.json()
    assert override_payload["provenance"] == "manual_override"
    assert override_payload["is_locked"] is True
    unlock_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/fields/field-1/unlock",
        headers={"X-Session-Version": "1"},
    )
    assert unlock_response.status_code == 200
    assert unlock_response.json()["is_locked"] is False
    candidate_response = client.patch(
        "/api/dataset-orchestration/sessions/sess-1/fields/field-1/semantic",
        json={"candidate_id": "cand-1", "lock_field": True},
        headers={"X-Session-Version": "2"},
    )
    assert candidate_response.status_code == 200
    candidate_payload = candidate_response.json()
    assert candidate_payload["verbose_name"] == "Recognized Revenue"
    assert candidate_payload["provenance"] == "dictionary_exact"
    assert candidate_payload["is_locked"] is True
    batch_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/fields/semantic/approve-batch",
        json={
            "items": [
                {"field_id": "field-1", "candidate_id": "cand-1", "lock_field": False}
            ]
        },
        headers={"X-Session-Version": "3"},
    )
    assert batch_response.status_code == 200
    assert batch_response.json()[0]["field_id"] == "field-1"
    feedback_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/fields/field-1/feedback",
        json={"feedback": "down"},
        headers={"X-Session-Version": "4"},
    )
    assert feedback_response.status_code == 200
    assert feedback_response.json() == {"target_id": "field-1", "feedback": "down"}
    assert session.semantic_fields[0].user_feedback == "down"
# #endregion test_us2_field_semantic_override_lock_unlock_and_feedback
# #region test_us3_mapping_patch_approval_preview_and_launch_endpoints [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF US3 execution endpoints should persist manual overrides, preserve explicit approval semantics, return contract-shaped preview truth, and expose audited launch handoff.
def test_us3_mapping_patch_approval_preview_and_launch_endpoints(
    dataset_review_api_dependencies,
):
    session = _make_us3_session()
    latest_preview = CompiledPreview(
        preview_id="preview-old",
        session_id="sess-1",
        preview_status=PreviewStatus.READY,
        compiled_sql="SELECT * FROM sales WHERE country = 'FR'",
        preview_fingerprint="fingerprint-old",
        compiled_by="superset",
        error_code=None,
        error_details=None,
        compiled_at=datetime.now(UTC),
        created_at=datetime.now(UTC),
    )
    session.previews = [latest_preview]
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.require_session_version.side_effect = lambda current, expected: current
    repository.bump_session_version.side_effect = lambda current: setattr(
        current, "version", int(getattr(current, "version", 0) or 0) + 1
    ) or getattr(current, "version", 0)
    repository.db = MagicMock()
    repository.db.commit.side_effect = lambda: None
    repository.db.refresh.side_effect = lambda obj: None
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    repository.event_logger.log_for_session.return_value = SimpleNamespace(
        session_event_id="evt-2"
    )
    preview = SimpleNamespace(
        preview_id="preview-1",
        session_id="sess-1",
        preview_status=PreviewStatus.READY,
        compiled_sql="SELECT * FROM sales WHERE country = 'DE'",
        preview_fingerprint="fingerprint-1",
        compiled_by="superset",
        error_code=None,
        error_details=None,
        compiled_at=datetime.now(UTC),
        created_at=datetime.now(UTC),
    )
    run_context = SimpleNamespace(
        run_context_id="run-1",
        session_id="sess-1",
        dataset_ref="public.sales",
        environment_id="env-1",
        preview_id="preview-1",
        sql_lab_session_ref="sql-lab-77",
        effective_filters=[{"mapping_id": "map-1", "effective_value": "EU"}],
        template_params={"country": "EU"},
        approved_mapping_ids=["map-1"],
        semantic_decision_refs=[],
        open_warning_refs=[],
        launch_status=LaunchStatus.STARTED,
        launch_error=None,
        created_at=datetime.now(UTC),
    )
    orchestrator = MagicMock()
    orchestrator.prepare_launch_preview.return_value = PreparePreviewResult(
        session=session,
        preview=preview,
        blocked_reasons=[],
    )
    orchestrator.launch_dataset.return_value = LaunchDatasetResult(
        session=session,
        run_context=run_context,
        blocked_reasons=[],
    )
    def _assert_expected_preview_version(command):
        assert command.expected_version == 3
        return PreparePreviewResult(
            session=session, preview=preview, blocked_reasons=[]
        )
    def _assert_expected_launch_version(command):
        assert command.expected_version == 5
        return LaunchDatasetResult(
            session=session, run_context=run_context, blocked_reasons=[]
        )
    orchestrator.prepare_launch_preview.side_effect = _assert_expected_preview_version
    orchestrator.launch_dataset.side_effect = _assert_expected_launch_version
    app.dependency_overrides[_get_repository] = lambda: repository
    app.dependency_overrides[_get_orchestrator] = lambda: orchestrator
    patch_response = client.patch(
        "/api/dataset-orchestration/sessions/sess-1/mappings/map-1",
        json={
            "effective_value": "EU",
            "mapping_method": "manual_override",
            "transformation_note": "Manual override for SQL Lab launch",
        },
        headers={"X-Session-Version": "0"},
    )
    assert patch_response.status_code == 200
    patch_payload = patch_response.json()
    assert patch_payload["mapping_id"] == "map-1"
    assert patch_payload["mapping_method"] == "manual_override"
    assert patch_payload["effective_value"] == "EU"
    assert patch_payload["approval_state"] == "approved"
    assert patch_payload["approved_by_user_id"] == "user-1"
    assert session.execution_mappings[0].mapping_method == MappingMethod.MANUAL_OVERRIDE
    assert (
        session.execution_mappings[0].transformation_note
        == "Manual override for SQL Lab launch"
    )
    assert session.execution_mappings[0].effective_value == "EU"
    assert session.recommended_action == RecommendedAction.GENERATE_SQL_PREVIEW
    assert latest_preview.preview_status == PreviewStatus.STALE
    approve_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/mappings/map-1/approve",
        json={"approval_note": "Approved after reviewing transformation"},
        headers={"X-Session-Version": "1"},
    )
    assert approve_response.status_code == 200
    approve_payload = approve_response.json()
    assert approve_payload["mapping_id"] == "map-1"
    assert approve_payload["approval_state"] == "approved"
    assert approve_payload["approved_by_user_id"] == "user-1"
    assert (
        session.execution_mappings[0].transformation_note
        == "Approved after reviewing transformation"
    )
    batch_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/mappings/approve-batch",
        json={"mapping_ids": ["map-1"]},
        headers={"X-Session-Version": "2"},
    )
    assert batch_response.status_code == 200
    assert batch_response.json()[0]["mapping_id"] == "map-1"
    list_response = client.get("/api/dataset-orchestration/sessions/sess-1/mappings")
    assert list_response.status_code == 200
    assert list_response.json()["items"][0]["mapping_id"] == "map-1"
    preview_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/preview",
        headers={"X-Session-Version": "3"},
    )
    assert preview_response.status_code == 200
    preview_payload = preview_response.json()
    assert preview_payload["preview_id"] == "preview-1"
    assert preview_payload["preview_status"] == "ready"
    assert preview_payload["compiled_by"] == "superset"
    assert "SELECT * FROM sales" in preview_payload["compiled_sql"]
    def _assert_expected_pending_preview_version(command):
        assert command.expected_version == 4
        return PreparePreviewResult(
            session=session,
            preview=SimpleNamespace(
                preview_id="preview-pending",
                session_id="sess-1",
                preview_status=PreviewStatus.PENDING,
                compiled_sql=None,
                preview_fingerprint="fingerprint-pending",
                compiled_by="superset",
                error_code=None,
                error_details=None,
                compiled_at=None,
                created_at=datetime.now(UTC),
            ),
            blocked_reasons=[],
        )
    orchestrator.prepare_launch_preview.side_effect = (
        _assert_expected_pending_preview_version
    )
    preview_enqueue_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/preview",
        headers={"X-Session-Version": "4"},
    )
    assert preview_enqueue_response.status_code == 202
    assert preview_enqueue_response.json() == {
        "session_id": "sess-1",
        "session_version": 3,
        "preview_status": "pending",
        "task_id": None,
    }
    launch_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/launch",
        headers={"X-Session-Version": "5"},
    )
    assert launch_response.status_code == 201
    launch_payload = launch_response.json()
    assert launch_payload["run_context"]["run_context_id"] == "run-1"
    assert launch_payload["run_context"]["sql_lab_session_ref"] == "sql-lab-77"
    assert launch_payload["run_context"]["launch_status"] == "started"
    assert (
        launch_payload["redirect_url"]
        == "http://superset.local/superset/sqllab?queryId=sql-lab-77"
    )
# #endregion test_us3_mapping_patch_approval_preview_and_launch_endpoints
# #region test_us3_preview_response_propagates_refreshed_session_version_for_launch_follow_up [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Preview response should expose the refreshed session version so the normal preview-then-launch UI flow can satisfy optimistic locking without a forced full reload.
def test_us3_preview_response_propagates_refreshed_session_version_for_launch_follow_up(
    dataset_review_api_dependencies,
):
    session = _make_us3_session()
    session.version = 4
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    def _require_session_version(current, expected):
        if int(getattr(current, "version", 0) or 0) != expected:
            raise DatasetReviewSessionVersionConflictError(
                current.session_id,
                expected,
                int(getattr(current, "version", 0) or 0),
            )
        return current
    repository.require_session_version.side_effect = _require_session_version
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    preview = SimpleNamespace(
        preview_id="preview-2",
        session_id="sess-1",
        preview_status=PreviewStatus.READY,
        compiled_sql="SELECT * FROM sales WHERE country = 'FR'",
        preview_fingerprint="fingerprint-2",
        compiled_by="superset",
        error_code=None,
        error_details=None,
        compiled_at=datetime.now(UTC),
        created_at=datetime.now(UTC),
    )
    run_context = SimpleNamespace(
        run_context_id="run-2",
        session_id="sess-1",
        dataset_ref="public.sales",
        environment_id="env-1",
        preview_id="preview-2",
        sql_lab_session_ref="sql-lab-88",
        effective_filters=[{"mapping_id": "map-1", "effective_value": "FR"}],
        template_params={"country": "FR"},
        approved_mapping_ids=[],
        semantic_decision_refs=[],
        open_warning_refs=[],
        launch_status=LaunchStatus.STARTED,
        launch_error=None,
        created_at=datetime.now(UTC),
    )
    orchestrator = MagicMock()
    def _prepare_preview(command):
        assert command.expected_version == 4
        session.version = 5
        return PreparePreviewResult(
            session=session,
            preview=preview,
            blocked_reasons=[],
        )
    def _launch_dataset(command):
        assert command.expected_version == 5
        return LaunchDatasetResult(
            session=session,
            run_context=run_context,
            blocked_reasons=[],
        )
    orchestrator.prepare_launch_preview.side_effect = _prepare_preview
    orchestrator.launch_dataset.side_effect = _launch_dataset
    app.dependency_overrides[_get_repository] = lambda: repository
    app.dependency_overrides[_get_orchestrator] = lambda: orchestrator
    preview_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/preview",
        headers={"X-Session-Version": "4"},
    )
    assert preview_response.status_code == 200
    preview_payload = preview_response.json()
    assert preview_payload["session_version"] == 5
    launch_response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/launch",
        headers={"X-Session-Version": str(preview_payload["session_version"])},
    )
    assert launch_response.status_code == 201
    assert launch_response.json()["run_context"]["run_context_id"] == "run-2"
# #endregion test_us3_preview_response_propagates_refreshed_session_version_for_launch_follow_up
# #region test_us3_preview_endpoint_returns_failed_preview_without_false_dashboard_not_found_contract_drift [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Preview endpoint should preserve API contract and surface generic upstream preview failures without fabricating dashboard-not-found semantics for non-dashboard 404s.
def test_us3_preview_endpoint_returns_failed_preview_without_false_dashboard_not_found_contract_drift(
    dataset_review_api_dependencies,
):
    session = _make_us3_session()
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    failed_preview = SimpleNamespace(
        preview_id="preview-failed",
        session_id="sess-1",
        preview_status=PreviewStatus.FAILED,
        compiled_sql=None,
        preview_fingerprint="fingerprint-failed",
        compiled_by="superset",
        error_code="superset_preview_failed",
        error_details="RuntimeError: [API_FAILURE] API resource not found at endpoint '/chart/data' | Context: {'status_code': 404, 'endpoint': '/chart/data', 'subtype': 'not_found'}",
        compiled_at=None,
        created_at=datetime.now(UTC),
    )
    orchestrator = MagicMock()
    orchestrator.prepare_launch_preview.return_value = PreparePreviewResult(
        session=session,
        preview=failed_preview,
        blocked_reasons=[],
    )
    app.dependency_overrides[_get_repository] = lambda: repository
    app.dependency_overrides[_get_orchestrator] = lambda: orchestrator
    response = client.post(
        "/api/dataset-orchestration/sessions/sess-1/preview",
        headers={"X-Session-Version": "0"},
    )
    assert response.status_code == 200
    payload = response.json()
    assert payload["preview_id"] == "preview-failed"
    assert payload["preview_status"] == "failed"
    assert payload["compiled_sql"] is None
    assert payload["compiled_by"] == "superset"
    assert payload["error_code"] == "superset_preview_failed"
    assert "/chart/data" in payload["error_details"]
    assert "API resource not found" in payload["error_details"]
    assert "Dashboard not found" not in payload["error_details"]
# #endregion test_us3_preview_endpoint_returns_failed_preview_without_false_dashboard_not_found_contract_drift
# #region test_mutation_endpoints_surface_session_version_conflict_payload [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Dataset review mutation endpoints should return deterministic 409 conflict semantics when optimistic-lock versions are stale.
def test_mutation_endpoints_surface_session_version_conflict_payload(
    dataset_review_api_dependencies,
):
    session = _make_us3_session()
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.require_session_version.side_effect = (
        DatasetReviewSessionVersionConflictError(
            session_id="sess-1",
            expected_version=2,
            actual_version=5,
        )
    )
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    app.dependency_overrides[_get_repository] = lambda: repository
    response = client.patch(
        "/api/dataset-orchestration/sessions/sess-1/mappings/map-1",
        json={
            "effective_value": "EU",
            "mapping_method": "manual_override",
        },
        headers={"X-Session-Version": "2"},
    )
    assert response.status_code == 409
    payload = response.json()["detail"]
    assert payload["error_code"] == "session_version_conflict"
    assert payload["expected_version"] == 2
    assert payload["actual_version"] == 5
# #endregion test_mutation_endpoints_surface_session_version_conflict_payload
# #region test_update_session_surfaces_commit_time_session_version_conflict_payload [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Session lifecycle mutation should return deterministic 409 conflict semantics when commit-time optimistic locking rejects a stale write.
def test_update_session_surfaces_commit_time_session_version_conflict_payload(
    dataset_review_api_dependencies,
):
    session = _make_session()
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.require_session_version.return_value = session
    repository.commit_session_mutation.side_effect = (
        DatasetReviewSessionVersionConflictError(
            session_id="sess-1",
            expected_version=0,
            actual_version=1,
        )
    )
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    app.dependency_overrides[_get_repository] = lambda: repository
    response = client.patch(
        "/api/dataset-orchestration/sessions/sess-1",
        json={"status": "paused"},
        headers={"X-Session-Version": "0"},
    )
    assert response.status_code == 409
    payload = response.json()["detail"]
    assert payload["error_code"] == "session_version_conflict"
    assert payload["expected_version"] == 0
    assert payload["actual_version"] == 1
# #endregion test_update_session_surfaces_commit_time_session_version_conflict_payload
# #region test_execution_snapshot_includes_recovered_imported_filters_without_template_mapping [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Recovered imported filters with values should flow into preview filter context even when no template variable mapping exists.
def test_execution_snapshot_includes_recovered_imported_filters_without_template_mapping(
    dataset_review_api_dependencies,
):
    repository = MagicMock()
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    orchestrator = DatasetReviewOrchestrator(
        repository=repository,
        config_manager=dataset_review_api_dependencies["config_manager"],
        # WARNING: task_manager=None — all async task dispatch paths are bypassed.
        # Task dispatch failures are invisible to this test.
        task_manager=None,
    )
    session = _make_preview_ready_session()
    recovered_filter = MagicMock()
    recovered_filter.filter_id = "filter-country"
    recovered_filter.filter_name = "country"
    recovered_filter.display_name = "Country"
    recovered_filter.raw_value = ["DE", "FR"]
    recovered_filter.normalized_value = ["DE", "FR"]
    recovered_filter.requires_confirmation = False
    recovered_filter.recovery_status = "recovered"
    session.imported_filters = [recovered_filter]
    session.template_variables = []
    session.execution_mappings = []
    session.semantic_fields = []
    snapshot = build_execution_snapshot(session)
    assert snapshot["template_params"] == {}
    assert snapshot["preview_blockers"] == []
    recovered_filter.normalized_value = {
        "filter_clauses": [{"col": "country_code", "op": "IN", "val": ["DE", "FR"]}],
        "extra_form_data": {
            "filters": [{"col": "country_code", "op": "IN", "val": ["DE", "FR"]}]
        },
        "value_origin": "extra_form_data.filters",
    }
    snapshot = build_execution_snapshot(session)
    assert snapshot["template_params"] == {}
    assert snapshot["preview_blockers"] == []
    assert snapshot["effective_filters"] == [
        {
            "filter_id": "filter-country",
            "filter_name": "country",
            "display_name": "Country",
            "effective_value": ["DE", "FR"],
            "raw_input_value": ["DE", "FR"],
            "normalized_filter_payload": {
                "filter_clauses": [
                    {"col": "country_code", "op": "IN", "val": ["DE", "FR"]}
                ],
                "extra_form_data": {
                    "filters": [
                        {"col": "country_code", "op": "IN", "val": ["DE", "FR"]}
                    ]
                },
                "value_origin": "extra_form_data.filters",
            },
        }
    ]
# #endregion test_execution_snapshot_includes_recovered_imported_filters_without_template_mapping
# #region test_execution_snapshot_preserves_mapped_template_variables_and_filter_context [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Mapped template variables should still populate template params while contributing their effective filter context.
def test_execution_snapshot_preserves_mapped_template_variables_and_filter_context(
    dataset_review_api_dependencies,
):
    repository = MagicMock()
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    orchestrator = DatasetReviewOrchestrator(
        repository=repository,
        config_manager=dataset_review_api_dependencies["config_manager"],
        # WARNING: task_manager=None — all async task dispatch paths are bypassed.
        # Task dispatch failures are invisible to this test.
        task_manager=None,
    )
    session = _make_preview_ready_session()
    snapshot = build_execution_snapshot(session)
    assert snapshot["template_params"] == {"country": "DE"}
    assert snapshot["preview_blockers"] == []
    assert snapshot["effective_filters"] == [
        {
            "mapping_id": "map-1",
            "filter_id": "filter-1",
            "filter_name": "country",
            "variable_id": "var-1",
            "variable_name": "country",
            "effective_value": "DE",
            "raw_input_value": "DE",
        }
    ]
    assert snapshot["open_warning_refs"] == ["map-1"]
# #endregion test_execution_snapshot_preserves_mapped_template_variables_and_filter_context
# #region test_execution_snapshot_skips_partial_imported_filters_without_values [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Partial imported filters without raw or normalized values must not emit bogus active preview filters.
def test_execution_snapshot_skips_partial_imported_filters_without_values(
    dataset_review_api_dependencies,
):
    repository = MagicMock()
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    orchestrator = DatasetReviewOrchestrator(
        repository=repository,
        config_manager=dataset_review_api_dependencies["config_manager"],
        # WARNING: task_manager=None — all async task dispatch paths are bypassed.
        # Task dispatch failures are invisible to this test.
        task_manager=None,
    )
    session = _make_preview_ready_session()
    unresolved_filter = MagicMock()
    unresolved_filter.filter_id = "filter-region"
    unresolved_filter.filter_name = "region"
    unresolved_filter.display_name = "Region"
    unresolved_filter.raw_value = None
    unresolved_filter.normalized_value = None
    unresolved_filter.requires_confirmation = True
    unresolved_filter.recovery_status = "partial"
    session.imported_filters = [unresolved_filter]
    session.template_variables = []
    session.execution_mappings = []
    session.semantic_fields = []
    snapshot = build_execution_snapshot(session)
    assert snapshot["template_params"] == {}
    assert snapshot["effective_filters"] == []
    assert snapshot["preview_blockers"] == []
# #endregion test_execution_snapshot_skips_partial_imported_filters_without_values
# #region test_us3_launch_endpoint_requires_launch_permission [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Launch endpoint should enforce the contract RBAC permission instead of the generic session-manage permission.
def test_us3_launch_endpoint_requires_launch_permission(
    dataset_review_api_dependencies,
):
    session = _make_us3_session()
    repository = MagicMock()
    repository.load_session_detail.return_value = session
    repository.db = MagicMock()
    repository.event_logger = MagicMock(spec=SessionEventLogger)
    run_context = SimpleNamespace(
        run_context_id="run-1",
        session_id="sess-1",
        dataset_ref="public.sales",
        environment_id="env-1",
        preview_id="preview-1",
        sql_lab_session_ref="sql-lab-77",
        effective_filters=[],
        template_params={},
        approved_mapping_ids=["map-1"],
        semantic_decision_refs=[],
        open_warning_refs=[],
        launch_status=LaunchStatus.STARTED,
        launch_error=None,
        created_at=datetime.now(UTC),
    )
    orchestrator = MagicMock()
    orchestrator.launch_dataset.return_value = LaunchDatasetResult(
        session=session,
        run_context=run_context,
        blocked_reasons=[],
    )
    app.dependency_overrides[_get_repository] = lambda: repository
    app.dependency_overrides[_get_orchestrator] = lambda: orchestrator
    dataset_review_api_dependencies["user"].roles = [
        SimpleNamespace(
            name="SessionManager",
            permissions=[SimpleNamespace(resource="dataset:session", action="MANAGE")],
        )
    ]
    response = client.post("/api/dataset-orchestration/sessions/sess-1/launch")
    assert response.status_code == 403
    assert (
        response.json()["detail"]
        == "Permission denied for dataset:execution:launch:EXECUTE"
    )
# #endregion test_us3_launch_endpoint_requires_launch_permission
# #region test_semantic_source_version_propagation_preserves_locked_fields [TYPE Function]
# @RELATION BINDS_TO -> DatasetReviewApiTests
# @BRIEF Updated semantic source versions should mark unlocked fields reviewable while preserving locked manual values.
def test_semantic_source_version_propagation_preserves_locked_fields():
    resolver = SemanticSourceResolver()
    source = SimpleNamespace(source_id="src-1", source_version="2026.04")
    unlocked_field = SimpleNamespace(
        source_id="src-1",
        source_version="2026.03",
        is_locked=False,
        provenance=FieldProvenance.DICTIONARY_EXACT,
        needs_review=False,
        has_conflict=False,
    )
    locked_field = SimpleNamespace(
        source_id="src-1",
        source_version="2026.03",
        is_locked=True,
        provenance=FieldProvenance.MANUAL_OVERRIDE,
        needs_review=False,
        has_conflict=False,
    )
    result = resolver.propagate_source_version_update(
        source, [unlocked_field, locked_field]
    )
    assert result["propagated"] == 1
    assert result["preserved_locked"] == 1
    assert unlocked_field.source_version == "2026.04"
    assert unlocked_field.needs_review is True
    assert locked_field.source_version == "2026.03"
    assert locked_field.needs_review is False
# #endregion test_semantic_source_version_propagation_preserves_locked_fields
# #endregion DatasetReviewApiTests
