"""DAX-022 data-access driver interface tests."""
from __future__ import annotations

import inspect
from dataclasses import fields
from pathlib import Path

import pytest

from app.domains.data_access.constants import (
    BIGQUERY_DATA_VIEWER_ROLE,
    BIGQUERY_GROUP_MEMBER_PREFIX,
    DOMAIN_DATA_ACCESS,
    PROVIDER_BIGQUERY,
)
from app.domains.data_access.models import (
    DataAccessBinding,
    DataAccessCompiledEffect,
    DataAccessEffectRun,
    DataAccessRequest,
    DataAccessRequestItem,
)
from app.integrations.driver_base import (
    DataAccessDatasetIamBinding,
    DataAccessDatasetIamPolicy,
    DataAccessDatasetIamPolicyPatch,
    DataAccessDatasetIamUpdateResult,
    DataAccessDriver,
)
from app.integrations.drivers.bigquery_data_access import BigQueryDataAccessDriver
from app.integrations.registry import get_driver_class

SERVER_ROOT = Path(__file__).resolve().parents[2]


def test_dataset_iam_contract_types_are_locator_scoped() -> None:
    binding = DataAccessDatasetIamBinding(
        role="roles/bigquery.dataViewer",
        members=("group:analytics@example.com",),
    )
    policy = DataAccessDatasetIamPolicy(
        project_id="prod-data",
        dataset_id="finance",
        etag="etag-1",
        bindings=(binding,),
    )
    patch = DataAccessDatasetIamPolicyPatch(
        add_bindings=(binding,),
        remove_bindings=(),
    )
    result = DataAccessDatasetIamUpdateResult(
        ok=True,
        changed=True,
        etag="etag-2",
        policy=policy,
    )

    assert policy.project_id == "prod-data"
    assert policy.dataset_id == "finance"
    assert patch.add_bindings == (binding,)
    assert result.policy is policy


def test_data_access_driver_exposes_exact_locator_iam_methods() -> None:
    get_sig = inspect.signature(DataAccessDriver.get_dataset_iam)
    set_sig = inspect.signature(DataAccessDriver.set_dataset_iam)

    assert list(get_sig.parameters) == ["self", "project_id", "dataset_id"]
    assert list(set_sig.parameters) == [
        "self",
        "project_id",
        "dataset_id",
        "expected_etag",
        "policy_patch",
    ]
    assert get_sig.return_annotation == "DataAccessDatasetIamPolicy"
    assert set_sig.return_annotation == "DataAccessDatasetIamUpdateResult"
    assert all(
        parameter.default is inspect.Parameter.empty
        for parameter in get_sig.parameters.values()
    )
    assert all(
        parameter.default is inspect.Parameter.empty
        for parameter in set_sig.parameters.values()
    )


def test_bigquery_driver_inherits_locator_iam_interface_only() -> None:
    driver = BigQueryDataAccessDriver(integration_id="dummy", config={})

    assert hasattr(driver, "get_dataset_iam")
    assert hasattr(driver, "set_dataset_iam")
    for forbidden_method in (
        "list_datasets",
        "list_all_datasets",
        "list_resources",
        "list_all_resources",
        "scan_resources",
        "manage_datasets",
        "manage_dataset_iam",
        "create_dataset",
        "delete_dataset",
        "get_project_iam",
        "grant_project_iam",
        "set_project_iam",
        "get_project_policy",
        "set_project_policy",
        "apply_project_iam",
        "apply_project_policy",
        "grant_access",
        "revoke_access",
        "list_scopes",
        "list_scope_grants",
    ):
        assert not hasattr(driver, forbidden_method)


def test_data_access_driver_exposes_no_broad_dataset_or_cloud_grant_methods() -> None:
    forbidden_methods = {
        "list_datasets",
        "list_all_datasets",
        "list_resources",
        "list_all_resources",
        "scan_resources",
        "manage_datasets",
        "manage_dataset_iam",
        "get_project_iam",
        "set_project_iam",
        "get_project_policy",
        "set_project_policy",
        "apply_project_iam",
        "apply_project_policy",
        "grant_access",
        "revoke_access",
    }
    public_methods = {
        name
        for name, member in inspect.getmembers(DataAccessDriver)
        if callable(member) and not name.startswith("_")
    }
    assert forbidden_methods.isdisjoint(public_methods)

    dax_cls = get_driver_class(DOMAIN_DATA_ACCESS, PROVIDER_BIGQUERY)
    assert dax_cls is not None
    driver = dax_cls(integration_id="dax-interface-test", config={})
    assert all(not hasattr(driver, method) for method in forbidden_methods)


