"""End-to-end tests for /api/v1/integrations + /api/v1/platform-settings.

Cover the happy path + the main failure modes:
  * unknown driver, duplicate slug, invalid slug, missing required field
  * secrets never leaked in responses (redacted form)
  * admin guard enforced
  * is_primary toggle unsets siblings
  * approval-flow sub-CRUD
  * platform-settings CRUD
"""
from __future__ import annotations

import pytest
from sqlalchemy import select

from app.audit.models import AuditLog
from app.domains.data_access import constants as data_access_constants
from app.domains.data_access.models import (
    OWNER_RESOLUTION_UNRESOLVED,
    OWNER_SOURCE_BOOTSTRAP,
    RESOURCE_STATUS_ACTIVE,
    DataAccessResource,
)
from app.integrations.models import Integration, PlatformSetting
from app.integrations.schemas import IntegrationTestResult


def _admin_headers(admin_user):
    _, token = admin_user
    return {"Authorization": f"Bearer {token}"}


@pytest.mark.asyncio
async def test_connection_scrubs_token_prefix_from_response_storage_and_audit(
    client, admin_user, db_session, monkeypatch,
):
    """The service boundary is fail-safe even for a leaking driver."""
    from app.integrations.drivers.feishu_chat_approval import (
        FeishuChatApprovalDriver,
    )

    leaked_prefix = "tenant-s..."

    async def _leaking_probe(_self):
        return IntegrationTestResult(
            ok=True,
            message=f"OK · token_prefix={leaked_prefix} · healthy",
            detail={"token_prefix": leaked_prefix, "probe": "green"},
        )

    monkeypatch.setattr(
        FeishuChatApprovalDriver, "test_connection", _leaking_probe,
    )
    created = await client.post(
        "/api/v1/integrations",
        json={
            "kind": "chat_approval",
            "provider": "feishu",
            "name": "Feishu scrub boundary",
            "slug": "feishu-scrub-boundary",
            "config": {"app_id": "cli-test", "app_secret": "secret"},
        },
        headers=_admin_headers(admin_user),
    )
    assert created.status_code == 200, created.text

    response = await client.post(
        "/api/v1/integrations/feishu-scrub-boundary/test",
        headers=_admin_headers(admin_user),
    )
    assert response.status_code == 200, response.text
    response_text = response.text
    assert leaked_prefix not in response_text
    assert "<redacted>" in response_text
    assert "token_prefix" not in response.json()["data"]["detail"]

    row = await db_session.scalar(select(Integration).where(
        Integration.slug == "feishu-scrub-boundary",
    ))
    assert row is not None
    await db_session.refresh(row)
    assert leaked_prefix not in (row.last_test_message or "")
    assert "<redacted>" in (row.last_test_message or "")

    audit = await db_session.scalar(
        select(AuditLog)
        .where(
            AuditLog.action == "integration.test_connection",
            AuditLog.resource_id == str(row.id),
        )
        .order_by(AuditLog.created_at.desc())
    )
    assert audit is not None
    assert leaked_prefix not in str(audit.detail)
    assert "<redacted>" in str(audit.detail)


