619 lines
20 KiB
Python
619 lines
20 KiB
Python
"""Integration tests for SmartTools Registry.
|
|
|
|
These tests verify the CLI and server work together correctly.
|
|
Run with: pytest tests/test_registry_integration.py -v
|
|
|
|
Note: These tests require the registry server to be running locally:
|
|
python -m smarttools.registry.app
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
# Test without requiring server for unit tests
|
|
from smarttools.config import Config, RegistryConfig, load_config, save_config
|
|
from smarttools.manifest import (
|
|
Manifest, Dependency, ToolOverride,
|
|
load_manifest, save_manifest, create_manifest,
|
|
parse_version_constraint
|
|
)
|
|
from smarttools.resolver import (
|
|
ToolSpec, ToolResolver, ResolvedTool,
|
|
resolve_tool, find_tool, list_installed_tools
|
|
)
|
|
from smarttools.registry_client import (
|
|
RegistryClient, RegistryError, PaginatedResponse,
|
|
ToolInfo, DownloadResult
|
|
)
|
|
|
|
|
|
class TestToolSpec:
|
|
"""Tests for ToolSpec parsing."""
|
|
|
|
def test_parse_simple_name(self):
|
|
spec = ToolSpec.parse("summarize")
|
|
assert spec.owner is None
|
|
assert spec.name == "summarize"
|
|
assert spec.version is None
|
|
|
|
def test_parse_qualified_name(self):
|
|
spec = ToolSpec.parse("rob/summarize")
|
|
assert spec.owner == "rob"
|
|
assert spec.name == "summarize"
|
|
assert spec.version is None
|
|
|
|
def test_parse_with_version(self):
|
|
spec = ToolSpec.parse("rob/summarize@1.0.0")
|
|
assert spec.owner == "rob"
|
|
assert spec.name == "summarize"
|
|
assert spec.version == "1.0.0"
|
|
|
|
def test_parse_constraint_version(self):
|
|
spec = ToolSpec.parse("summarize@^1.0.0")
|
|
assert spec.owner is None
|
|
assert spec.name == "summarize"
|
|
assert spec.version == "^1.0.0"
|
|
|
|
def test_full_name_qualified(self):
|
|
spec = ToolSpec.parse("rob/summarize")
|
|
assert spec.full_name == "rob/summarize"
|
|
|
|
def test_full_name_unqualified(self):
|
|
spec = ToolSpec.parse("summarize")
|
|
assert spec.full_name == "summarize"
|
|
|
|
|
|
class TestManifest:
|
|
"""Tests for project manifest handling."""
|
|
|
|
def test_create_manifest(self):
|
|
manifest = Manifest(name="test-project", version="1.0.0")
|
|
assert manifest.name == "test-project"
|
|
assert manifest.version == "1.0.0"
|
|
assert manifest.dependencies == []
|
|
|
|
def test_add_dependency(self):
|
|
manifest = Manifest()
|
|
manifest.add_dependency("rob/summarize", "^1.0.0")
|
|
assert len(manifest.dependencies) == 1
|
|
assert manifest.dependencies[0].name == "rob/summarize"
|
|
assert manifest.dependencies[0].version == "^1.0.0"
|
|
|
|
def test_add_duplicate_dependency_updates(self):
|
|
manifest = Manifest()
|
|
manifest.add_dependency("rob/summarize", "^1.0.0")
|
|
manifest.add_dependency("rob/summarize", "^2.0.0")
|
|
assert len(manifest.dependencies) == 1
|
|
assert manifest.dependencies[0].version == "^2.0.0"
|
|
|
|
def test_get_override(self):
|
|
manifest = Manifest(
|
|
overrides={"rob/summarize": ToolOverride(provider="ollama")}
|
|
)
|
|
override = manifest.get_override("rob/summarize")
|
|
assert override is not None
|
|
assert override.provider == "ollama"
|
|
|
|
def test_get_override_short_name(self):
|
|
manifest = Manifest(
|
|
overrides={"rob/summarize": ToolOverride(provider="ollama")}
|
|
)
|
|
# Should match by short name
|
|
override = manifest.get_override("summarize")
|
|
assert override is not None
|
|
assert override.provider == "ollama"
|
|
|
|
def test_to_dict_roundtrip(self):
|
|
manifest = Manifest(
|
|
name="test",
|
|
version="2.0.0",
|
|
dependencies=[Dependency("rob/summarize", "^1.0.0")],
|
|
overrides={"rob/summarize": ToolOverride(provider="claude")}
|
|
)
|
|
data = manifest.to_dict()
|
|
restored = Manifest.from_dict(data)
|
|
assert restored.name == manifest.name
|
|
assert restored.version == manifest.version
|
|
assert len(restored.dependencies) == 1
|
|
assert restored.dependencies[0].name == "rob/summarize"
|
|
|
|
|
|
class TestVersionConstraint:
|
|
"""Tests for version constraint parsing."""
|
|
|
|
def test_exact_version(self):
|
|
result = parse_version_constraint("1.2.3")
|
|
assert result["operator"] == "="
|
|
assert result["version"] == "1.2.3"
|
|
|
|
def test_any_version(self):
|
|
result = parse_version_constraint("*")
|
|
assert result["operator"] == "*"
|
|
assert result["version"] is None
|
|
|
|
def test_caret_constraint(self):
|
|
result = parse_version_constraint("^1.2.3")
|
|
assert result["operator"] == "^"
|
|
assert result["version"] == "1.2.3"
|
|
|
|
def test_tilde_constraint(self):
|
|
result = parse_version_constraint("~1.2.3")
|
|
assert result["operator"] == "~"
|
|
assert result["version"] == "1.2.3"
|
|
|
|
def test_gte_constraint(self):
|
|
result = parse_version_constraint(">=1.0.0")
|
|
assert result["operator"] == ">="
|
|
assert result["version"] == "1.0.0"
|
|
|
|
|
|
class TestDependency:
|
|
"""Tests for Dependency parsing."""
|
|
|
|
def test_from_dict_object(self):
|
|
dep = Dependency.from_dict({"name": "rob/tool", "version": "^1.0.0"})
|
|
assert dep.name == "rob/tool"
|
|
assert dep.version == "^1.0.0"
|
|
|
|
def test_from_dict_string_simple(self):
|
|
dep = Dependency.from_dict("rob/tool")
|
|
assert dep.name == "rob/tool"
|
|
assert dep.version == "*"
|
|
|
|
def test_from_dict_string_with_version(self):
|
|
dep = Dependency.from_dict("rob/tool@^2.0.0")
|
|
assert dep.name == "rob/tool"
|
|
assert dep.version == "^2.0.0"
|
|
|
|
def test_owner_property(self):
|
|
dep = Dependency(name="rob/summarize")
|
|
assert dep.owner == "rob"
|
|
|
|
def test_tool_name_property(self):
|
|
dep = Dependency(name="rob/summarize")
|
|
assert dep.tool_name == "summarize"
|
|
|
|
|
|
class TestConfig:
|
|
"""Tests for configuration handling."""
|
|
|
|
def test_default_config(self):
|
|
config = Config()
|
|
assert config.registry.url == "https://gitea.brrd.tech/api/v1"
|
|
assert config.auto_fetch_from_registry is True
|
|
assert config.client_id.startswith("anon_")
|
|
|
|
def test_config_to_dict_roundtrip(self):
|
|
config = Config(
|
|
registry=RegistryConfig(token="test_token"),
|
|
auto_fetch_from_registry=False,
|
|
default_provider="claude"
|
|
)
|
|
data = config.to_dict()
|
|
restored = Config.from_dict(data)
|
|
assert restored.registry.token == "test_token"
|
|
assert restored.auto_fetch_from_registry is False
|
|
assert restored.default_provider == "claude"
|
|
|
|
|
|
class TestRegistryClient:
|
|
"""Tests for the registry client (mocked)."""
|
|
|
|
def test_tool_info_from_dict(self):
|
|
data = {
|
|
"owner": "rob",
|
|
"name": "summarize",
|
|
"version": "1.0.0",
|
|
"description": "Test tool",
|
|
"downloads": 100
|
|
}
|
|
info = ToolInfo.from_dict(data)
|
|
assert info.owner == "rob"
|
|
assert info.name == "summarize"
|
|
assert info.full_name == "rob/summarize"
|
|
assert info.downloads == 100
|
|
|
|
def test_paginated_response(self):
|
|
response = PaginatedResponse(
|
|
data=[{"name": "tool1"}, {"name": "tool2"}],
|
|
page=1,
|
|
per_page=20,
|
|
total=2,
|
|
total_pages=1
|
|
)
|
|
assert len(response.data) == 2
|
|
assert response.total == 2
|
|
|
|
|
|
class TestToolResolver:
|
|
"""Tests for tool resolution."""
|
|
|
|
def test_deterministic_owner_order(self, tmp_path):
|
|
"""Test that official namespace is preferred over others."""
|
|
# Create fake tool directories
|
|
tools_dir = tmp_path / ".smarttools"
|
|
|
|
# Create alice/mytool
|
|
alice_tool = tools_dir / "alice" / "mytool"
|
|
alice_tool.mkdir(parents=True)
|
|
(alice_tool / "config.yaml").write_text(yaml.dump({
|
|
"name": "mytool",
|
|
"description": "Alice's version"
|
|
}))
|
|
|
|
# Create official/mytool
|
|
official_tool = tools_dir / "official" / "mytool"
|
|
official_tool.mkdir(parents=True)
|
|
(official_tool / "config.yaml").write_text(yaml.dump({
|
|
"name": "mytool",
|
|
"description": "Official version"
|
|
}))
|
|
|
|
# Create zebra/mytool (should come after official alphabetically)
|
|
zebra_tool = tools_dir / "zebra" / "mytool"
|
|
zebra_tool.mkdir(parents=True)
|
|
(zebra_tool / "config.yaml").write_text(yaml.dump({
|
|
"name": "mytool",
|
|
"description": "Zebra's version"
|
|
}))
|
|
|
|
# Test resolution prefers official
|
|
resolver = ToolResolver(project_dir=tmp_path, auto_fetch=False)
|
|
result = resolver._find_in_local(ToolSpec.parse("mytool"), [])
|
|
|
|
assert result is not None
|
|
assert result.owner == "official"
|
|
assert result.tool.description == "Official version"
|
|
|
|
|
|
# Integration tests (require server)
|
|
@pytest.mark.integration
|
|
class TestRegistryIntegration:
|
|
"""Integration tests requiring a running registry server.
|
|
|
|
Run with: pytest tests/test_registry_integration.py -v -m integration
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
return RegistryClient(base_url="http://localhost:5000/api/v1")
|
|
|
|
def test_list_tools(self, client):
|
|
"""Test listing tools from registry."""
|
|
result = client.list_tools(per_page=5)
|
|
assert isinstance(result, PaginatedResponse)
|
|
# May be empty if no tools seeded
|
|
|
|
def test_search_tools(self, client):
|
|
"""Test searching for tools."""
|
|
result = client.search_tools("test", per_page=5)
|
|
assert isinstance(result, PaginatedResponse)
|
|
|
|
def test_get_categories(self, client):
|
|
"""Test getting categories."""
|
|
categories = client.get_categories()
|
|
assert isinstance(categories, list)
|
|
|
|
def test_get_index(self, client):
|
|
"""Test getting full index."""
|
|
index = client.get_index(force_refresh=True)
|
|
assert "tools" in index
|
|
assert "tool_count" in index
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestAuthIntegration:
|
|
"""Integration tests for authentication endpoints.
|
|
|
|
Run with: pytest tests/test_registry_integration.py -v -m integration
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def base_url(self):
|
|
return "http://localhost:5000/api/v1"
|
|
|
|
@pytest.fixture
|
|
def session(self):
|
|
import requests
|
|
return requests.Session()
|
|
|
|
def test_register_validation(self, session, base_url):
|
|
"""Test registration validation errors."""
|
|
# Missing fields
|
|
resp = session.post(f"{base_url}/register", json={})
|
|
assert resp.status_code == 400
|
|
data = resp.json()
|
|
assert data["error"]["code"] == "VALIDATION_ERROR"
|
|
|
|
# Invalid email
|
|
resp = session.post(f"{base_url}/register", json={
|
|
"email": "invalid",
|
|
"password": "testpass123",
|
|
"slug": "testuser",
|
|
"display_name": "Test"
|
|
})
|
|
assert resp.status_code == 400
|
|
assert "email" in resp.json()["error"]["message"].lower()
|
|
|
|
# Short password
|
|
resp = session.post(f"{base_url}/register", json={
|
|
"email": "test@example.com",
|
|
"password": "short",
|
|
"slug": "testuser",
|
|
"display_name": "Test"
|
|
})
|
|
assert resp.status_code == 400
|
|
assert "password" in resp.json()["error"]["message"].lower()
|
|
|
|
# Invalid slug
|
|
resp = session.post(f"{base_url}/register", json={
|
|
"email": "test@example.com",
|
|
"password": "testpass123",
|
|
"slug": "A", # Too short, wrong case
|
|
"display_name": "Test"
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
def test_login_validation(self, session, base_url):
|
|
"""Test login validation errors."""
|
|
# Missing fields
|
|
resp = session.post(f"{base_url}/login", json={})
|
|
assert resp.status_code == 400
|
|
data = resp.json()
|
|
assert data["error"]["code"] == "VALIDATION_ERROR"
|
|
|
|
# Invalid credentials
|
|
resp = session.post(f"{base_url}/login", json={
|
|
"email": "nonexistent@example.com",
|
|
"password": "wrongpass"
|
|
})
|
|
assert resp.status_code == 401
|
|
assert resp.json()["error"]["code"] == "UNAUTHORIZED"
|
|
|
|
def test_protected_endpoints_require_auth(self, session, base_url):
|
|
"""Test that protected endpoints require authentication."""
|
|
# No auth header
|
|
resp = session.get(f"{base_url}/tokens")
|
|
assert resp.status_code == 401
|
|
assert resp.json()["error"]["code"] == "UNAUTHORIZED"
|
|
|
|
resp = session.get(f"{base_url}/me/tools")
|
|
assert resp.status_code == 401
|
|
|
|
resp = session.post(f"{base_url}/tools", json={})
|
|
assert resp.status_code == 401
|
|
|
|
# Invalid token
|
|
headers = {"Authorization": "Bearer invalid_token"}
|
|
resp = session.get(f"{base_url}/tokens", headers=headers)
|
|
assert resp.status_code == 401
|
|
|
|
def test_full_auth_flow(self, session, base_url):
|
|
"""Test complete registration -> login -> token flow."""
|
|
import uuid
|
|
|
|
# Generate unique test user
|
|
unique = uuid.uuid4().hex[:8]
|
|
email = f"test_{unique}@example.com"
|
|
slug = f"testuser{unique}"
|
|
|
|
# Register
|
|
resp = session.post(f"{base_url}/register", json={
|
|
"email": email,
|
|
"password": "testpass123",
|
|
"slug": slug,
|
|
"display_name": "Test User"
|
|
})
|
|
# May fail if user already exists from previous test run
|
|
if resp.status_code == 201:
|
|
data = resp.json()["data"]
|
|
assert data["slug"] == slug
|
|
assert data["email"] == email
|
|
|
|
# Login
|
|
resp = session.post(f"{base_url}/login", json={
|
|
"email": email,
|
|
"password": "testpass123"
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert "token" in data
|
|
assert data["token"].startswith("reg_")
|
|
token = data["token"]
|
|
|
|
# Use token to access protected endpoint
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = session.get(f"{base_url}/me/tools", headers=headers)
|
|
assert resp.status_code == 200
|
|
assert "data" in resp.json()
|
|
|
|
# List tokens
|
|
resp = session.get(f"{base_url}/tokens", headers=headers)
|
|
assert resp.status_code == 200
|
|
tokens = resp.json()["data"]
|
|
assert len(tokens) >= 1
|
|
|
|
# Create another token
|
|
resp = session.post(f"{base_url}/tokens", headers=headers, json={
|
|
"name": "CLI token"
|
|
})
|
|
assert resp.status_code == 201
|
|
new_token = resp.json()["data"]
|
|
assert new_token["name"] == "CLI token"
|
|
assert "token" in new_token
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestPublishIntegration:
|
|
"""Integration tests for publishing tools.
|
|
|
|
Run with: pytest tests/test_registry_integration.py -v -m integration
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def base_url(self):
|
|
return "http://localhost:5000/api/v1"
|
|
|
|
@pytest.fixture
|
|
def session(self):
|
|
import requests
|
|
return requests.Session()
|
|
|
|
@pytest.fixture
|
|
def auth_headers(self, session, base_url):
|
|
"""Get auth headers for a test user."""
|
|
import uuid
|
|
unique = uuid.uuid4().hex[:8]
|
|
email = f"pub_{unique}@example.com"
|
|
slug = f"publisher{unique}"
|
|
|
|
# Register
|
|
session.post(f"{base_url}/register", json={
|
|
"email": email,
|
|
"password": "testpass123",
|
|
"slug": slug,
|
|
"display_name": "Publisher"
|
|
})
|
|
|
|
# Login
|
|
resp = session.post(f"{base_url}/login", json={
|
|
"email": email,
|
|
"password": "testpass123"
|
|
})
|
|
token = resp.json()["data"]["token"]
|
|
return {"Authorization": f"Bearer {token}"}, slug
|
|
|
|
def test_publish_validation(self, session, base_url, auth_headers):
|
|
"""Test publish validation errors."""
|
|
headers, slug = auth_headers
|
|
|
|
# Empty config
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": "",
|
|
"readme": ""
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
# Invalid YAML
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": "{{invalid yaml",
|
|
"readme": ""
|
|
})
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error"]["code"] == "VALIDATION_ERROR"
|
|
|
|
# Missing required fields
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": "description: no name or version",
|
|
"readme": ""
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
# Invalid version (not semver)
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": "name: test\nversion: bad",
|
|
"readme": ""
|
|
})
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error"]["code"] == "INVALID_VERSION"
|
|
|
|
def test_publish_dry_run(self, session, base_url, auth_headers):
|
|
"""Test publish dry run mode."""
|
|
headers, slug = auth_headers
|
|
import uuid
|
|
tool_name = f"testtool{uuid.uuid4().hex[:8]}"
|
|
|
|
config = f"""name: {tool_name}
|
|
version: 1.0.0
|
|
description: A test tool
|
|
category: text-processing
|
|
"""
|
|
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": config,
|
|
"readme": "# Test Tool",
|
|
"dry_run": True
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert data["status"] == "validated"
|
|
assert data["name"] == tool_name
|
|
assert data["owner"] == slug
|
|
|
|
def test_publish_and_retrieve(self, session, base_url, auth_headers):
|
|
"""Test publishing a tool and retrieving it."""
|
|
headers, slug = auth_headers
|
|
import uuid
|
|
tool_name = f"testtool{uuid.uuid4().hex[:8]}"
|
|
|
|
config = f"""name: {tool_name}
|
|
version: 1.0.0
|
|
description: A test tool for integration testing
|
|
category: text-processing
|
|
tags:
|
|
- test
|
|
- integration
|
|
"""
|
|
|
|
# Publish
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": config,
|
|
"readme": "# Test Tool\n\nThis is a test."
|
|
})
|
|
assert resp.status_code == 201
|
|
data = resp.json()["data"]
|
|
assert data["owner"] == slug
|
|
assert data["name"] == tool_name
|
|
assert data["version"] == "1.0.0"
|
|
|
|
# Retrieve
|
|
resp = session.get(f"{base_url}/tools/{slug}/{tool_name}")
|
|
assert resp.status_code == 200
|
|
tool = resp.json()["data"]
|
|
assert tool["name"] == tool_name
|
|
assert tool["description"] == "A test tool for integration testing"
|
|
assert "test" in tool["tags"]
|
|
|
|
# Check my-tools includes it
|
|
resp = session.get(f"{base_url}/me/tools", headers=headers)
|
|
assert resp.status_code == 200
|
|
my_tools = resp.json()["data"]
|
|
assert any(t["name"] == tool_name for t in my_tools)
|
|
|
|
def test_publish_duplicate_version(self, session, base_url, auth_headers):
|
|
"""Test that publishing duplicate version fails."""
|
|
headers, slug = auth_headers
|
|
import uuid
|
|
tool_name = f"testtool{uuid.uuid4().hex[:8]}"
|
|
|
|
config = f"""name: {tool_name}
|
|
version: 1.0.0
|
|
description: First version
|
|
"""
|
|
|
|
# First publish
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": config,
|
|
"readme": ""
|
|
})
|
|
assert resp.status_code == 201
|
|
|
|
# Duplicate publish
|
|
resp = session.post(f"{base_url}/tools", headers=headers, json={
|
|
"config": config,
|
|
"readme": ""
|
|
})
|
|
assert resp.status_code == 409
|
|
assert resp.json()["error"]["code"] == "VERSION_EXISTS"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|