def test_dataset_iam_policy_patch_is_delta_only() -> None:
    binding = DataAccessDatasetIamBinding(
        role=BIGQUERY_DATA_VIEWER_ROLE,
        members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}analytics@example.com",),
    )
    patch = DataAccessDatasetIamPolicyPatch(
        add_bindings=(binding,),
        remove_bindings=(),
    )

    assert patch.add_bindings == (binding,)

    patch_field_names = {field.name for field in fields(DataAccessDatasetIamPolicyPatch)}
    assert patch_field_names == {"add_bindings", "remove_bindings"}
    assert {
        "project_id",
        "dataset_id",
        "bindings",
        "full_policy",
        "raw_policy",
    }.isdisjoint(patch_field_names)


@pytest.mark.asyncio
async def test_bigquery_driver_write_path_fails_closed_without_apply_gate() -> None:
    driver = BigQueryDataAccessDriver(integration_id="dummy", config={})
    patch = DataAccessDatasetIamPolicyPatch()

    result = await driver.set_dataset_iam("prod-data", "finance", "etag-1", patch)

    assert result == DataAccessDatasetIamUpdateResult(
        ok=False,
        changed=False,
        error_message="BigQuery DAX apply mode is disabled",
    )


def test_request_and_feishu_callback_paths_do_not_call_bigquery_driver_interface() -> None:
    checked_paths = [
        SERVER_ROOT / "app/domains/data_access/router.py",
        SERVER_ROOT / "app/domains/data_access/service.py",
        SERVER_ROOT / "app/approval/service.py",
    ]
    forbidden_tokens = (
        "get_dataset_iam(",
        "set_dataset_iam(",
        "BigQueryDataAccessDriver",
        "google.cloud.bigquery",
        "googleapiclient",
    )

    for path in checked_paths:
        source = path.read_text(encoding="utf-8")
        for token in forbidden_tokens:
            assert token not in source, f"{path} must not call {token}"


def test_data_access_durable_runtime_rows_do_not_store_provider_secrets() -> None:
    checked_models = [
        DataAccessRequest,
        DataAccessRequestItem,
        DataAccessBinding,
        DataAccessCompiledEffect,
        DataAccessEffectRun,
    ]
    forbidden_column_fragments = (
        "access_token",
        "client_email",
        "credential",
        "oauth",
        "private_key",
        "raw_policy",
        "raw_response",
        "secret",
        "service_account",
    )

    for model in checked_models:
        column_names = {column.name for column in model.__table__.columns}
        for fragment in forbidden_column_fragments:
            assert all(fragment not in name for name in column_names), (
                f"{model.__name__} must not persist provider secret material "
                f"via columns matching {fragment!r}: {sorted(column_names)}"
            )


def test_data_access_runtime_paths_have_no_project_iam_or_broad_scan_tokens() -> None:
    checked_paths = [
        SERVER_ROOT / "app/domains/data_access/reconciler.py",
        SERVER_ROOT / "app/integrations/drivers/bigquery_data_access.py",
    ]
    forbidden_tokens = (
        "list_datasets(",
        "list_all_datasets",
        "datasets().list",
        ".datasets().list",
        "list_resources(",
        "list_all_resources",
        "scan_resources(",
        "get_project_iam",
        "set_project_iam",
        "get_project_policy",
        "set_project_policy",
        "projects().getIamPolicy",
        "projects().setIamPolicy",
        "roles/bigquery.jobUser",
        "bigquery.jobUser",
        "grant_access(",
        "revoke_access(",
    )

    for path in checked_paths:
        source = path.read_text(encoding="utf-8")
        for token in forbidden_tokens:
            assert token not in source, f"{path} must not contain {token}"


def test_data_access_contract_documents_bigquery_least_privilege() -> None:
    contract = (
        SERVER_ROOT.parent / "docs" / "architecture" / "DATA_ACCESS.md"
    ).read_text(encoding="utf-8")

    assert "bigquery.datasets.getIamPolicy" in contract
    assert "bigquery.datasets.setIamPolicy" in contract
    assert "https://www.googleapis.com/auth/bigquery" in contract
    assert "bigquery.readonly" in contract
    assert "roles/bigquery.jobUser" in contract
    assert "service_account_json" in contract
    assert "datasetView=ACL" in contract
    assert "UPDATE_ACL" in contract
    assert "groupByEmail" in contract
    assert "Conditional access entries" in contract
    assert "preserved byte-for-byte" in contract
    assert "If-Match" in contract
    assert "Terraform-managed BigQuery IAM" in contract
    assert "provider 409/412" in contract