from pathlib import Path
import pytest

from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import sessionmaker

from src.core.database import _ensure_dataset_review_session_columns
from src.models.auth import User
from src.models.dataset_review import (
    BusinessSummarySource,
    CompiledPreview,
    ConfidenceState,
    DatasetProfile,
    DatasetReviewSession,
    DatasetRunContext,
    FindingArea,
    FindingSeverity,
    ImportedFilter,
    SessionCollaborator,
    SessionCollaboratorRole,
    ValidationFinding,
)
from src.models.mapping import Base, Environment
from src.services.dataset_review.repositories.session_repository import (
    DatasetReviewSessionRepository,
    DatasetReviewSessionVersionConflictError,
)

# region SessionRepositoryTests [TYPE Module]
# @RELATION  BELONGS_TO -> SrcRoot
# @PURPOSE: Unit tests for DatasetReviewSessionRepository.


@pytest.fixture
def db_session():
    # region db_session [TYPE Function]
    # @PURPOSE: Build isolated in-memory SQLAlchemy session seeded with baseline user/environment rows.
    # @RELATION BINDS_TO -> [EXT:frontend:SessionRepositoryTests]
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()

    # Create test data
    user = User(
        id="user1", username="testuser", email="test@example.com", password_hash="pw"
    )
    env = Environment(
        id="env1", name="Prod", url="http://superset", credentials_id="cred1"
    )
    session.add_all([user, env])
    session.commit()

    yield session
    session.close()


# endregion db_session


# region test_create_session [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
def test_create_session(db_session):
    # @PURPOSE: Verify session creation and persistence.
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    assert session.session_id is not None
    loaded = (
        db_session.query(DatasetReviewSession)
        .filter_by(session_id=session.session_id)
        .first()
    )
    assert loaded.user_id == "user1"
    assert loaded.version == 0


# endregion test_create_session


# region test_require_session_version_conflict [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
# @PURPOSE: Verify optimistic-lock conflict is raised when caller version is stale.
def test_require_session_version_conflict(db_session):
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
        version=2,
    )
    repo.create_session(session)

    with pytest.raises(DatasetReviewSessionVersionConflictError) as exc_info:
        repo.require_session_version(session, 1)

    assert exc_info.value.expected_version == 1
    assert exc_info.value.actual_version == 2


# endregion test_require_session_version_conflict


# region test_bump_session_version_updates_last_activity [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
# @PURPOSE: Verify repository version bump increments monotonically and refreshes last activity.
def test_bump_session_version_updates_last_activity(db_session):
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    before_activity = session.last_activity_at
    next_version = repo.bump_session_version(session)

    assert next_version == 1
    assert session.version == 1
    assert session.last_activity_at >= before_activity


# endregion test_bump_session_version_updates_last_activity


# region test_save_recovery_state_preserves_raw_value_masked_flag [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
# @PURPOSE: Verify imported-filter masking metadata persists with recovery bootstrap state.
def test_save_recovery_state_preserves_raw_value_masked_flag(db_session):
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    imported_filter = ImportedFilter(
        session_id=session.session_id,
        filter_name="email",
        raw_value="***@example.com",
        raw_value_masked=True,
        normalized_value=None,
        source="manual",
        confidence_state="confirmed",
        requires_confirmation=False,
        recovery_status="recovered",
    )

    updated = repo.save_recovery_state(
        session.session_id,
        "user1",
        [imported_filter],
        [],
        [],
    )

    assert updated.imported_filters[0].raw_value_masked is True
    assert updated.version == 1


# endregion test_save_recovery_state_preserves_raw_value_masked_flag


# region test_load_session_detail_ownership [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
def test_load_session_detail_ownership(db_session):
    # @PURPOSE: Verify ownership enforcement in detail loading.
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    # Correct user
    loaded = repo.load_session_detail(session.session_id, "user1")
    assert loaded is not None

    # Wrong user
    loaded_wrong = repo.load_session_detail(session.session_id, "wrong_user")
    assert loaded_wrong is None


# endregion test_load_session_detail_ownership


