diff --git a/backend/tests/test_settings_meta_stdio_live.py b/backend/tests/test_settings_meta_stdio_live.py new file mode 100644 index 00000000..4892bcb4 --- /dev/null +++ b/backend/tests/test_settings_meta_stdio_live.py @@ -0,0 +1,115 @@ +"""Definitive proof: a SettingsWrite from the REAL stdio MCP server, against a +REAL running backend, on a REAL session whose model is powered by a specific +API key, refuses to clear THAT key but clears a different provider's key. + +Earlier coverage either drove the endpoint with an injected session (skipping the +stdio subprocess) or drove the stdio subprocess with no session (hitting the +fail-safe). This closes the gap: it boots uvicorn in-process (so the subprocess's +HTTP call lands on a backend whose `agent_manager.sessions` we can populate), +then runs `settings_meta_server.py` exactly as an agent would. No model required, +the guard is provider-routing logic, not an LLM call. +""" + +from __future__ import annotations + +import json +import os +import secrets +import socket +import subprocess +import sys +import threading +import time + +import pytest +import uvicorn + +from backend.main import app + +SERVER = os.path.join(os.path.dirname(os.path.dirname(__file__)), "apps", "agents", "settings_meta_server.py") + + +def _free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +@pytest.fixture +def live_backend(): + import backend.auth as auth_mod + if not auth_mod._TOKEN: + auth_mod._TOKEN = secrets.token_urlsafe(32) + port = _free_port() + server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + for _ in range(200): + if getattr(server, "started", False): + break + time.sleep(0.05) + assert getattr(server, "started", False), "uvicorn did not start" + yield port, auth_mod._TOKEN + server.should_exit = True + thread.join(timeout=5) + + +@pytest.fixture +def reset_settings(): + from backend.apps.settings.settings import load_settings, _save_settings + original = load_settings().model_copy(deep=True) + yield + _save_settings(original) + + +def _run_stdio(port: int, token: str, session_id: str, changes: dict) -> str: + env = { + **os.environ, + "OPENSWARM_PORT": str(port), + "OPENSWARM_AUTH_TOKEN": token, + "OPENSWARM_PARENT_SESSION_ID": session_id, + } + rpc = "\n".join([ + json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}), + json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": {"name": "SettingsWrite", "arguments": {"changes": changes}}}), + ]) + "\n" + proc = subprocess.run([sys.executable, SERVER], input=rpc, capture_output=True, text=True, env=env, timeout=30) + msgs = [json.loads(l) for l in proc.stdout.splitlines() if l.strip()] + resp = next(m for m in msgs if m.get("id") == 2) + return resp["result"]["content"][0]["text"] + + +def test_live_stdio_settingswrite_refuses_live_key_clears_other(live_backend, reset_settings): + port, token = live_backend + from backend.apps.settings.settings import load_settings, _save_settings + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.core.models import AgentSession + + # A real run on opus-4-8 in own_key mode: the Anthropic key powers it; an + # OpenAI key is also connected (the "other provider"). + s = load_settings() + s.connection_mode = "own_key" + s.anthropic_api_key = "sk-ant-LIVE-do-not-clear" + s.openai_api_key = "sk-oai-OTHER-ok-to-clear" + _save_settings(s) + agent_manager.sessions["live-stdio-test"] = AgentSession(id="live-stdio-test", name="t", model="opus-4-8") + + try: + text = _run_stdio(port, token, "live-stdio-test", + {"anthropic_api_key": "", "openai_api_key": "", "theme": "light"}) + finally: + agent_manager.sessions.pop("live-stdio-test", None) + + # The tool's own rendered result, exactly what the agent would read. + assert "Refused anthropic_api_key" in text, text + assert "powering this run" in text + assert "Applied" in text and "theme" in text + + # And the truth on disk: live key kept, other cleared, benign applied. + final = load_settings() + assert final.anthropic_api_key == "sk-ant-LIVE-do-not-clear", "live key was cleared!" + assert not final.openai_api_key, "the other provider's key should have been cleared" + assert final.theme == "light" diff --git a/backend/tests/test_settings_select_and_send.py b/backend/tests/test_settings_select_and_send.py new file mode 100644 index 00000000..01cbaa03 --- /dev/null +++ b/backend/tests/test_settings_select_and_send.py @@ -0,0 +1,53 @@ +"""Proof that selecting a Settings row and sending threads selected_setting_ids +from the HTTP boundary into the run, and produces the focused context block. + +Covers the backend half definitively (HTTP -> send_message kwarg; the context +builder output). The browser half (click a row -> the POST body carries +selected_setting_ids) is proven live via CDP. +""" + +from __future__ import annotations + +import secrets + +import pytest +from fastapi.testclient import TestClient + +from backend.main import app + + +@pytest.fixture +def client(): + import backend.auth as auth_mod + if not auth_mod._TOKEN: + auth_mod._TOKEN = secrets.token_urlsafe(32) + return TestClient(app, headers={"Authorization": f"Bearer {auth_mod._TOKEN}"}) + + +def test_message_endpoint_threads_selected_setting_ids(client, monkeypatch): + from backend.apps.agents.agent_manager import agent_manager + captured: dict = {} + + async def fake_send(session_id, prompt, **kwargs): + captured["session_id"] = session_id + captured["kwargs"] = kwargs + + monkeypatch.setattr(agent_manager, "send_message", fake_send) + r = client.post( + "/api/agents/sessions/sess-x/message", + json={"prompt": "flip my theme to light", "selected_setting_ids": ["theme", "default_model"]}, + ) + assert r.status_code == 200, r.text + assert captured["kwargs"].get("selected_setting_ids") == ["theme", "default_model"] + + +def test_selected_settings_context_block_targets_the_fields(): + from backend.apps.agents.manager.prompt.prompt_context import _build_selected_settings_context + assert _build_selected_settings_context(None) is None + assert _build_selected_settings_context([]) is None + block = _build_selected_settings_context(["theme", "default_model"]) + assert "theme" in block and "default_model" in block + # It tells the agent to use the always-on settings tools on exactly these fields. + assert "SettingsRead" in block and "SettingsWrite" in block + # Targeting aid, not a gate: it focuses, never claims to unlock anything. + assert "Leave unrelated settings alone" in block diff --git a/backend/tests/test_skill_registry_community.py b/backend/tests/test_skill_registry_community.py index c7fa3900..5bf726fe 100644 --- a/backend/tests/test_skill_registry_community.py +++ b/backend/tests/test_skill_registry_community.py @@ -139,6 +139,47 @@ def test_install_dedups_instead_of_clobbering_existing_skill(skills_dir): assert {"pdf", "pdf-2"} <= ids +def test_confirm_install_writes_folder_lists_and_injects(skills_dir, monkeypatch): + """End-to-end install->usable: confirm=true through the real /install endpoint + writes the folder skill, it shows up in /api/skills/list with supporting + files, and _resolve_attached_skills injects it with the folder path so an + agent can read its scripts. (resolve is mocked to skip the network; the live + GitHub resolve is proven separately.)""" + import secrets as _secrets + from fastapi.testclient import TestClient + from backend.main import app + from backend.apps.agents.manager.prompt.prompt_context import _resolve_attached_skills + import backend.auth as auth_mod + if not auth_mod._TOKEN: + auth_mod._TOKEN = _secrets.token_urlsafe(32) + client = TestClient(app, headers={"Authorization": f"Bearer {auth_mod._TOKEN}"}) + + async def fake_resolve(source, skill_id): + return { + "name": "PDF Tools", "description": "work with pdfs", "repo_url": "https://github.com/o/r", + "skill_id": skill_id, + "files": {"SKILL.md": "# PDF Tools\nRun scripts/extract.py to pull text.", + "scripts/extract.py": "print('extract')"}, + "scripts": ["scripts/extract.py"], "secret_findings": [], + } + monkeypatch.setattr("backend.apps.skill_registry.skill_registry.resolve_community_skill", fake_resolve) + + r = client.post("/api/skill-registry/install", json={"source": "o/r", "skill_id": "pdf-tools", "confirm": True}) + assert r.status_code == 200 and r.json()["installed"] is True + slug = r.json()["skill"]["id"] + + # Listed via the real skills API, flagged as multi-file. + listed = {s["id"]: s for s in client.get("/api/skills/list").json()["skills"]} + assert slug in listed and listed[slug]["has_supporting_files"] is True + # On disk as a folder with the script. + assert (skills_dir / slug / "SKILL.md").exists() + assert (skills_dir / slug / "scripts" / "extract.py").exists() + # Injectable: the agent gets the body AND a pointer to the folder for on-demand reads. + block = _resolve_attached_skills([{"id": slug, "name": "PDF Tools", "content": "# PDF Tools\nRun scripts/extract.py to pull text."}]) + assert "[Using skill: PDF Tools]" in block + assert str(skills_dir / slug) in block + + def test_write_folder_skill_blocks_path_traversal(skills_dir): skills_mod.write_folder_skill( "evil",