from types import SimpleNamespace
from unittest.mock import AsyncMock
from uuid import uuid4

import pytest
from src.domain.connectors.catalog import get_action, slugs_implementing
from src.domain.connectors.resource_resolve import (
    ResourceConnection,
    ResourceError,
    resolve_resource,
)
from src.domain.connectors.service import ConnectorService

CAPS = {"integrations.execute"}
READ = "spreadsheet.rows.read"


def sheet(**overrides):
    values = dict(
        connection_id=uuid4(),
        connector_key="google_sheets",
        has_secrets=True,
        scopes={"https://www.googleapis.com/auth/spreadsheets"},
        permissions={
            s: "allow"
            for verb in (READ, "spreadsheet.rows.append", "spreadsheet.rows.update")
            for s in slugs_implementing(verb)
        },
        config={
            "spreadsheet_map": {
                "orders": {
                    "spreadsheet_id": "customer-sheet-id",
                    "sheet_name": "Bob's Orders",
                    "column_count": 2,
                    "start_row": 2,
                }
            }
        },
    )
    values.update(overrides)
    return ResourceConnection(**values)


def resolve(rows, payload=None, capability=READ, **kwargs):
    return resolve_resource(
        capability, payload or {"spreadsheet": "orders"}, rows, enabled_capabilities=CAPS, **kwargs
    )


def test_sheet_mapping_is_private_raw_values_and_row_offsets():
    row = sheet()
    plan = resolve(
        [row],
        {
            "spreadsheet": "orders",
            "offset": 2,
            "limit": 3,
            "metadata": {"range": "malicious", "value_input_option": "USER_ENTERED"},
        },
    )
    assert plan.input == {"spreadsheet_id": "customer-sheet-id", "range": "'Bob''s Orders'!A4:B6"}
    plan = resolve(
        [row], {"spreadsheet": "orders", "rows": [["=1+1", True]]}, "spreadsheet.rows.append"
    )
    assert plan.input["value_input_option"] == "RAW"
    assert plan.input["range"] == "'Bob''s Orders'!A2:B"
    assert "metadata" not in plan.input


@pytest.mark.parametrize(
    "payload",
    [
        {"spreadsheet": "orders", "range": "A1"},
        {"spreadsheet": "orders", "offset": True},
        {"spreadsheet": "https://sheets.google.com/id"},
        {"spreadsheet": "orders", "limit": 0},
    ],
)
def test_invalid_generic_input(payload):
    with pytest.raises(ResourceError, match="invalid_input"):
        resolve([sheet()], payload)


def test_binding_ambiguity_hint_and_explicit_selection():
    first, second = sheet(), sheet()
    with pytest.raises(ResourceError, match="binding_ambiguous"):
        resolve([first, second], provider_hint="google_sheets")
    assert (
        resolve([first, second], binding_connection_id=second.connection_id).connection_id
        == second.connection_id
    )
    with pytest.raises(ResourceError, match="capability_not_bound"):
        resolve([first], provider_hint="amazon_s3", binding_connection_id=first.connection_id)


def test_unique_unbound_and_platform_resources_cannot_supply_customer_work():
    with pytest.raises(ResourceError, match="resource_unbound"):
        resolve([sheet(config={"spreadsheet_id": "unaliased"})])
    with pytest.raises(ResourceError, match="capability_not_bound"):
        resolve([sheet(execution_domain="P")])
    with pytest.raises(ResourceError, match="capability_not_bound"):
        resolve([sheet(connector_key="google_docs")])


@pytest.mark.parametrize(
    "override,code",
    [
        ({"status": "disabled"}, "connection_inactive"),
        ({"has_secrets": False}, "credentials_missing"),
        ({"scopes": set()}, "scope_missing"),
        ({"permissions": {}}, "connector_action_denied"),
    ],
)
def test_explicit_connection_failures(override, code):
    row = sheet(**override)
    with pytest.raises(ResourceError, match=code):
        resolve([row], binding_connection_id=row.connection_id)