# region test_load_session_detail_collaborator [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
def test_load_session_detail_collaborator(db_session):
    # @PURPOSE: Verify collaborator access in detail loading.
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    # Add collaborator
    collab_user = User(
        id="collab1", username="collab", email="c@e.com", password_hash="p"
    )
    db_session.add(collab_user)

    collaborator = SessionCollaborator(
        session_id=session.session_id,
        user_id="collab1",
        role=SessionCollaboratorRole.REVIEWER,
    )
    db_session.add(collaborator)
    db_session.commit()

    # Collaborator access
    loaded = repo.load_session_detail(session.session_id, "collab1")
    assert loaded is not None
    assert loaded.session_id == session.session_id


# endregion test_load_session_detail_collaborator


# region test_save_preview_marks_stale [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
def test_save_preview_marks_stale(db_session):
    # @PURPOSE: Verify that saving a new preview marks old ones as stale.
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    p1 = CompiledPreview(
        session_id=session.session_id, preview_status="ready", preview_fingerprint="f1"
    )
    repo.save_preview(session.session_id, "user1", p1)

    p2 = CompiledPreview(
        session_id=session.session_id, preview_status="ready", preview_fingerprint="f2"
    )
    repo.save_preview(session.session_id, "user1", p2)

    db_session.refresh(p1)
    assert p1.preview_status == "stale"
    assert p2.preview_status == "ready"
    assert session.last_preview_id == p2.preview_id


# endregion test_save_preview_marks_stale


# region test_save_preview_increments_session_version_once_per_call [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
# @PURPOSE: Verify preview persistence itself contributes exactly one optimistic-lock version increment so higher orchestration layers do not need to bump again for the same preview mutation.
def test_save_preview_increments_session_version_once_per_call(db_session):
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    repo.save_preview(
        session.session_id,
        "user1",
        CompiledPreview(
            session_id=session.session_id,
            preview_status="ready",
            preview_fingerprint="fp-ready",
        ),
    )
    db_session.refresh(session)
    assert session.version == 1

    repo.save_preview(
        session.session_id,
        "user1",
        CompiledPreview(
            session_id=session.session_id,
            preview_status="pending",
            preview_fingerprint="fp-pending",
        ),
    )
    db_session.refresh(session)
    assert session.version == 2


# endregion test_save_preview_increments_session_version_once_per_call


# region test_save_profile_and_findings [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
def test_save_profile_and_findings(db_session):
    # @PURPOSE: Verify persistence of profile and findings.
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    profile = DatasetProfile(
        session_id=session.session_id,
        dataset_name="Test DS",
        business_summary="Summary",
        business_summary_source=BusinessSummarySource.INFERRED,
        confidence_state=ConfidenceState.UNRESOLVED,
    )

    finding = ValidationFinding(
        session_id=session.session_id,
        area=FindingArea.SOURCE_INTAKE,
        severity=FindingSeverity.BLOCKING,
        code="ERR1",
        title="Error",
        message="Failure",
    )

    repo.save_profile_and_findings(session.session_id, "user1", profile, [finding])

    updated_session = repo.load_session_detail(session.session_id, "user1")
    assert updated_session.profile.dataset_name == "Test DS"
    assert len(updated_session.findings) == 1
    assert updated_session.findings[0].code == "ERR1"
    assert updated_session.version == 1

    # Verify removal of old findings
    new_finding = ValidationFinding(
        session_id=session.session_id,
        area=FindingArea.DATASET_PROFILE,
        severity=FindingSeverity.WARNING,
        code="WARN1",
        title="Warning",
        message="Something",
    )

    repo.save_profile_and_findings(session.session_id, "user1", profile, [new_finding])

    db_session.expire_all()
    final_session = repo.load_session_detail(session.session_id, "user1")
    assert len(final_session.findings) == 1
    assert final_session.findings[0].code == "WARN1"
    assert final_session.version == 2


# endregion test_save_profile_and_findings


