690 lines
24 KiB
Python
690 lines
24 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
|
|
|
|
# Keep the environment override active for the entire test. Registry
|
|
# request handlers open fresh connections rather than reusing the one
|
|
# created during app initialization.
|
|
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, timezone
|
|
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.now(timezone.utc).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.now(timezone.utc).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 TestDeprecationChains:
|
|
def test_replacement_chain_is_validated_and_returned(self, client, auth_headers):
|
|
target = client.post(
|
|
"/api/v1/tools", headers=auth_headers,
|
|
json={"config": "name: replacement\nversion: 1.0.0\noutput: ok\n"},
|
|
)
|
|
assert target.status_code == 201
|
|
response = client.post(
|
|
"/api/v1/tools/testuser/tool1/deprecate",
|
|
headers=auth_headers,
|
|
json={"deprecated_message": "Moved", "replacement": "replacement"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.get_json()["data"]["replacement_chain"] == [
|
|
"testuser/replacement"
|
|
]
|
|
|
|
def test_replacement_cycle_is_rejected(self, client, auth_headers):
|
|
target = client.post(
|
|
"/api/v1/tools", headers=auth_headers,
|
|
json={"config": "name: cycle-target\nversion: 1.0.0\noutput: ok\n"},
|
|
)
|
|
assert target.status_code == 201
|
|
assert client.post(
|
|
"/api/v1/tools/testuser/cycle-target/deprecate",
|
|
headers=auth_headers,
|
|
json={"replacement": "tool1"},
|
|
).status_code == 200
|
|
response = client.post(
|
|
"/api/v1/tools/testuser/tool1/deprecate",
|
|
headers=auth_headers,
|
|
json={"replacement": "cycle-target"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "cycle" in response.get_json()["error"]["message"]
|
|
|
|
|
|
@flask_required
|
|
class TestPublishPreflightEndpoint:
|
|
def test_dry_run_includes_shared_preflight(self, client, auth_headers):
|
|
response = client.post(
|
|
"/api/v1/tools",
|
|
headers=auth_headers,
|
|
json={
|
|
"dry_run": True,
|
|
"config": (
|
|
"name: preflight-tool\n"
|
|
"version: 1.0.0\n"
|
|
"description: Constant output\n"
|
|
"output: constant\n"
|
|
"input_schema:\n type: string\n"
|
|
"output_schema:\n type: string\n"
|
|
),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
report = response.get_json()["data"]["preflight"]
|
|
assert report["errors"] == []
|
|
assert report["generated_tests"][0]["state"] == "passed"
|
|
assert response.get_json()["data"]["quality"]["evidence_coverage"] > 0
|
|
|
|
def test_real_publish_persists_immutable_audit(self, client, auth_headers):
|
|
from cmdforge.registry.db import connect_db
|
|
|
|
response = client.post(
|
|
"/api/v1/tools",
|
|
headers=auth_headers,
|
|
json={
|
|
"config": (
|
|
"name: audited-tool\nversion: 1.0.0\noutput: stable\n"
|
|
"input_schema: {}\noutput_schema:\n type: string\n"
|
|
),
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
conn = connect_db()
|
|
try:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT tool_audits.trigger, tool_audits.evidence_json
|
|
FROM tool_audits JOIN tools ON tools.id = tool_audits.tool_id
|
|
WHERE tools.owner = 'testuser' AND tools.name = 'audited-tool'
|
|
"""
|
|
).fetchall()
|
|
conn.execute(
|
|
"""
|
|
UPDATE tools SET moderation_status = 'approved'
|
|
WHERE owner = 'testuser' AND name = 'audited-tool'
|
|
"""
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
assert len(rows) == 1
|
|
assert rows[0]["trigger"] == "publish"
|
|
assert "contract_conformance" in rows[0]["evidence_json"]
|
|
search = client.get("/api/v1/tools/search?q=audited")
|
|
result = search.get_json()["data"][0]
|
|
assert isinstance(result["quality_score"], int)
|
|
assert isinstance(result["quality_coverage"], int)
|
|
assert result["quality_evaluated_at"]
|
|
|
|
def test_real_publish_blocks_deterministic_contract_failure(
|
|
self, client, auth_headers
|
|
):
|
|
response = client.post(
|
|
"/api/v1/tools",
|
|
headers=auth_headers,
|
|
json={
|
|
"config": (
|
|
"name: broken-contract\nversion: 1.0.0\noutput: text\n"
|
|
"input_schema: {}\noutput_schema:\n type: integer\n"
|
|
),
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"]["code"] == "PREFLIGHT_FAILED"
|
|
|
|
def test_real_publish_blocks_missing_registry_dependency(
|
|
self, client, auth_headers
|
|
):
|
|
response = client.post(
|
|
"/api/v1/tools",
|
|
headers=auth_headers,
|
|
json={
|
|
"config": (
|
|
"name: missing-dependency\nversion: 1.0.0\n"
|
|
"dependencies:\n - absent-tool\n"
|
|
),
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "absent-tool" in response.get_json()["error"]["message"]
|
|
|
|
def test_invalid_contract_is_rejected(self, client, auth_headers):
|
|
response = client.post(
|
|
"/api/v1/tools",
|
|
headers=auth_headers,
|
|
json={
|
|
"dry_run": True,
|
|
"config": (
|
|
"name: invalid-contract\n"
|
|
"version: 1.0.0\n"
|
|
"output_schema:\n type: nonsense\n"
|
|
),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"]["code"] == "INVALID_CONFIG"
|
|
|
|
def test_non_mapping_config_is_rejected(self, client, auth_headers):
|
|
response = client.post(
|
|
"/api/v1/tools",
|
|
headers=auth_headers,
|
|
json={"dry_run": True, "config": "- not\n- a\n- tool\n"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"]["code"] == "VALIDATION_ERROR"
|
|
|
|
def test_legacy_source_string_remains_publishable(self, client, auth_headers):
|
|
response = client.post(
|
|
"/api/v1/tools",
|
|
headers=auth_headers,
|
|
json={
|
|
"dry_run": True,
|
|
"config": (
|
|
"name: legacy-source\n"
|
|
"version: 1.0.0\n"
|
|
"source: old/tool\n"
|
|
"output: constant\n"
|
|
),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
@flask_required
|
|
class TestM9RegistryTrustAndCommunity:
|
|
def test_signed_publish_download_and_content_lookup(
|
|
self, client, auth_headers
|
|
):
|
|
from cmdforge.attestation import (
|
|
Attestation, generate_keypair, sign_tool, verify_attestation,
|
|
)
|
|
from cmdforge.registry.db import connect_db
|
|
|
|
private_key, public_key = generate_keypair()
|
|
response = client.put(
|
|
"/api/v1/me/signing-key", headers=auth_headers,
|
|
json={"public_key": public_key},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
config = "name: signed-tool\nversion: 1.0.0\noutput: stable\n"
|
|
preflight = client.post(
|
|
"/api/v1/tools", headers=auth_headers,
|
|
json={"config": config, "dry_run": True},
|
|
)
|
|
assert preflight.status_code == 200
|
|
content_hash = preflight.get_json()["data"]["content_hash"]
|
|
attestation = sign_tool(
|
|
"signed-tool", "1.0.0", content_hash, "testuser", private_key
|
|
)
|
|
published = client.post(
|
|
"/api/v1/tools", headers=auth_headers,
|
|
json={"config": config, "attestation": attestation.to_dict()},
|
|
)
|
|
assert published.status_code == 201
|
|
assert published.get_json()["data"]["content_hash"] == content_hash
|
|
|
|
conn = connect_db()
|
|
try:
|
|
conn.execute(
|
|
"UPDATE tools SET moderation_status = 'approved' "
|
|
"WHERE owner = 'testuser' AND name = 'signed-tool'"
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
downloaded = client.get(
|
|
"/api/v1/tools/testuser/signed-tool/download?install=false"
|
|
)
|
|
assert downloaded.status_code == 200
|
|
data = downloaded.get_json()["data"]
|
|
assert data["content_hash"] == content_hash
|
|
assert verify_attestation(Attestation.from_dict(data["attestation"]), public_key)
|
|
lookup = client.get(f"/api/v1/tools/by-content-hash/{content_hash}")
|
|
assert lookup.status_code == 200
|
|
assert lookup.get_json()["data"]["name"] == "signed-tool"
|
|
|
|
def test_registered_key_requires_valid_signature(self, client, auth_headers):
|
|
from cmdforge.attestation import generate_keypair
|
|
_, public_key = generate_keypair()
|
|
assert client.put(
|
|
"/api/v1/me/signing-key", headers=auth_headers,
|
|
json={"public_key": public_key},
|
|
).status_code == 200
|
|
response = client.post(
|
|
"/api/v1/tools", headers=auth_headers,
|
|
json={"config": "name: unsigned-tool\nversion: 1.0.0\n"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"]["code"] == "ATTESTATION_REQUIRED"
|
|
|
|
def test_improvement_is_tested_reviewed_and_credited(
|
|
self, client, auth_headers
|
|
):
|
|
from cmdforge.registry.db import connect_db
|
|
config = (
|
|
"name: improvable\nversion: 1.0.0\noutput: '{result}'\n"
|
|
"input_schema:\n type: string\noutput_schema:\n type: string\n"
|
|
"steps:\n - type: prompt\n prompt: 'Summarize: {input}'\n"
|
|
" provider: mock\n output_var: result\n"
|
|
)
|
|
published = client.post(
|
|
"/api/v1/tools", headers=auth_headers, json={"config": config}
|
|
)
|
|
assert published.status_code == 201
|
|
conn = connect_db()
|
|
try:
|
|
conn.execute(
|
|
"UPDATE tools SET moderation_status = 'approved' "
|
|
"WHERE owner = 'testuser' AND name = 'improvable'"
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
submitted = client.post(
|
|
"/api/v1/tools/testuser/improvable/1.0.0/improvements",
|
|
headers=auth_headers,
|
|
json={
|
|
"step_index": 0,
|
|
"proposed": "Summarize the input accurately and concisely: {input}",
|
|
"rationale": "Clearer expected behavior",
|
|
},
|
|
)
|
|
assert submitted.status_code == 201
|
|
submission_id = submitted.get_json()["data"]["id"]
|
|
reviewed = client.patch(
|
|
f"/api/v1/improvements/{submission_id}", headers=auth_headers,
|
|
json={"decision": "approve", "notes": "Validated"},
|
|
)
|
|
assert reviewed.status_code == 200
|
|
assert reviewed.get_json()["data"]["ready_to_apply"] is True
|
|
improved_config = config.replace("version: 1.0.0", "version: 1.0.1").replace(
|
|
"Summarize: {input}",
|
|
"Summarize the input accurately and concisely: {input}",
|
|
)
|
|
applied = client.post(
|
|
"/api/v1/tools", headers=auth_headers,
|
|
json={"config": improved_config, "improvement_id": submission_id},
|
|
)
|
|
assert applied.status_code == 201
|
|
conn = connect_db()
|
|
try:
|
|
conn.execute(
|
|
"UPDATE tools SET moderation_status = 'approved' "
|
|
"WHERE owner = 'testuser' AND name = 'improvable' "
|
|
"AND version = '1.0.1'"
|
|
)
|
|
conn.commit()
|
|
credited = conn.execute(
|
|
"SELECT COUNT(*) AS count FROM tool_contributors"
|
|
).fetchone()["count"]
|
|
finally:
|
|
conn.close()
|
|
assert credited == 1
|
|
detail = client.get("/api/v1/tools/testuser/improvable?version=1.0.1")
|
|
assert detail.status_code == 200
|
|
assert detail.get_json()["data"]["badges"] == [
|
|
"optimized", "community-reviewed"
|
|
]
|
|
|
|
|
|
@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
|