from __future__ import annotations

import importlib.util
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from app.domains.data_access import constants as c
from app.domains.data_access.models import (
    ENTITLEMENT_STATUS_ACTIVE,
    ENTITLEMENT_VERSION_V1,
    DataAccessEntitlement,
)

_MIGRATION_PATH = (
    Path(__file__).resolve().parents[2]
    / "alembic"
    / "versions"
    / "332_data_access_entitlements.py"
)


def _load_migration():
    spec = importlib.util.spec_from_file_location(
        "migration_279_data_access_entitlements",
        _MIGRATION_PATH,
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


async def _run_read_v1_seed(db_session) -> None:
    migration = _load_migration()
    connection = await db_session.connection()
    await connection.run_sync(lambda sync_conn: migration._seed_read_v1(sync_conn))


async def test_read_v1_seed_is_exact_and_idempotent(db_session) -> None:
    await _run_read_v1_seed(db_session)
    await _run_read_v1_seed(db_session)

    rows = (await db_session.scalars(select(DataAccessEntitlement))).all()

    assert len(rows) == 1
    entitlement = rows[0]
    assert entitlement.is_canonical_read_v1 is True
    assert entitlement.semantic_key == "read:v1"
    assert entitlement.status == ENTITLEMENT_STATUS_ACTIVE
    assert entitlement.requestable is True
    assert entitlement.provider_effect == {
        "member_prefix": c.BIGQUERY_GROUP_MEMBER_PREFIX,
        "role": c.BIGQUERY_DATA_VIEWER_ROLE,
    }


async def test_duplicate_read_v1_entitlement_is_rejected(db_session) -> None:
    await _run_read_v1_seed(db_session)

    db_session.add(
        DataAccessEntitlement(
            domain=c.DOMAIN_DATA_ACCESS,
            provider=c.PROVIDER_BIGQUERY,
            resource_type=c.RESOURCE_TYPE_DATASET,
            principal_type=c.PRINCIPAL_TYPE_GOOGLE_GROUP,
            entitlement=c.ENTITLEMENT_READ,
            version=ENTITLEMENT_VERSION_V1,
            provider_member_prefix=c.BIGQUERY_GROUP_MEMBER_PREFIX,
            provider_role=c.BIGQUERY_DATA_VIEWER_ROLE,
            requestable=True,
            status=ENTITLEMENT_STATUS_ACTIVE,
        ),
    )

    try:
        await db_session.flush()
    except IntegrityError:
        pass
    else:  # pragma: no cover - assertion path only
        raise AssertionError("duplicate read:v1 entitlement must be rejected")


def test_entitlement_table_has_no_per_resource_override_path() -> None:
    forbidden_columns = {
        "resource_id",
        "data_access_resource_id",
        "integration_id",
        "project_id",
        "dataset_id",
        "locator",
        "provider_role_override",
        "resource_provider_role",
    }

    actual_columns = set(DataAccessEntitlement.__table__.columns.keys())

    assert actual_columns.isdisjoint(forbidden_columns)


def test_migration_seed_uses_canonical_read_semantics() -> None:
    migration = _load_migration()
    seed = migration.READ_V1_SEED

    assert seed["domain"] == c.DOMAIN_DATA_ACCESS
    assert seed["provider"] == c.PROVIDER_BIGQUERY
    assert seed["resource_type"] == c.RESOURCE_TYPE_DATASET
    assert seed["principal_type"] == c.PRINCIPAL_TYPE_GOOGLE_GROUP
    assert seed["entitlement"] == c.ENTITLEMENT_READ
    assert seed["version"] == ENTITLEMENT_VERSION_V1
    assert seed["provider_member_prefix"] == c.BIGQUERY_GROUP_MEMBER_PREFIX
    assert seed["provider_role"] == c.BIGQUERY_DATA_VIEWER_ROLE
    assert seed["entitlement"] not in c.FORBIDDEN_ENTITLEMENT_ALIASES