flowdeck/tests/test_v758_plugins.py

"""v7.58.0 — Add plugins : registre persistant + on/off à effet réel."""

import asyncio
from pathlib import Path

from conftest import anon_csrf

from app.services import plugins
from app.services.tool_registry import ToolRegistry

ROOT = Path(__file__).resolve().parent.parent
JS_PATH = ROOT / "static" / "js" / "agent_panel_2.js"
PANEL_HTML = ROOT / "app" / "templates" / "agent_panel.html"
SETTINGS_HTML = ROOT / "app" / "templates" / "settings.html"
AUTOMATIONS_PY = ROOT / "app" / "services" / "automations.py"


def _set(client, slug: str, enabled: bool) -> dict:
    r = client.patch(f"/api/agent/plugins/{slug}", json={"enabled": enabled})
    assert r.status_code == 200, r.text
    return r.json()


# ── Catalogue ───────────────────────────────────────────────────────────────

def test_catalog_lists_the_three_wired_plugins(client):
    data = client.get("/api/agent/plugins").json()["plugins"]
    assert [p["slug"] for p in data] == ["web-tools", "web-clipper", "automations"]
    assert all(p["enabled"] for p in data)
    assert all(p["name"] and p["description"] for p in data)


def test_unknown_slug_is_404(client):
    assert client.patch("/api/agent/plugins/nope",
                        json={"enabled": False}).status_code == 404


def test_toggle_persists_in_db(client):
    _set(client, "web-clipper", False)
    assert plugins.is_enabled("web-clipper") is False      # relu depuis la base
    _set(client, "web-clipper", True)
    assert plugins.is_enabled("web-clipper") is True


# ── Effet réel : routes refusées ────────────────────────────────────────────

def test_automations_routes_refused_when_plugin_off(client):
    # 404 hors /api = redirection vers /workspaces (handler unifié de main.py)
    def get():
        return client.get("/workspace/automations", follow_redirects=False)

    assert get().status_code == 200
    _set(client, "automations", False)
    assert get().status_code == 302
    assert get().headers["location"] == "/workspaces"
    assert client.post("/workspace/automations", json={},
                       follow_redirects=False).status_code == 302
    assert plugins.is_enabled("automations") is False
    _set(client, "automations", True)
    assert get().status_code == 200


def test_scheduler_tick_guarded_when_automations_off(client):
    # le scheduler doit vérifier le plugin avant chaque tick ( effet réel, pas un
    # drapeau lu nulle part )
    src = AUTOMATIONS_PY.read_text(encoding="utf-8")
    assert 'plugins.is_enabled("automations")' in src
    _set(client, "automations", False)
    assert plugins.is_enabled("automations") is False


def test_web_clipper_api_and_page_refused_when_plugin_off(client):
    def page():
        return client.get("/extensions", follow_redirects=False)

    assert client.get("/api/v2/web-clipper/status").status_code == 200
    assert page().status_code == 200
    _set(client, "web-clipper", False)
    assert client.get("/api/v2/web-clipper/status").status_code == 404
    assert client.post("/api/v2/web-clipper/clip", json={}).status_code == 404
    assert page().status_code == 302 and page().headers["location"] == "/workspaces"
    _set(client, "web-clipper", True)
    assert client.get("/api/v2/web-clipper/status").status_code == 200
    assert page().status_code == 200


# ── Effet réel : outils de l'agent ──────────────────────────────────────────

def test_web_tools_leaves_the_llm_registry_when_plugin_off(client):
    names = {t["name"] for t in ToolRegistry().schema()}
    assert {"web_search", "fetch_url"} <= names

    _set(client, "web-tools", False)
    names = {t["name"] for t in ToolRegistry().schema()}
    assert not ({"web_search", "fetch_url"} & names)
    res = asyncio.run(ToolRegistry().execute("fetch_url",
                                              {"url": "https://example.com"}))
    assert res.status == "error" and "inconnu" in res.message.lower()

    _set(client, "web-tools", True)
    names = {t["name"] for t in ToolRegistry().schema()}
    assert {"web_search", "fetch_url"} <= names


# ── UI : settings rendu côté serveur + menu ＋ ──────────────────────────────

def test_settings_hides_disabled_plugins(client):
    html = client.get("/settings").text
    assert "navTo('extensions')" in html and "navTo('automations')" in html   # ON

    _set(client, "automations", False)
    _set(client, "web-clipper", False)
    html = client.get("/settings").text
    assert "navTo('automations')" not in html                            # nav retirée
    assert "navTo('extensions')" not in html
    assert "x-show=\"false &amp;&amp; activeSection==='automations'\"" in html

    _set(client, "automations", True)
    _set(client, "web-clipper", True)
    html = client.get("/settings").text
    assert "navTo('automations')" in html and "navTo('extensions')" in html


def test_menu_catalogue_is_live(client):
    src = JS_PATH.read_text(encoding="utf-8")
    for token in ("action:'plugins'", "_pluginItems()", "pluginToggle(slug)",
                  "/api/agent/plugins", "Add plugins"):
        assert token in src, token
    # la section n'est plus une coquille « bientôt »
    root = src.split("var FD_PLUS_MENU = [", 1)[1].split("];", 1)[0]
    assert "disabled:true" not in root
    panel = PANEL_HTML.read_text(encoding="utf-8")
    assert "plusSection!=='root'" in panel              # en-tête/rétour génériques
    settings = SETTINGS_HTML.read_text(encoding="utf-8")
    assert settings.count("plugin_enabled(") >= 4


def test_plugins_routes_require_session(client):
    anon_csrf(client)
    assert client.get("/api/agent/plugins").status_code == 401
    assert client.patch("/api/agent/plugins/automations",
                        json={"enabled": False}).status_code == 401