# region test_save_profile_and_findings_rejects_stale_concurrent_write [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
# @PURPOSE: Verify repository save path translates concurrent stale session writes into deterministic optimistic-lock conflicts.
def test_save_profile_and_findings_rejects_stale_concurrent_write(tmp_path: Path):
    db_path = tmp_path / "dataset_review_session_repository.sqlite"
    engine = create_engine(f"sqlite:///{db_path}")
    Base.metadata.create_all(engine)
    SessionFactory = sessionmaker(bind=engine)

    seed_session = SessionFactory()
    seed_session.add_all(
        [
            User(
                id="user1",
                username="testuser",
                email="test@example.com",
                password_hash="pw",
            ),
            Environment(
                id="env1",
                name="Prod",
                url="http://superset",
                credentials_id="cred1",
            ),
        ]
    )
    seed_session.commit()
    created_session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    seed_session.add(created_session)
    seed_session.commit()
    session_id = created_session.session_id
    seed_session.close()

    writer_a = SessionFactory()
    writer_b = SessionFactory()
    repo_a = DatasetReviewSessionRepository(writer_a)
    repo_b = DatasetReviewSessionRepository(writer_b)

    repo_a.save_profile_and_findings(
        session_id,
        "user1",
        DatasetProfile(
            session_id=session_id,
            dataset_name="Writer A",
            business_summary="Summary A",
            business_summary_source=BusinessSummarySource.INFERRED,
            confidence_state=ConfidenceState.UNRESOLVED,
        ),
        [
            ValidationFinding(
                session_id=session_id,
                area=FindingArea.SOURCE_INTAKE,
                severity=FindingSeverity.WARNING,
                code="WARN_A",
                title="Warning A",
                message="Writer A saved first",
            )
        ],
    )

    with pytest.raises(DatasetReviewSessionVersionConflictError) as exc_info:
        repo_b.save_profile_and_findings(
            session_id,
            "user1",
            DatasetProfile(
                session_id=session_id,
                dataset_name="Writer B",
                business_summary="Summary B",
                business_summary_source=BusinessSummarySource.INFERRED,
                confidence_state=ConfidenceState.UNRESOLVED,
            ),
            [
                ValidationFinding(
                    session_id=session_id,
                    area=FindingArea.DATASET_PROFILE,
                    severity=FindingSeverity.BLOCKING,
                    code="ERR_B",
                    title="Error B",
                    message="Writer B lost optimistic lock",
                )
            ],
            expected_version=0,
        )

    assert exc_info.value.expected_version == 0
    assert exc_info.value.actual_version == 1

    writer_a.close()
    writer_b.close()


# endregion test_save_profile_and_findings_rejects_stale_concurrent_write


# region test_save_run_context [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
def test_save_run_context(db_session):
    # @PURPOSE: Verify saving of run context.
    repo = DatasetReviewSessionRepository(db_session)
    session = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="superset_link",
        source_input="http://link",
        dataset_ref="dataset1",
    )
    repo.create_session(session)

    rc = DatasetRunContext(
        session_id=session.session_id,
        dataset_ref="ds1",
        environment_id="env1",
        preview_id="p1",
        sql_lab_session_ref="s1",
        effective_filters={},
        template_params={},
        approved_mapping_ids=[],
        semantic_decision_refs=[],
        open_warning_refs=[],
        launch_status="success",
    )
    repo.save_run_context(session.session_id, "user1", rc)

    assert session.last_run_context_id == rc.run_context_id


# endregion test_save_run_context


