"""Tests for the workflow MCP tool: search, install, check, update.""" import hashlib import io import json import os import subprocess import sys import tarfile import threading from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path import pytest import yaml from gcontext import registry as registry_mod, server, fs INDEX_MD = """--- id: demo-flow name: Demo Flow description: A tiny demo workflow for tests. tags: [demo] --- Objective paragraph. """ SETUP_MD = """--- description: Set up the demo workflow --- Interview the user. """ BUNDLE_FILES = [ {"path": "index.md", "content": INDEX_MD}, {"path": "steps/index.md", "content": "1-sync.md: sync things\n"}, {"path": "steps/1-sync.md", "content": "# Step 1\n\nSync.\n"}, {"path": "commands/setup.md", "content": SETUP_MD}, {"path": "runs/example/index.md", "content": "# Example run\n"}, ] CATALOG = { "generated": "2026-08-10T12:00:00Z", "workflows": [ { "id": "demo-flow", "name": "Demo Flow", "description": "A tiny demo workflow for tests.", "tags": ["demo"], "files": [f["path"] for f in BUNDLE_FILES], }, { "id": "ops-flow", "name": "Ops Flow", "description": "An operations workflow.", "tags": ["ops", "infra"], "files": ["index.md"], }, ], } def _build_tarball(files, prefix="workflows-main"): buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tf: for f in files: member_path = f"{prefix}/{f['path']}" data = f["content"].encode("utf-8") info = tarfile.TarInfo(name=member_path) info.size = len(data) tf.addfile(info, io.BytesIO(data)) buf.seek(0) return buf.read() def _registry_files(workflow_id="demo-flow"): return [{"path": f"{workflow_id}/{f['path']}", "content": f["content"]} for f in BUNDLE_FILES] def _file_hash(content): return "sha256:" + hashlib.sha256(content.encode("utf-8")).hexdigest() @pytest.fixture def registry(monkeypatch): tarball_data = [None] class Handler(BaseHTTPRequestHandler): def do_GET(self): if tarball_data[0] is not None: self.send_response(200) self.send_header("Content-Type", "application/gzip") self.end_headers() self.wfile.write(tarball_data[0]) else: self.send_response(404) self.end_headers() def log_message(self, *args): pass srv = HTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=srv.serve_forever, daemon=True) thread.start() url = f"http://127.0.0.1:{srv.server_port}/registry.tar.gz" monkeypatch.setenv("GCONTEXT_REGISTRY", url) yield tarball_data srv.shutdown() @pytest.fixture def project(tmp_path, monkeypatch): p = tmp_path / "agent" p.mkdir() (p / "gcontext.yaml").write_text("name: test-agent\n") (p / "modules").mkdir() (p / "connections").mkdir() monkeypatch.setattr(server, "PROJECT_DIR", p) return p def _tarball_with_catalog(extra_files=None): files = _registry_files() catalog_content = json.dumps(CATALOG) files.append({"path": "registry.json", "content": catalog_content}) if extra_files: files.extend(extra_files) return _build_tarball(files) # --- Search tests --- def test_search_returns_all(registry, project): registry[0] = _tarball_with_catalog() result = server.workflow(action="search") assert "demo-flow" in result assert "ops-flow" in result def test_search_filters_by_query(registry, project): registry[0] = _tarball_with_catalog() result = server.workflow(action="search", query="demo") assert "demo-flow" in result assert "ops-flow" not in result def test_search_case_insensitive(registry, project): registry[0] = _tarball_with_catalog() result = server.workflow(action="search", query="DEMO") assert "demo-flow" in result def test_search_no_match(registry, project): registry[0] = _tarball_with_catalog() result = server.workflow(action="search", query="nomatch") assert "No workflows match" in result def test_search_without_catalog_errors(registry, project): registry[0] = _build_tarball(_registry_files()) result = server.workflow(action="search") assert result.startswith("Error:") assert "registry.json" in result # --- Install tests --- def test_install_creates_module_and_manifest(registry, project): registry[0] = _tarball_with_catalog() result = server.workflow(action="install", id="demo-flow") assert "Demo Flow" in result assert "commands/setup.md" in result module = project / "modules" / "demo-flow" assert (module / "index.md").exists() assert (module / "steps" / "1-sync.md").exists() manifest = yaml.safe_load((module / ".template.yaml").read_text()) assert manifest["template"] == "demo-flow" assert manifest["installed_ref"] == "unknown" for f in BUNDLE_FILES: assert manifest["files"][f["path"]] == _file_hash(f["content"]) def test_install_existing_module_refuses(registry, project): marker = project / "modules" / "demo-flow" / "personal.md" marker.parent.mkdir(parents=True) marker.write_text("mine") registry[0] = _tarball_with_catalog() result = server.workflow(action="install", id="demo-flow") assert result.startswith("Error:") assert "already exists" in result assert marker.read_text() == "mine" def test_install_missing_id(registry, project): result = server.workflow(action="install") assert result.startswith("Error:") assert "needs an id" in result def test_unknown_action(registry, project): result = server.workflow(action="frobnicate") assert result.startswith("Error:") assert "unknown action" in result # --- Hidden manifest tests --- def test_template_manifest_hidden_from_list_dir(project): mod = project / "modules" / "demo" mod.mkdir(parents=True) (mod / ".template.yaml").write_text("template: demo\n") (mod / "index.md").write_text("# demo\n") result = fs.list_dir(project, "modules/demo") assert "index.md" in result assert ".template.yaml" not in result def test_template_manifest_hidden_from_grep(project): mod = project / "modules" / "demo" mod.mkdir(parents=True) (mod / ".template.yaml").write_text("template: demo\n") (mod / "index.md").write_text("# demo\n") result = fs.grep(project, "template", "modules") assert ".template.yaml" not in result def test_template_manifest_hidden_from_walk(project): mod = project / "modules" / "demo" mod.mkdir(parents=True) (mod / ".template.yaml").write_text("template: demo\n") (mod / "index.md").write_text("# demo\n") walked = fs.walk_files(project) assert not any(".template.yaml" in p for p in walked) def test_template_manifest_still_readable(project): mod = project / "modules" / "demo" mod.mkdir(parents=True) (mod / ".template.yaml").write_text("template: demo\n") result = fs.read_file(project, "modules/demo/.template.yaml") assert "template: demo" in result def test_index_warning_ignores_template_manifest(project): mod = project / "modules" / "demo" mod.mkdir(parents=True) (mod / ".template.yaml").write_text("template: demo\n") (mod / "steps").mkdir() result = fs.write_file(project, "modules/demo/index.md", "# demo\n\n- [steps](steps/)\n") assert ".template.yaml" not in result # --- Check tests --- def test_check_up_to_date(registry, project): registry[0] = _tarball_with_catalog() server.workflow(action="install", id="demo-flow") result = server.workflow(action="check", id="demo-flow") assert "up to date" in result def test_check_detects_changes(registry, project): registry[0] = _tarball_with_catalog() server.workflow(action="install", id="demo-flow") modified_step = "# Step 1 MODIFIED\n\nNew sync.\n" modified_files = [] for f in BUNDLE_FILES: if f["path"] == "steps/1-sync.md": modified_files.append({"path": f["path"], "content": modified_step}) else: modified_files.append(f) new_registry = _registry_files_from(modified_files) new_registry.append({"path": "registry.json", "content": json.dumps(CATALOG)}) registry[0] = _build_tarball(new_registry) (project / "modules" / "demo-flow" / "commands" / "setup.md").write_text("local edit") result = server.workflow(action="check", id="demo-flow") assert "upstream changed" in result assert "locally modified" in result def test_check_nonexistent_module(registry, project): result = server.workflow(action="check", id="nope") assert result.startswith("Error:") assert "does not exist" in result def test_check_all_no_tracked(registry, project): result = server.workflow(action="check") assert "No installed workflows" in result # --- Update tests --- def test_update_applies_three_way(registry, project): registry[0] = _tarball_with_catalog() server.workflow(action="install", id="demo-flow") (project / "modules" / "demo-flow" / "commands" / "setup.md").write_text("local edit") new_step = "# Step 1 UPDATED\n" new_index = INDEX_MD.replace("Objective paragraph.", "Updated objective.") modified_files = [] for f in BUNDLE_FILES: if f["path"] == "steps/1-sync.md": modified_files.append({"path": f["path"], "content": new_step}) elif f["path"] == "index.md": modified_files.append({"path": f["path"], "content": new_index}) else: modified_files.append(f) modified_files.append({"path": "steps/2-verify.md", "content": "# Verify\n"}) new_registry = _registry_files_from(modified_files) new_registry.append({"path": "registry.json", "content": json.dumps(CATALOG)}) registry[0] = _build_tarball(new_registry) (project / "modules" / "demo-flow" / "index.md").write_text( INDEX_MD.replace("Objective paragraph.", "My local objective.") ) result = server.workflow(action="update", id="demo-flow") assert (project / "modules" / "demo-flow" / "steps" / "1-sync.md").read_text() == new_step assert (project / "modules" / "demo-flow" / "commands" / "setup.md").read_text() == "local edit" assert (project / "modules" / "demo-flow" / "index.md.new").exists() assert (project / "modules" / "demo-flow" / "steps" / "2-verify.md").exists() assert "Conflicts" in result or "conflicts" in result.lower() def test_update_without_manifest_errors(registry, project): mod = project / "modules" / "handmade" mod.mkdir(parents=True) (mod / "index.md").write_text("# handmade\n") result = server.workflow(action="update", id="handmade") assert result.startswith("Error:") assert ".template.yaml" in result def test_update_missing_id(registry, project): result = server.workflow(action="update") assert result.startswith("Error:") assert "needs an id" in result # --- CLI wrapper tests --- def _run_cli(*args, cwd, env=None): return subprocess.run( [sys.executable, "-m", "gcontext.cli", *args], capture_output=True, text=True, cwd=cwd, env=env, ) def test_cli_search(registry, tmp_path): registry[0] = _tarball_with_catalog() agent = tmp_path / "a" _run_cli("init", "a", cwd=tmp_path) result = _run_cli("search", "demo", cwd=agent) assert result.returncode == 0 assert "demo-flow" in result.stdout def test_cli_update_up_to_date(registry, tmp_path): registry[0] = _tarball_with_catalog() agent = tmp_path / "a" _run_cli("init", "a", cwd=tmp_path) _run_cli("add", "demo-flow", cwd=agent) result = _run_cli("update", "demo-flow", cwd=agent) assert result.returncode == 0 assert "up to date" in result.stdout def test_cli_update_unknown_module(registry, tmp_path): registry[0] = _tarball_with_catalog() agent = tmp_path / "a" _run_cli("init", "a", cwd=tmp_path) result = _run_cli("update", "nope", cwd=agent) assert result.returncode == 1 assert "Error:" in result.stderr # --- Helpers --- def _registry_files_from(bundle_files, workflow_id="demo-flow"): return [{"path": f"{workflow_id}/{f['path']}", "content": f["content"]} for f in bundle_files]