"""Settings › Secrets (app/routers/secrets.py, app/crud/secrets.py): stored
encrypted, never returned, usable only where they travel to another system,
and masked out of everything a run hands back."""

import uuid

from sqlalchemy import select

from app.core.auth import create_access_token
from tests.conftest import AsyncSessionLocal
from app.crud.secrets import MASK, Scrubber, connector_config, fill_templates, referenced_names
from app.models.audit import AuditLogEntry
from app.models.reporting import Connector
from app.models.secret import Secret

VALUE = "sk_live_9f8e7d6c5b4a"


async def _secret(client, auth_headers, name="PAY_KEY", value=VALUE, description="Gateway key"):
    resp = await client.post("/secrets", headers=auth_headers, json={"name": name, "value": value, "description": description})
    assert resp.status_code == 201, resp.text
    return resp.json()


async def _run(client, auth_headers, nodes, context=None):
    task = await client.post("/tasks", headers=auth_headers, json={"name": "Secret task", "nodes": nodes})
    assert task.status_code == 201, task.text
    resp = await client.post(f"/tasks/{task.json()['id']}/run", headers=auth_headers, json={"context": context or {}, "allow_draft": True})
    assert resp.status_code == 200, resp.text
    return task.json()["id"], resp.json()


async def test_secrets_are_stored_encrypted_and_never_returned(client, auth_headers, test_tenant):
    created = await _secret(client, auth_headers)
    assert VALUE not in str(created) and "value" not in created
    assert created["name"] == "PAY_KEY" and created["description"] == "Gateway key"

    listed = (await client.get("/secrets", headers=auth_headers)).json()
    assert [s["name"] for s in listed] == ["PAY_KEY"] and VALUE not in str(listed)
    assert (await client.get("/secrets/names", headers=auth_headers)).json() == ["PAY_KEY"]

    async with AsyncSessionLocal() as db:
        stored = (await db.execute(select(Secret.value_encrypted).where(Secret.id == uuid.UUID(created["id"])))).scalar_one()
    assert VALUE not in stored  # ciphertext at rest

    dup = await client.post("/secrets", headers=auth_headers, json={"name": "PAY_KEY", "value": "x"})
    assert dup.status_code == 409
    for bad in ("1ABC", "has-dash", "with space", ""):
        assert (await client.post("/secrets", headers=auth_headers, json={"name": bad, "value": "x"})).status_code == 422

    replaced = await client.patch(f"/secrets/{created['id']}", headers=auth_headers, json={"value": "new_value_1234"})
    assert replaced.status_code == 200 and "new_value_1234" not in replaced.text

    async with AsyncSessionLocal() as db:
        actions = (await db.execute(select(AuditLogEntry.action, AuditLogEntry.details).where(AuditLogEntry.tenant_id == test_tenant["tenant_id"], AuditLogEntry.entity_type == "secret"))).all()
    assert {a for a, _ in actions} == {"secret.create", "secret.update"}
    assert VALUE not in str(actions) and "new_value_1234" not in str(actions)

    assert (await client.delete(f"/secrets/{created['id']}", headers=auth_headers)).status_code == 204
    assert (await client.get("/secrets", headers=auth_headers)).json() == []


async def test_portal_users_cannot_reach_secrets(client, test_tenant):
    token = create_access_token(tenant_id=test_tenant["tenant_id"], user_id=test_tenant["admin_id"], vendor_id=uuid.uuid4())
    headers = {"Authorization": f"Bearer {token}"}
    assert (await client.get("/secrets", headers=headers)).status_code in (401, 403)
    assert (await client.get("/secrets/names", headers=headers)).status_code in (401, 403)


async def test_outbound_node_uses_a_secret_and_every_echo_is_masked(client, auth_headers):
    await _secret(client, auth_headers)
    code = (
        "import sys, json\nctx = json.load(sys.stdin)\nkey = ctx['api_key']\n"
        "print('calling with', key)\n"
        "print(json.dumps({'length': len(key), 'echo': key, 'starts': key[:3]}))"
    )
    nodes = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "call"}]},
        {
            "key": "call", "type": "custom_code", "label": "Call gateway", "language": "python", "code": code,
            "input_params": [{"name": "api_key", "type": "string", "source_field": "secret.PAY_KEY"}],
            "output_params": [{"name": "outcome", "type": "object", "source": "*"}],
        },
    ]
    task_id, result = await _run(client, auth_headers, nodes)
    assert result["ok"] is True, result
    outcome = result["context"]["outcome"]
    assert outcome["length"] == len(VALUE)  # the script really had it...
    assert outcome["echo"] == MASK  # ...but it never comes back
    assert VALUE not in str(result)
    assert "secret" not in result["context"]  # never part of the shared context

    used = (await client.get("/secrets", headers=auth_headers)).json()[0]["used_by"]
    assert used == [{"kind": "task", "id": task_id, "name": "Secret task"}]