@pytest.mark.asyncio
async def test_connection_scrubs_provider_credentials_everywhere(
    client, admin_user, db_session, monkeypatch,
):
    """Provider-shaped secrets are absent from API, status row, and audit."""
    from app.integrations.drivers.feishu_chat_approval import (
        FeishuChatApprovalDriver,
    )

    leaked = {
        "secret_id": "tencent-secret-id-value",
        "secret_key": "tencent-secret-key-value",
        "authorization": "bearer-token-value-123456",
        "signature": "amz-signature-value-123456",
        "oss_token": "oss-security-token-value-123456",
    }
    raw_error = (
        f"SecretId={leaked['secret_id']} SecretKey={leaked['secret_key']} "
        f"Authorization: Bearer {leaked['authorization']} "
        "https://example.invalid/object?X-Amz-Signature="
        f"{leaked['signature']} "
        f"x-oss-security-token:{leaked['oss_token']}"
    )

    async def _leaking_probe(_self):
        return IntegrationTestResult(
            ok=False,
            message=raw_error,
            detail={"nested": {"raw_error": raw_error}},
        )

    monkeypatch.setattr(
        FeishuChatApprovalDriver, "test_connection", _leaking_probe,
    )
    created = await client.post(
        "/api/v1/integrations",
        json={
            "kind": "chat_approval",
            "provider": "feishu",
            "name": "Feishu provider scrub boundary",
            "slug": "feishu-provider-scrub-boundary",
            "config": {"app_id": "cli-test", "app_secret": "secret"},
        },
        headers=_admin_headers(admin_user),
    )
    assert created.status_code == 200, created.text

    response = await client.post(
        "/api/v1/integrations/feishu-provider-scrub-boundary/test",
        headers=_admin_headers(admin_user),
    )
    assert response.status_code == 200, response.text

    row = await db_session.scalar(select(Integration).where(
        Integration.slug == "feishu-provider-scrub-boundary",
    ))
    assert row is not None
    await db_session.refresh(row)
    audit = await db_session.scalar(
        select(AuditLog)
        .where(
            AuditLog.action == "integration.test_connection",
            AuditLog.resource_id == str(row.id),
        )
        .order_by(AuditLog.created_at.desc())
    )
    assert audit is not None

    rendered_surfaces = (
        response.text,
        row.last_test_message or "",
        str(audit.detail),
    )
    for surface in rendered_surfaces:
        for secret in leaked.values():
            assert secret not in surface
        assert "<redacted>" in surface