@pytest.mark.asyncio
async def test_authorization_is_separate_and_wrong_workspace_is_excluded():
    workspace, project, connection = uuid4(), uuid4(), uuid4()
    snapshot = sheet()
    row = SimpleNamespace(
        id=connection,
        workspace_id=workspace,
        connector_id=uuid4(),
        config_json=snapshot.config,
        status="active",
        secrets_enc="encrypted",
        scopes=list(snapshot.scopes),
    )
    repo = SimpleNamespace(
        list_connections_for_project=AsyncMock(return_value=[row]),
        get_connector_by_id=AsyncMock(return_value=SimpleNamespace(key="google_sheets")),
        get_permission=AsyncMock(return_value=SimpleNamespace(mode="ask")),
    )
    service = ConnectorService(repo=repo, cloud_client=SimpleNamespace())
    service.authorize_execution = AsyncMock(
        return_value=SimpleNamespace(allowed=False, reason_code="approval_required")
    )
    enqueue = AsyncMock(
        return_value={"ok": True, "job_id": "private-job", "output": {"secret": "x"}}
    )
    kwargs = dict(
        workspace_id=workspace,
        project_id=project,
        capability=READ,
        payload={"spreadsheet": "orders"},
        enabled_capabilities=CAPS,
        enqueue_execute=enqueue,
    )
    assert (await service.invoke_resource(**kwargs))["reason_code"] == "approval_required"
    enqueue.assert_not_called()
    service.authorize_execution.return_value = SimpleNamespace(allowed=True)
    assert await service.invoke_resource(**kwargs) == {"capability": READ, "status": "queued"}
    assert enqueue.call_args.kwargs["action"] == "google_sheets.read_values"
    row.workspace_id = uuid4()
    service.authorize_execution.reset_mock()
    assert (await service.invoke_resource(**kwargs))["reason_code"] == "capability_not_bound"
    service.authorize_execution.assert_not_called()


def test_resource_defaults_are_derived_and_routes_are_registered():
    from src.domain.connectors.public_gateway import PUBLIC_INVOCABLE_CAPABILITIES
    from src.domain.connectors.resource_contracts import RESOURCE_INPUTS
    from src.domain.manifest.service import _connectors_manifest_envelope
    from src.main import app

    paths = {r.path for r in app.routes}
    envelope = _connectors_manifest_envelope(items={})
    for verb in RESOURCE_INPUTS:
        assert verb in PUBLIC_INVOCABLE_CAPABILITIES
        assert f"/api/projects/{{project_id}}/capabilities/{verb}" in paths
        cap = envelope["capabilities"][verb]
        assert cap["implemented_by"] == list(slugs_implementing(verb))
        assert cap["role"] == "default" and cap["invocable"]
        for slug in cap["implemented_by"]:
            assert get_action(slug)[1].required_capabilities == ("integrations.execute",)


def store(provider="cloudflare_r2", **overrides):
    values = dict(
        connection_id=uuid4(),
        connector_key=provider,
        has_secrets=True,
        permissions={
            f"{provider}.{op}": "allow"
            for op in ("put_object", "get_object", "delete_object", "head_object", "list_objects")
        },
        config={"storage_map": {"uploads": {"bucket_name": "customer-bucket", "prefix": "app"}}},
    )
    values.update(overrides)
    return ResourceConnection(**values)


@pytest.mark.parametrize("provider", ["cloudflare_r2", "amazon_s3"])
@pytest.mark.parametrize(
    "verb,operation",
    [
        ("put", "put_object"),
        ("get", "get_object"),
        ("delete", "delete_object"),
        ("exists", "head_object"),
        ("list", "list_objects"),
    ],
)
def test_storage_same_contract_for_both_customer_providers(provider, verb, operation):
    payload = {"storage": "uploads"}
    if verb != "list":
        payload["key"] = "a.txt"
    if verb == "put":
        payload["content_base64"] = "aGk="
    plan = resolve([store(provider)], payload, f"storage.file.{verb}")
    assert plan.action == f"{provider}.{operation}"
    assert plan.input["bucket_name"] == "customer-bucket"
    assert plan.input.get("key", plan.input.get("prefix")) == (
        "app/" if verb == "list" else "app/a.txt"
    )
    assert "storage" not in plan.input