async def test_secrets_are_refused_where_people_could_read_them(client, auth_headers):
    await _secret(client, auth_headers)
    master = await client.post("/masters", headers=auth_headers, json={"name": f"Keys {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "note", "type": "varchar"}]})
    nodes = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "save"}]},
        {
            "key": "save", "type": "data_update", "label": "Save", "datasource_type": "master", "master_id": master.json()["id"],
            "input_params": [{"name": "note", "type": "string", "required": False, "value": "key is {{secret.PAY_KEY}}"}],
        },
    ]
    _, result = await _run(client, auth_headers, nodes)
    assert result["ok"] is False
    assert "secrets can only be used in nodes that send them to another system" in result["trace"][-1]["detail"]
    rows = (await client.get(f"/masters/{master.json()['id']}/records", headers=auth_headers)).json()
    assert (rows.get("items", rows) if isinstance(rows, dict) else rows) == []


async def test_unknown_secret_names_fail_clearly(client, auth_headers):
    nodes = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "call"}]},
        {"key": "call", "type": "custom_code", "label": "Call", "language": "python", "code": "print(1)",
         "input_params": [{"name": "k", "type": "string", "source_field": "secret.NOPE"}]},
    ]
    _, result = await _run(client, auth_headers, nodes)
    assert result["ok"] is False and "unknown secret NOPE" in result["trace"][-1]["detail"]


async def test_code_editor_run_masks_secrets(client, auth_headers):
    await _secret(client, auth_headers)
    resp = await client.post(
        "/tasks/code/run", headers=auth_headers,
        json={
            "language": "python",
            "code": "import sys, json\nctx = json.load(sys.stdin)\nprint(ctx['k'])\nprint(json.dumps({'k': ctx['k'], 'n': len(ctx['k'])}))",
            "input_params": [{"name": "k", "type": "string", "source_field": "secret.PAY_KEY"}],
        },
    )
    body = resp.json()
    assert body["ok"] is True, body
    assert body["result"] == {"k": MASK, "n": len(VALUE)} and body["inputs"] == {"k": MASK} and body["logs"] == [MASK]
    assert VALUE not in resp.text


async def test_connector_settings_fill_secrets_without_touching_the_stored_config(client, auth_headers, test_tenant):
    await _secret(client, auth_headers)
    async with AsyncSessionLocal() as db:
        async with db.begin():
            connector = Connector(tenant_id=test_tenant["tenant_id"], name="Gateway", type="payment_gateway", status="enabled",
                                  configuration={"provider": "custom", "secret_key": "{{secret.PAY_KEY}}", "other": "{{ secret.MISSING }}"})
            db.add(connector)
            await db.flush()
            resolved = await connector_config(db, test_tenant["tenant_id"], connector)
            assert resolved["secret_key"] == VALUE
            assert resolved["other"] == "{{ secret.MISSING }}"  # unknown left as written
            assert connector.configuration["secret_key"] == "{{secret.PAY_KEY}}"  # stored config untouched
    used = (await client.get("/secrets", headers=auth_headers)).json()[0]["used_by"]
    assert [u["kind"] for u in used] == ["connector"]


def test_reference_and_scrub_helpers():
    assert referenced_names({"url": "https://x/{{secret.A}}", "src": "secret.B_2", "note": "mysecret.C", "x": "a.secret.D"}) == {"A", "B_2"}
    assert fill_templates(["{{secret.A}}-{{ secret.A }}"], {"A": "v"}) == ["v-v"]
    scrub = Scrubber({"A": "abcd1234", "B": "xy"})  # too short to mask
    assert scrub({"t": "token abcd1234 xy", "n": [1, "abcd1234"]}) == {"t": f"token {MASK} xy", "n": [1, MASK]}