@pytest.mark.asyncio
async def test_list_driver_kinds_admin_only(client, admin_user, employee_user):
    # Admin can list drivers
    r = await client.get("/api/v1/integrations/kinds", headers=_admin_headers(admin_user))
    assert r.status_code == 200
    data = r.json()["data"]
    assert any(d["provider"] == "successfactors" for d in data)
    # Every driver advertises a field list
    for d in data:
        assert "fields" in d
        assert "binding_fields" in d

    # Employee is rejected
    _, emp_token = employee_user
    r = await client.get(
        "/api/v1/integrations/kinds",
        headers={"Authorization": f"Bearer {emp_token}"},
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_create_read_update_delete_integration(
    client, admin_user, db_session,
):
    # 1. Create a Feishu chat_approval integration
    payload = {
        "kind": "chat_approval",
        "provider": "feishu",
        "name": "Feishu Prod",
        "slug": "feishu-prod",
        "description": "Primary Feishu tenant",
        "enabled": True,
        "is_primary": True,
        "config": {
            "app_id": "cli_feishu_test",
            "app_secret": "super-secret-1",
            "event_encrypt_key": "encrypt-key-1",
        },
    }
    r = await client.post(
        "/api/v1/integrations", json=payload, headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200, r.text
    created = r.json()["data"]
    assert created["slug"] == "feishu-prod"
    # Secrets NEVER appear in the response — only *_set flags
    assert "app_secret" not in created["config"]
    assert created["config"]["app_secret_set"] is True
    assert created["config"]["event_encrypt_key_set"] is True
    assert created["config"]["app_id"] == "cli_feishu_test"
    assert "secrets_encrypted" not in created
    assert "secrets_nonce" not in created

    # 2. GET by slug
    r = await client.get(
        "/api/v1/integrations/feishu-prod", headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    assert r.json()["data"]["slug"] == "feishu-prod"

    # 3. Update — rotate app_secret, keep event_encrypt_key (empty string means keep)
    r = await client.patch(
        "/api/v1/integrations/feishu-prod",
        json={"config": {"app_secret": "super-secret-2"}},
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200

    # Verify in DB that the encrypted blob changed
    row = await db_session.scalar(
        select(Integration).where(Integration.slug == "feishu-prod")
    )
    await db_session.refresh(row)
    assert row.secrets_encrypted  # still encrypted
    # (Can't assert plaintext without decrypting; sufficient that secret_set remains true)

    # 4. Delete
    r = await client.delete(
        "/api/v1/integrations/feishu-prod", headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    r = await client.get(
        "/api/v1/integrations/feishu-prod", headers=_admin_headers(admin_user),
    )
    assert r.status_code == 404


@pytest.mark.asyncio
async def test_delete_integration_with_data_access_resources_is_rejected(
    client,
    admin_user,
    db_session,
):
    integration = Integration(
        kind=data_access_constants.DOMAIN_DATA_ACCESS,
        provider=data_access_constants.PROVIDER_BIGQUERY,
        name="DAX BigQuery",
        slug="dax-bigquery-delete-guard",
        enabled=True,
        is_primary=False,
    )
    db_session.add(integration)
    await db_session.flush()
    resource = DataAccessResource(
        integration_id=str(integration.id),
        provider=data_access_constants.PROVIDER_BIGQUERY,
        resource_type=data_access_constants.RESOURCE_TYPE_DATASET,
        project_id="helix-data-prod",
        dataset_id="finance_2026",
        requestable=True,
        status=RESOURCE_STATUS_ACTIVE,
        owner_resolution_status=OWNER_RESOLUTION_UNRESOLVED,
        owner_source=OWNER_SOURCE_BOOTSTRAP,
        source_version="bootstrap-2026-06-10",
        source_etag="etag-001",
    )
    db_session.add(resource)
    await db_session.flush()

    r = await client.delete(
        "/api/v1/integrations/dax-bigquery-delete-guard",
        headers=_admin_headers(admin_user),
    )

    assert r.status_code == 409, r.text
    assert r.json()["errors"][0]["code"] == "INTEGRATION_HAS_HISTORY"
    assert "durable history/resource" in r.json()["errors"][0]["message"]
    assert await db_session.scalar(
        select(Integration).where(Integration.id == integration.id),
    )
    assert await db_session.scalar(
        select(DataAccessResource).where(DataAccessResource.id == resource.id),
    )


@pytest.mark.asyncio
async def test_bigquery_data_access_integration_persists_apply_gate_config(
    client,
    admin_user,
    db_session,
):
    service_account_json = (
        '{"type":"service_account","client_email":"dax@example.iam.gserviceaccount.com",'
        '"private_key":"-----BEGIN PRIVATE KEY-----\\nredacted\\n-----END PRIVATE KEY-----\\n",'
        '"project_id":"helix-dax-admin"}'
    )

    r = await client.post(
        "/api/v1/integrations",
        json={
            "kind": data_access_constants.DOMAIN_DATA_ACCESS,
            "provider": data_access_constants.PROVIDER_BIGQUERY,
            "name": "DAX BigQuery",
            "slug": "dax-bigquery-prod",
            "enabled": True,
            "is_primary": False,
            "config": {
                "service_account_json": service_account_json,
                "apply_mode_enabled": "true",
            },
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200, r.text
    created = r.json()["data"]
    assert created["config"]["service_account_json_set"] is True
    assert created["config"]["apply_mode_enabled"] == "true"

    row = await db_session.scalar(
        select(Integration).where(Integration.slug == "dax-bigquery-prod"),
    )
    assert row is not None
    assert row.config_public == {
        "apply_mode_enabled": "true",
    }

    r = await client.patch(
        "/api/v1/integrations/dax-bigquery-prod",
        json={
            "config": {
                "apply_mode_enabled": "false",
            },
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200, r.text
    await db_session.refresh(row)
    assert row.config_public == {
        "apply_mode_enabled": "false",
    }


@pytest.mark.asyncio
async def test_create_rejects_unknown_driver(client, admin_user):
    r = await client.post(
        "/api/v1/integrations",
        json={
            "kind": "hris",
            "provider": "workday",  # not registered
            "name": "Workday",
            "slug": "workday-prod",
            "config": {},
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 400
    assert r.json()["errors"][0]["code"] == "INTEGRATION_DRIVER_UNKNOWN"


@pytest.mark.asyncio
async def test_create_rejects_duplicate_slug(client, admin_user):
    base = {
        "kind": "software_provisioning",
        "provider": "manual",
        "name": "Manual A",
        "slug": "manual-dup",
        "config": {},
    }
    r = await client.post("/api/v1/integrations", json=base, headers=_admin_headers(admin_user))
    assert r.status_code == 200
    r = await client.post(
        "/api/v1/integrations",
        json={**base, "name": "Manual B"},
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 409 or r.json()["errors"][0]["code"] == "CONFLICT"


@pytest.mark.asyncio
async def test_create_rejects_invalid_slug(client, admin_user):
    r = await client.post(
        "/api/v1/integrations",
        json={
            "kind": "software_provisioning",
            "provider": "manual",
            "name": "Bad",
            "slug": "Invalid Slug!",  # uppercase + space + !
            "config": {},
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 400
    assert r.json()["errors"][0]["code"] == "INVALID_SLUG"


@pytest.mark.asyncio
async def test_create_rejects_missing_required_field(client, admin_user):
    # Feishu requires app_id + app_secret
    r = await client.post(
        "/api/v1/integrations",
        json={
            "kind": "chat_approval",
            "provider": "feishu",
            "name": "Bad",
            "slug": "feishu-bad",
            "config": {},  # missing app_id + app_secret
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 400
    assert r.json()["errors"][0]["code"] == "INTEGRATION_CONFIG_REQUIRED"


@pytest.mark.asyncio
async def test_is_primary_toggle_unsets_siblings(client, admin_user, db_session):
    """Setting one integration primary must clear is_primary on its kind-siblings."""
    await client.post(
        "/api/v1/integrations",
        json={
            "kind": "hris", "provider": "successfactors",
            "name": "SF Dev", "slug": "sf-dev", "is_primary": True,
            "config": {"auth_method": "basic", "base_url": "https://dev.sapsf.cn/odata/v2",
                       "username": "u", "password": "p"},
        },
        headers=_admin_headers(admin_user),
    )
    await client.post(
        "/api/v1/integrations",
        json={
            "kind": "hris", "provider": "successfactors",
            "name": "SF Prod", "slug": "sf-prod", "is_primary": True,
            "config": {"auth_method": "basic", "base_url": "https://prod.sapsf.cn/odata/v2",
                       "username": "u2", "password": "p2"},
        },
        headers=_admin_headers(admin_user),
    )
    # Only sf-prod should be primary now
    rows = (
        await db_session.execute(
            select(Integration).where(Integration.kind == "hris")
        )
    ).scalars().all()
    await db_session.flush()  # ensure we see the latest
    primary_slugs = [r.slug for r in rows if r.is_primary]
    assert primary_slugs == ["sf-prod"], f"expected only sf-prod primary, got {primary_slugs}"


@pytest.mark.asyncio
async def test_approval_flows_crud(client, admin_user, db_session):
    # Bootstrap a Feishu integration
    await client.post(
        "/api/v1/integrations",
        json={
            "kind": "chat_approval", "provider": "feishu",
            "name": "Feishu", "slug": "feishu-flows-test", "is_primary": True,
            "config": {"app_id": "cli_x", "app_secret": "s"},
        },
        headers=_admin_headers(admin_user),
    )

    # Create a flow
    r = await client.put(
        "/api/v1/integrations/feishu-flows-test/approval-flows",
        json={
            "flow_key": "custom_flow_1",
            "label": "自定义流程 1",
            "external_code": "ABC-123",
            "active": True,
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    flow_id = r.json()["data"]["id"]

    # List
    r = await client.get(
        "/api/v1/integrations/feishu-flows-test/approval-flows",
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    flows = r.json()["data"]
    assert any(f["flow_key"] == "custom_flow_1" for f in flows)

    # Update (same endpoint — upsert semantics)
    r = await client.put(
        "/api/v1/integrations/feishu-flows-test/approval-flows",
        json={
            "flow_key": "custom_flow_1",
            "label": "新标签",
            "external_code": "ABC-123",
            "active": False,
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    assert r.json()["data"]["label"] == "新标签"
    assert r.json()["data"]["active"] is False

    # Delete
    r = await client.delete(
        f"/api/v1/integrations/feishu-flows-test/approval-flows/{flow_id}",
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200

    # Non-chat_approval integration can't host approval flows
    await client.post(
        "/api/v1/integrations",
        json={
            "kind": "software_provisioning", "provider": "manual",
            "name": "Manual", "slug": "manual-flows-test",
            "config": {},
        },
        headers=_admin_headers(admin_user),
    )
    r = await client.put(
        "/api/v1/integrations/manual-flows-test/approval-flows",
        json={
            "flow_key": "nope", "label": "x", "external_code": "y", "active": True,
        },
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 400
    assert r.json()["errors"][0]["code"] == "APPROVAL_FLOW_WRONG_KIND"


@pytest.mark.asyncio
async def test_all_mutation_endpoints_require_admin(client, employee_user):
    _, token = employee_user
    h = {"Authorization": f"Bearer {token}"}
    # POST create
    r = await client.post(
        "/api/v1/integrations",
        json={"kind": "software_provisioning", "provider": "manual", "name": "x", "slug": "y", "config": {}},
        headers=h,
    )
    assert r.status_code == 403
    # PUT approval flow
    r = await client.put(
        "/api/v1/integrations/anything/approval-flows",
        json={"flow_key": "x", "label": "y", "external_code": "z", "active": True},
        headers=h,
    )
    assert r.status_code == 403
    # DELETE
    r = await client.delete("/api/v1/integrations/anything", headers=h)
    assert r.status_code == 403
    # POST /test
    r = await client.post("/api/v1/integrations/anything/test", headers=h)
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_platform_settings_crud(client, admin_user, db_session):
    # Create
    r = await client.put(
        "/api/v1/platform-settings/offboarding.grace_period_days",
        json={"value": 7, "description": "离职缓冲天数"},
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    assert r.json()["data"]["value"] == 7

    # List
    r = await client.get(
        "/api/v1/platform-settings", headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    keys = [i["key"] for i in r.json()["data"]]
    assert "offboarding.grace_period_days" in keys

    # Get one
    r = await client.get(
        "/api/v1/platform-settings/offboarding.grace_period_days",
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    assert r.json()["data"]["value"] == 7

    # Update (upsert semantics)
    r = await client.put(
        "/api/v1/platform-settings/offboarding.grace_period_days",
        json={"value": 14},
        headers=_admin_headers(admin_user),
    )
    assert r.json()["data"]["value"] == 14

    # Delete
    r = await client.delete(
        "/api/v1/platform-settings/offboarding.grace_period_days",
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    r = await client.get(
        "/api/v1/platform-settings/offboarding.grace_period_days",
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 404

    # DB check: JSON value for structured data
    r = await client.put(
        "/api/v1/platform-settings/notifications.defaults",
        json={"value": {"channels": ["email", "feishu"], "enabled": True}},
        headers=_admin_headers(admin_user),
    )
    assert r.status_code == 200
    row = await db_session.scalar(
        select(PlatformSetting).where(PlatformSetting.key == "notifications.defaults")
    )
    await db_session.refresh(row)
    assert row.value == {"channels": ["email", "feishu"], "enabled": True}


@pytest.mark.asyncio
async def test_integration_response_never_contains_plaintext_secrets(
    client, admin_user, db_session,
):
    """Explicit negative: secret values never round-trip through the API."""
    await client.post(
        "/api/v1/integrations",
        json={
            "kind": "chat_approval", "provider": "feishu",
            "name": "Feishu", "slug": "feishu-leak-check",
            "config": {
                "app_id": "cli_leakcheck",
                "app_secret": "SUPER_SENSITIVE_XYZ123",
                "event_encrypt_key": "ENCRYPT_KEY_VALUE_456",
            },
        },
        headers=_admin_headers(admin_user),
    )

    # Check list, get, and raw DB — none should contain plaintext "SUPER_SENSITIVE..."
    r = await client.get(
        "/api/v1/integrations", headers=_admin_headers(admin_user),
    )
    body = r.text
    assert "SUPER_SENSITIVE_XYZ123" not in body
    assert "ENCRYPT_KEY_VALUE_456" not in body

    r = await client.get(
        "/api/v1/integrations/feishu-leak-check",
        headers=_admin_headers(admin_user),
    )
    body = r.text
    assert "SUPER_SENSITIVE_XYZ123" not in body
    assert "ENCRYPT_KEY_VALUE_456" not in body

    # But the _set flags are surfaced
    data = r.json()["data"]
    assert data["config"]["app_secret_set"] is True
    assert data["config"]["event_encrypt_key_set"] is True

    # Raw DB: only ciphertext, no plaintext
    row = await db_session.scalar(
        select(Integration).where(Integration.slug == "feishu-leak-check")
    )
    await db_session.refresh(row)
    assert row.secrets_encrypted
    assert "SUPER_SENSITIVE_XYZ123" not in (row.secrets_encrypted or "")
    assert "ENCRYPT_KEY_VALUE_456" not in (row.secrets_encrypted or "")