flowdeck/tests/test_v755_connectors.py

"""v7.55.0 — Connecteurs de l'agent : catalogue, CRUD, SSRF, statut, outil LLM."""

import asyncio
from pathlib import Path

import pytest
from conftest import anon_csrf

from app.db import get_conn
from app.services import connectors
from app.services.sso_provisioning import decrypt_secret

ROOT = Path(__file__).resolve().parent.parent
JS_PATH = ROOT / "static" / "js" / "agent_panel_2.js"
PANEL_HTML = ROOT / "app" / "templates" / "agent_panel.html"

PUBLIC_URL = "https://example.com/api"


def _create(client, **kw):
    body = {"name": "Conn perso", "url": PUBLIC_URL, "secret": "sk-live-abc", **kw}
    r = client.post("/api/agent/connectors", json=body)
    assert r.status_code == 200, r.text
    return r.json()


def test_native_connectors_always_listed(client):
    rows = client.get("/api/agent/connectors").json()["connectors"]
    native = {r["kind"]: r for r in rows if r["builtin"]}
    assert set(native) == {"gitea", "github", "web"}
    for r in native.values():
        assert r["id"] is None
        assert r["status"] in ("ok", "missing")
        assert r["enabled"] is True
    assert native["web"]["status"] == "ok"


def test_create_custom_connector_never_returns_secret(client):
    row = _create(client)
    assert row["kind"] == "custom" and row["builtin"] is False
    assert row["has_secret"] is True
    assert "sk-live-abc" not in str(row)          # jamais en clair dans la réponse
    with get_conn() as conn:
        stored = conn.execute(
            "SELECT secret_encrypted FROM agent_connectors WHERE id=?", (row["id"],)
        ).fetchone()["secret_encrypted"]
    assert stored and "sk-live-abc" not in stored  # chiffré (Fernet)
    assert decrypt_secret(stored) == "sk-live-abc"


@pytest.mark.parametrize("bad", [
    "http://localhost/secret",
    "http://127.0.0.1:8080/admin",
    "http://169.254.169.254/latest/meta-data/",
    "ftp://exemple.com/api",
    "file:///etc/passwd",
])
def test_create_rejects_non_public_urls(client, bad):
    r = client.post("/api/agent/connectors", json={"name": "interne", "url": bad})
    assert r.status_code == 400, r.text


def test_patch_toggle_and_delete(client):
    row = _create(client)
    cid = row["id"]
    off = client.patch(f"/api/agent/connectors/{cid}", json={"enabled": False})
    assert off.status_code == 200 and off.json()["enabled"] is False

    listed = {r["id"]: r for r in client.get("/api/agent/connectors").json()["connectors"]}
    assert listed[cid]["enabled"] is False

    assert client.delete(f"/api/agent/connectors/{cid}").status_code == 200
    assert client.delete(f"/api/agent/connectors/{cid}").status_code == 404
    assert client.patch(f"/api/agent/connectors/{cid}", json={"enabled": True}).status_code == 404


def test_connectors_require_session(client):
    anon_csrf(client)
    assert client.get("/api/agent/connectors").status_code == 401
    assert client.post("/api/agent/connectors",
                       json={"name": "x", "url": PUBLIC_URL}).status_code == 401


def test_probe_persists_status(client, monkeypatch):
    row = _create(client)
    cid = row["id"]

    async def ok_get(url, headers=None):
        assert url.startswith("https://example.com")       # URL du connecteur
        assert (headers or {}).get("Authorization") == "Bearer sk-live-abc"  # clé déchiffrée
        return 200, "{}"

    monkeypatch.setattr(connectors, "_get", ok_get)
    res = _probe(client, str(cid))
    assert res["status"] == "ok", res
    with get_conn() as conn:
        st = conn.execute("SELECT status FROM agent_connectors WHERE id=?", (cid,)).fetchone()
    assert st["status"] == "ok"


def _probe(client, target):
    r = client.post("/api/agent/connectors/probe", json={"connector": target})
    assert r.status_code == 200, r.text
    return r.json()


def test_probe_error_is_persisted(client, monkeypatch):
    row = _create(client)
    cid = row["id"]

    async def boom(url, headers=None):
        raise ValueError("Hôte non autorisé")

    monkeypatch.setattr(connectors, "_get", boom)
    res = _probe(client, str(cid))
    assert res["status"] == "error"
    assert "Hôte non autorisé" in res["detail"]
    with get_conn() as conn:
        st = conn.execute("SELECT status, detail FROM agent_connectors WHERE id=?",
                          (cid,)).fetchone()
    assert st["status"] == "error"


def test_connector_fetch_tool_registered_and_works(client, monkeypatch):
    from app.services.tool_registry import ToolRegistry

    reg = ToolRegistry()
    schema = {t["name"]: t for t in reg.schema()}
    assert "connector_fetch" in schema
    assert schema["connector_fetch"]["parameters"]["required"] == ["connector"]

    row = _create(client)

    async def ok_get(url, headers=None):
        return 200, '{"ok": true, "source": "exemple"}'

    monkeypatch.setattr(connectors, "_get", ok_get)
    res = asyncio.run(reg.execute("connector_fetch", {"connector": str(row["id"]),
                                                      "path": "/status"}))
    assert res.status == "success"
    assert "exemple" in res.data["text"]

    # connecteur désactivé → refusé
    client.patch(f"/api/agent/connectors/{row['id']}", json={"enabled": False})
    res2 = asyncio.run(reg.execute("connector_fetch", {"connector": str(row["id"])}))
    assert res2.status == "error"
    assert "désactivé" in res2.message.lower()

    # connecteur inconnu → erreur outil (pas d'exception qui casse le run)
    res3 = asyncio.run(reg.execute("connector_fetch", {"connector": "999999"}))
    assert res3.status == "error"


def test_connectors_menu_wired(client):
    src = JS_PATH.read_text(encoding="utf-8")
    root = src.split("var FD_PLUS_MENU = [", 1)[1].split("];", 1)[0]
    assert "action:'connectors'" in root
    # phases 1-5 livrées → seule la section Plugins reste « bientôt »
    assert root.count("disabled:true") == 1
    for action in ("'connector'", "'connector-new'", "'connector-probe'",
                   "'connector-toggle'", "'connector-delete'"):
        assert action in src, action
    for fn in ("fetchConnectors()", "connectorCreate()", "connectorProbe()",
               "connectorToggle()", "connectorDelete()"):
        assert fn in src, fn
    html = PANEL_HTML.read_text(encoding="utf-8")
    assert "plusSection==='connector-new'" in html
    assert 'type="password"' in html          # la clé n'est pas en clair à l'écran
    resp = client.get("/accounts")
    assert resp.status_code == 200 and "connectorForm" in resp.text