366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""Tests for collection-related API endpoints.
|
|
|
|
Tests for:
|
|
- GET /api/v1/me
|
|
- GET /api/v1/tools/<owner>/<name>/approved
|
|
- POST /api/v1/collections
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
# Check if Flask is available for API tests
|
|
try:
|
|
import flask
|
|
HAS_FLASK = True
|
|
except ImportError:
|
|
HAS_FLASK = False
|
|
|
|
flask_required = pytest.mark.skipif(not HAS_FLASK, reason="Flask not installed")
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
"""Create Flask test app with in-memory database."""
|
|
pytest.importorskip("flask", reason="Flask not installed")
|
|
from cmdforge.registry.app import create_app
|
|
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
|
|
with patch.dict(os.environ, {"CMDFORGE_REGISTRY_DB": db_path}):
|
|
app = create_app()
|
|
app.config["TESTING"] = True
|
|
|
|
yield app
|
|
|
|
# Cleanup
|
|
Path(db_path).unlink(missing_ok=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Create test client."""
|
|
return app.test_client()
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_headers(app):
|
|
"""Create auth headers with a valid token."""
|
|
import hashlib
|
|
from datetime import datetime
|
|
from cmdforge.registry.db import connect_db
|
|
|
|
token = "test-token"
|
|
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
conn = connect_db()
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT id FROM publishers WHERE slug = ?",
|
|
["testuser"],
|
|
).fetchone()
|
|
if row:
|
|
publisher_id = row["id"]
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO publishers (email, password_hash, slug, display_name, role)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
["test@example.com", "x", "testuser", "Test User", "user"],
|
|
)
|
|
publisher_id = conn.execute(
|
|
"SELECT id FROM publishers WHERE slug = ?",
|
|
["testuser"],
|
|
).fetchone()["id"]
|
|
|
|
# Insert token if missing
|
|
token_row = conn.execute(
|
|
"SELECT id FROM api_tokens WHERE token_hash = ?",
|
|
[token_hash],
|
|
).fetchone()
|
|
if not token_row:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO api_tokens (publisher_id, token_hash, name, created_at)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
[publisher_id, token_hash, "test-token", datetime.utcnow().isoformat()],
|
|
)
|
|
|
|
# Insert a public approved tool for tests if missing
|
|
tool_row = conn.execute(
|
|
"SELECT id FROM tools WHERE owner = ? AND name = ?",
|
|
["testuser", "tool1"],
|
|
).fetchone()
|
|
if not tool_row:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO tools (owner, name, version, description, category, tags,
|
|
config_yaml, readme, publisher_id, visibility, moderation_status, published_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
[
|
|
"testuser", "tool1", "1.0.0", "Test tool", "Other", "[]",
|
|
"name: tool1\nversion: 1.0.0\n", "", publisher_id,
|
|
"public", "approved", datetime.utcnow().isoformat(),
|
|
],
|
|
)
|
|
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@flask_required
|
|
class TestGetMeEndpoint:
|
|
"""Tests for GET /api/v1/me endpoint."""
|
|
|
|
def test_requires_auth(self, client):
|
|
response = client.get('/api/v1/me')
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_user_info(self, client, auth_headers):
|
|
response = client.get('/api/v1/me', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data["data"]["slug"] == "testuser"
|
|
|
|
|
|
@flask_required
|
|
class TestToolApprovedEndpoint:
|
|
"""Tests for GET /api/v1/tools/<owner>/<name>/approved endpoint."""
|
|
|
|
def test_invalid_owner_format(self, client):
|
|
response = client.get('/api/v1/tools/invalid@owner/tool/approved')
|
|
assert response.status_code == 400
|
|
|
|
def test_invalid_name_format(self, client):
|
|
response = client.get('/api/v1/tools/valid/invalid@name/approved')
|
|
assert response.status_code == 400
|
|
|
|
def test_tool_not_found(self, client):
|
|
response = client.get('/api/v1/tools/nonexistent/tool/approved')
|
|
data = json.loads(response.data)
|
|
# Should return success but with has_approved_public_version = False
|
|
assert response.status_code == 200
|
|
assert data['data']['has_approved_public_version'] is False
|
|
|
|
|
|
@flask_required
|
|
class TestPostCollectionsEndpoint:
|
|
"""Tests for POST /api/v1/collections endpoint."""
|
|
|
|
def test_requires_auth(self, client):
|
|
response = client.post('/api/v1/collections', json={
|
|
"name": "test-coll",
|
|
"display_name": "Test Collection",
|
|
"tools": ["official/tool1"]
|
|
})
|
|
assert response.status_code == 401
|
|
|
|
def test_invalid_name_format(self, client, auth_headers):
|
|
response = client.post('/api/v1/collections', headers=auth_headers, json={
|
|
"name": "Invalid Name",
|
|
"display_name": "Test Collection",
|
|
"tools": ["testuser/tool1"]
|
|
})
|
|
assert response.status_code == 400
|
|
|
|
def test_missing_display_name(self, client, auth_headers):
|
|
response = client.post('/api/v1/collections', headers=auth_headers, json={
|
|
"name": "test-coll",
|
|
"tools": ["testuser/tool1"]
|
|
})
|
|
assert response.status_code == 400
|
|
|
|
def test_missing_tools(self, client, auth_headers):
|
|
response = client.post('/api/v1/collections', headers=auth_headers, json={
|
|
"name": "test-coll",
|
|
"display_name": "Test Collection",
|
|
})
|
|
assert response.status_code == 400
|
|
|
|
def test_tools_must_be_list(self, client, auth_headers):
|
|
response = client.post('/api/v1/collections', headers=auth_headers, json={
|
|
"name": "test-coll",
|
|
"display_name": "Test Collection",
|
|
"tools": "not-a-list"
|
|
})
|
|
assert response.status_code == 400
|
|
|
|
def test_pinned_must_be_dict(self, client, auth_headers):
|
|
response = client.post('/api/v1/collections', headers=auth_headers, json={
|
|
"name": "test-coll",
|
|
"display_name": "Test Collection",
|
|
"tools": ["testuser/tool1"],
|
|
"pinned": ["not", "a", "dict"]
|
|
})
|
|
assert response.status_code == 400
|
|
|
|
def test_tags_must_be_list(self, client, auth_headers):
|
|
response = client.post('/api/v1/collections', headers=auth_headers, json={
|
|
"name": "test-coll",
|
|
"display_name": "Test Collection",
|
|
"tools": ["testuser/tool1"],
|
|
"tags": "not-a-list"
|
|
})
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestRegistryClientMethods:
|
|
"""Tests for new RegistryClient methods."""
|
|
|
|
@pytest.fixture
|
|
def mock_session(self):
|
|
"""Create a mock requests session."""
|
|
session = MagicMock()
|
|
return session
|
|
|
|
def test_get_me(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient
|
|
|
|
client = RegistryClient(token="test-token")
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"data": {"id": 1, "slug": "testuser", "role": "user"}
|
|
}
|
|
mock_session.request.return_value = mock_response
|
|
|
|
result = client.get_me()
|
|
assert result["slug"] == "testuser"
|
|
assert result["role"] == "user"
|
|
|
|
def test_get_me_unauthorized(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient, RegistryError
|
|
|
|
client = RegistryClient(token="bad-token")
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 401
|
|
mock_response.json.return_value = {
|
|
"error": {"code": "UNAUTHORIZED", "message": "Invalid token"}
|
|
}
|
|
mock_session.request.return_value = mock_response
|
|
|
|
with pytest.raises(RegistryError) as exc_info:
|
|
client.get_me()
|
|
assert exc_info.value.code == "UNAUTHORIZED"
|
|
|
|
def test_has_approved_public_tool_true(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient
|
|
|
|
client = RegistryClient()
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"data": {"has_approved_public_version": True}
|
|
}
|
|
mock_session.request.return_value = mock_response
|
|
|
|
result = client.has_approved_public_tool("official", "summarize")
|
|
assert result is True
|
|
|
|
def test_has_approved_public_tool_false(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient
|
|
|
|
client = RegistryClient()
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"data": {"has_approved_public_version": False}
|
|
}
|
|
mock_session.request.return_value = mock_response
|
|
|
|
result = client.has_approved_public_tool("official", "pending-tool")
|
|
assert result is False
|
|
|
|
def test_has_approved_public_tool_not_found(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient, RegistryError
|
|
|
|
client = RegistryClient()
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
mock_session.request.return_value = mock_response
|
|
|
|
with pytest.raises(RegistryError) as exc_info:
|
|
client.has_approved_public_tool("nonexistent", "tool")
|
|
assert exc_info.value.code == "TOOL_NOT_FOUND"
|
|
|
|
def test_publish_collection_success(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient
|
|
|
|
client = RegistryClient(token="test-token")
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_response.json.return_value = {
|
|
"data": {"name": "my-coll", "display_name": "My Collection", "created": True}
|
|
}
|
|
mock_session.request.return_value = mock_response
|
|
|
|
result = client.publish_collection({
|
|
"name": "my-coll",
|
|
"display_name": "My Collection",
|
|
"tools": ["official/tool1"]
|
|
})
|
|
assert result["name"] == "my-coll"
|
|
assert result["created"] is True
|
|
|
|
def test_publish_collection_conflict(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient, RegistryError
|
|
|
|
client = RegistryClient(token="test-token")
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 409
|
|
mock_session.request.return_value = mock_response
|
|
|
|
with pytest.raises(RegistryError) as exc_info:
|
|
client.publish_collection({
|
|
"name": "existing-coll",
|
|
"display_name": "Existing",
|
|
"tools": ["official/tool1"]
|
|
})
|
|
assert exc_info.value.code == "COLLECTION_EXISTS"
|
|
|
|
def test_publish_collection_update(self, mock_session):
|
|
from cmdforge.registry_client import RegistryClient
|
|
|
|
client = RegistryClient(token="test-token")
|
|
client._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"data": {"name": "my-coll", "display_name": "My Collection Updated", "updated": True}
|
|
}
|
|
mock_session.request.return_value = mock_response
|
|
|
|
result = client.publish_collection({
|
|
"name": "my-coll",
|
|
"display_name": "My Collection Updated",
|
|
"tools": ["official/tool1", "official/tool2"]
|
|
})
|
|
assert result["updated"] is True
|