# region test_ensure_dataset_review_session_columns_adds_missing_legacy_columns [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
# @PURPOSE: Verify additive dataset review migration creates missing legacy columns for session and imported-filter tables without dropping rows.
def test_ensure_dataset_review_session_columns_adds_missing_legacy_columns():
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as connection:
        connection.execute(
            text(
                """
                CREATE TABLE dataset_review_sessions (
                    session_id VARCHAR PRIMARY KEY,
                    user_id VARCHAR NOT NULL,
                    environment_id VARCHAR NOT NULL,
                    source_kind VARCHAR NOT NULL,
                    source_input VARCHAR NOT NULL,
                    dataset_ref VARCHAR NOT NULL,
                    dataset_id INTEGER,
                    dashboard_id INTEGER,
                    readiness_state VARCHAR NOT NULL,
                    recommended_action VARCHAR NOT NULL,
                    status VARCHAR NOT NULL,
                    current_phase VARCHAR NOT NULL,
                    active_task_id VARCHAR,
                    last_preview_id VARCHAR,
                    last_run_context_id VARCHAR,
                    created_at DATETIME NOT NULL,
                    updated_at DATETIME NOT NULL,
                    last_activity_at DATETIME NOT NULL,
                    closed_at DATETIME
                )
                """
            )
        )
        connection.execute(
            text(
                """
                CREATE TABLE imported_filters (
                    filter_id VARCHAR PRIMARY KEY,
                    session_id VARCHAR NOT NULL,
                    filter_name VARCHAR NOT NULL,
                    display_name VARCHAR,
                    raw_value JSON NOT NULL,
                    normalized_value JSON,
                    source VARCHAR NOT NULL,
                    confidence_state VARCHAR NOT NULL,
                    requires_confirmation BOOLEAN NOT NULL DEFAULT FALSE,
                    recovery_status VARCHAR NOT NULL,
                    notes TEXT,
                    created_at DATETIME NOT NULL,
                    updated_at DATETIME NOT NULL
                )
                """
            )
        )
        connection.execute(
            text(
                """
                INSERT INTO dataset_review_sessions (
                    session_id,
                    user_id,
                    environment_id,
                    source_kind,
                    source_input,
                    dataset_ref,
                    dataset_id,
                    dashboard_id,
                    readiness_state,
                    recommended_action,
                    status,
                    current_phase,
                    active_task_id,
                    last_preview_id,
                    last_run_context_id,
                    created_at,
                    updated_at,
                    last_activity_at,
                    closed_at
                ) VALUES (
                    'sess-legacy',
                    'user1',
                    'env1',
                    'superset_link',
                    'http://link',
                    'dataset1',
                    NULL,
                    NULL,
                    'EMPTY',
                    'IMPORT_FROM_SUPERSET',
                    'ACTIVE',
                    'INTAKE',
                    NULL,
                    NULL,
                    NULL,
                    '2026-03-21 00:00:00',
                    '2026-03-21 00:00:00',
                    '2026-03-21 00:00:00',
                    NULL
                )
                """
            )
        )
        connection.execute(
            text(
                """
                INSERT INTO imported_filters (
                    filter_id,
                    session_id,
                    filter_name,
                    display_name,
                    raw_value,
                    normalized_value,
                    source,
                    confidence_state,
                    requires_confirmation,
                    recovery_status,
                    notes,
                    created_at,
                    updated_at
                ) VALUES (
                    'filter-legacy',
                    'sess-legacy',
                    'country',
                    'Country',
                    '"DE"',
                    NULL,
                    'MANUAL',
                    'CONFIRMED',
                    FALSE,
                    'RECOVERED',
                    NULL,
                    '2026-03-21 00:00:00',
                    '2026-03-21 00:00:00'
                )
                """
            )
        )

    _ensure_dataset_review_session_columns(engine)

    inspector = inspect(engine)
    session_column_names = {
        column["name"] for column in inspector.get_columns("dataset_review_sessions")
    }
    imported_filter_column_names = {
        column["name"] for column in inspector.get_columns("imported_filters")
    }
    assert "version" in session_column_names
    assert "raw_value_masked" in imported_filter_column_names

    with engine.connect() as connection:
        version_value = connection.execute(
            text(
                "SELECT version FROM dataset_review_sessions WHERE session_id = 'sess-legacy'"
            )
        ).scalar_one()
        raw_value_masked = connection.execute(
            text(
                "SELECT raw_value_masked FROM imported_filters WHERE filter_id = 'filter-legacy'"
            )
        ).scalar_one()
    assert version_value == 0
    assert raw_value_masked in (False, 0)


# endregion test_ensure_dataset_review_session_columns_adds_missing_legacy_columns


# region test_list_sessions_for_user [TYPE Function]
# @RELATION  BINDS_TO -> SessionRepositoryTests
def test_list_sessions_for_user(db_session):
    # @PURPOSE: Verify listing of sessions by user.
    repo = DatasetReviewSessionRepository(db_session)
    s1 = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="k",
        source_input="i",
        dataset_ref="r1",
    )
    s2 = DatasetReviewSession(
        user_id="user1",
        environment_id="env1",
        source_kind="k",
        source_input="i",
        dataset_ref="r2",
    )
    s3 = DatasetReviewSession(
        user_id="other",
        environment_id="env1",
        source_kind="k",
        source_input="i",
        dataset_ref="r3",
    )

    db_session.add_all([s1, s2, s3])
    db_session.commit()

    sessions = repo.list_sessions_for_user("user1")
    assert len(sessions) == 2
    assert all(s.user_id == "user1" for s in sessions)


# endregion test_list_sessions_for_user
# endregion SessionRepositoryTests