def test_storage_never_uses_managed_r2_and_hint_is_only_family_filter():
    payload = {"storage": "uploads", "key": "a"}
    for rows in ([], [store("p2hs_managed_r2")], [store(execution_domain="P")]):
        with pytest.raises(ResourceError, match="capability_not_bound"):
            resolve(rows, payload, "storage.file.get")
    first, second, s3 = store(), store(), store("amazon_s3")
    with pytest.raises(ResourceError, match="binding_ambiguous"):
        resolve([first, second, s3], payload, "storage.file.get", provider_hint="cloudflare_r2")
    assert (
        resolve([first, s3], payload, "storage.file.get", provider_hint="amazon_s3").connection_id
        == s3.connection_id
    )
    assert (
        resolve(
            [first, second], payload, "storage.file.get", binding_connection_id=second.connection_id
        ).connection_id
        == second.connection_id
    )
    with pytest.raises(ResourceError, match="resource_unbound"):
        resolve([store(config={"bucket_name": "unaliased"})], payload, "storage.file.get")


@pytest.mark.parametrize("key", ["../x", "folder/../x", "/absolute", "x\\y", "\x00"])
def test_storage_binding_cannot_escape_prefix(key):
    with pytest.raises(ResourceError, match="invalid_input"):
        resolve([store()], {"storage": "uploads", "key": key}, "storage.file.get")


def test_put_rejects_provider_fields_bad_base64_and_large_objects():
    for extra in (
        {"content_base64": "!"},
        {"content_base64": "A" * 349528},
        {"content_base64": "", "endpoint": "https://internal"},
        {"content_base64": "", "content_type": "text/plain\r\nx: y"},
    ):
        with pytest.raises(ResourceError, match="invalid_input"):
            resolve([store()], {"storage": "uploads", "key": "a", **extra}, "storage.file.put")


def test_public_bff_allowlist_matches_resource_catalog():
    import re
    from pathlib import Path

    from src.domain.connectors.resource_contracts import RESOURCE_INPUTS

    path = (
        Path(__file__).resolve().parents[2]
        / "P2HS-Dash/server/api/projects/[projectId]/capabilities/[capability].post.ts"
    )
    if not path.exists():
        pytest.skip("Dashboard sibling checkout unavailable")
    source = path.read_text(encoding="utf-8")
    block = re.search(r"const PUBLIC_BFF_PROXY = new Set\(\[(.*?)\]\)", source, re.S)[1]
    assert set(RESOURCE_INPUTS) <= set(re.findall(r"'([^']+)'", block))


def test_private_resource_routes_keep_auth_and_project_billing_dependencies():
    from fastapi import FastAPI, HTTPException
    from fastapi.testclient import TestClient
    from src.core.dependencies import require_active_billing_for_project_id
    from src.domain.auth.router import get_current_user
    from src.domain.connectors.resource_contracts import RESOURCE_INPUTS
    from src.domain.connectors.router import capability_router

    api = FastAPI()
    api.include_router(capability_router)
    project = uuid4()

    def deny_auth():
        raise HTTPException(401, "Authentication required")

    def deny_project():
        raise HTTPException(403, "Project access denied")

    api.dependency_overrides[require_active_billing_for_project_id] = lambda: project
    api.dependency_overrides[get_current_user] = deny_auth
    client = TestClient(api)
    for verb in RESOURCE_INPUTS:
        path = f"/api/projects/{project}/capabilities/{verb}"
        assert client.post(path, json={}).status_code == 401
    api.dependency_overrides[require_active_billing_for_project_id] = deny_project
    for verb in RESOURCE_INPUTS:
        path = f"/api/projects/{project}/capabilities/{verb}"
        assert client.post(path, json={}).status_code == 403