a2f750269d
- Wire ImportOrchestrator into review_api start_import via BackgroundTasks - Sanitize docstrings in generated tool code to prevent code injection - Add Literal["read", "write"] validation for access_type - Add regex validation for agent_group - Validate URL scheme (http/https only) in ImportRequest - Validate LLM output fields (clamp confidence, validate access_type) - Use dataclasses.replace instead of manual reconstruction in importer - Expand SSRF blocked networks (Carrier-Grade NAT, IPv4-mapped IPv6, etc.) - Make _BLOCKED_NETWORKS immutable tuple - Use yaml.safe_dump instead of yaml.dump - Fix _to_snake_case for empty strings and Python keywords
220 lines
8.3 KiB
Python
220 lines
8.3 KiB
Python
"""Tests for OpenAPI review API endpoints.
|
|
|
|
RED phase: written before implementation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
_SAMPLE_URL = "https://example.com/api/spec.json"
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
"""Create TestClient for the review API app."""
|
|
from fastapi import FastAPI
|
|
|
|
from app.openapi.review_api import router
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture
|
|
def job_id(client):
|
|
"""Create a job and return its ID."""
|
|
response = client.post("/api/openapi/import", json={"url": _SAMPLE_URL})
|
|
assert response.status_code == 202
|
|
return response.json()["job_id"]
|
|
|
|
|
|
@pytest.fixture
|
|
def job_with_classifications(client, job_id):
|
|
"""Return job_id for a job that has mock classifications injected."""
|
|
from app.openapi.models import ClassificationResult, EndpointInfo
|
|
from app.openapi.review_api import _job_store
|
|
|
|
ep = EndpointInfo(
|
|
path="/orders",
|
|
method="GET",
|
|
operation_id="list_orders",
|
|
summary="List orders",
|
|
description="",
|
|
)
|
|
clf = ClassificationResult(
|
|
endpoint=ep,
|
|
access_type="read",
|
|
customer_params=(),
|
|
agent_group="read_agent",
|
|
confidence=0.9,
|
|
needs_interrupt=False,
|
|
)
|
|
# Inject classifications directly into the store
|
|
job = _job_store[job_id]
|
|
_job_store[job_id] = {**job, "classifications": [clf]}
|
|
return job_id
|
|
|
|
|
|
class TestImportEndpoint:
|
|
"""Tests for POST /api/openapi/import."""
|
|
|
|
def test_post_import_returns_job_id(self, client) -> None:
|
|
"""POST /import returns 202 with a job_id."""
|
|
response = client.post("/api/openapi/import", json={"url": _SAMPLE_URL})
|
|
assert response.status_code == 202
|
|
data = response.json()
|
|
assert "job_id" in data
|
|
assert len(data["job_id"]) > 0
|
|
|
|
def test_post_import_empty_url_returns_422(self, client) -> None:
|
|
"""POST /import with empty URL returns 422 validation error."""
|
|
response = client.post("/api/openapi/import", json={"url": ""})
|
|
assert response.status_code == 422
|
|
|
|
def test_post_import_missing_url_returns_422(self, client) -> None:
|
|
"""POST /import with missing URL field returns 422."""
|
|
response = client.post("/api/openapi/import", json={})
|
|
assert response.status_code == 422
|
|
|
|
def test_post_import_invalid_scheme_returns_422(self, client) -> None:
|
|
"""POST /import with non-http URL returns 422."""
|
|
response = client.post("/api/openapi/import", json={"url": "ftp://evil.com/spec"})
|
|
assert response.status_code == 422
|
|
|
|
def test_post_import_returns_pending_status(self, client) -> None:
|
|
"""Newly created job has pending status."""
|
|
response = client.post("/api/openapi/import", json={"url": _SAMPLE_URL})
|
|
data = response.json()
|
|
assert data["status"] == "pending"
|
|
|
|
def test_post_import_returns_spec_url(self, client) -> None:
|
|
"""Response includes the original spec URL."""
|
|
response = client.post("/api/openapi/import", json={"url": _SAMPLE_URL})
|
|
data = response.json()
|
|
assert data["spec_url"] == _SAMPLE_URL
|
|
|
|
|
|
class TestGetJobEndpoint:
|
|
"""Tests for GET /api/openapi/jobs/{job_id}."""
|
|
|
|
def test_get_job_returns_status(self, client, job_id) -> None:
|
|
"""GET /jobs/{id} returns job status."""
|
|
response = client.get(f"/api/openapi/jobs/{job_id}")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "status" in data
|
|
assert "job_id" in data
|
|
|
|
def test_get_unknown_job_returns_404(self, client) -> None:
|
|
"""GET /jobs/nonexistent returns 404."""
|
|
response = client.get("/api/openapi/jobs/nonexistent-id")
|
|
assert response.status_code == 404
|
|
|
|
def test_get_job_includes_spec_url(self, client, job_id) -> None:
|
|
"""Job response includes the spec URL."""
|
|
response = client.get(f"/api/openapi/jobs/{job_id}")
|
|
data = response.json()
|
|
assert data["spec_url"] == _SAMPLE_URL
|
|
|
|
|
|
class TestGetClassificationsEndpoint:
|
|
"""Tests for GET /api/openapi/jobs/{job_id}/classifications."""
|
|
|
|
def test_get_classifications_returns_list(self, client, job_with_classifications) -> None:
|
|
"""GET /classifications returns a list."""
|
|
response = client.get(
|
|
f"/api/openapi/jobs/{job_with_classifications}/classifications"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
|
|
def test_get_classifications_unknown_job_returns_404(self, client) -> None:
|
|
"""GET /classifications for unknown job returns 404."""
|
|
response = client.get("/api/openapi/jobs/unknown/classifications")
|
|
assert response.status_code == 404
|
|
|
|
def test_classification_has_expected_fields(self, client, job_with_classifications) -> None:
|
|
"""Each classification item has access_type and endpoint fields."""
|
|
response = client.get(
|
|
f"/api/openapi/jobs/{job_with_classifications}/classifications"
|
|
)
|
|
item = response.json()[0]
|
|
assert "access_type" in item
|
|
assert "endpoint" in item
|
|
assert "needs_interrupt" in item
|
|
|
|
|
|
class TestUpdateClassificationEndpoint:
|
|
"""Tests for PUT /api/openapi/jobs/{job_id}/classifications/{idx}."""
|
|
|
|
def test_update_classification_succeeds(self, client, job_with_classifications) -> None:
|
|
"""PUT /classifications/0 updates the classification."""
|
|
response = client.put(
|
|
f"/api/openapi/jobs/{job_with_classifications}/classifications/0",
|
|
json={"access_type": "write", "needs_interrupt": True, "agent_group": "write_agent"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_update_unknown_job_returns_404(self, client) -> None:
|
|
"""PUT /classifications/0 for unknown job returns 404."""
|
|
response = client.put(
|
|
"/api/openapi/jobs/unknown/classifications/0",
|
|
json={"access_type": "write", "needs_interrupt": True, "agent_group": "write_agent"},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
def test_update_invalid_access_type_returns_422(self, client, job_with_classifications) -> None:
|
|
"""PUT /classifications/0 with invalid access_type returns 422."""
|
|
response = client.put(
|
|
f"/api/openapi/jobs/{job_with_classifications}/classifications/0",
|
|
json={"access_type": "admin", "needs_interrupt": True, "agent_group": "x"},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
def test_update_invalid_agent_group_returns_422(self, client, job_with_classifications) -> None:
|
|
"""PUT /classifications/0 with invalid agent_group returns 422."""
|
|
response = client.put(
|
|
f"/api/openapi/jobs/{job_with_classifications}/classifications/0",
|
|
json={"access_type": "read", "needs_interrupt": False, "agent_group": "evil group!"},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
def test_update_out_of_range_index_returns_404(self, client, job_with_classifications) -> None:
|
|
"""PUT /classifications/999 returns 404 for out-of-range index."""
|
|
response = client.put(
|
|
f"/api/openapi/jobs/{job_with_classifications}/classifications/999",
|
|
json={"access_type": "read", "needs_interrupt": False, "agent_group": "read_agent"},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestApproveEndpoint:
|
|
"""Tests for POST /api/openapi/jobs/{job_id}/approve."""
|
|
|
|
def test_approve_job_succeeds(self, client, job_with_classifications) -> None:
|
|
"""POST /approve transitions job to approved status."""
|
|
response = client.post(
|
|
f"/api/openapi/jobs/{job_with_classifications}/approve"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_approve_unknown_job_returns_404(self, client) -> None:
|
|
"""POST /approve for unknown job returns 404."""
|
|
response = client.post("/api/openapi/jobs/unknown/approve")
|
|
assert response.status_code == 404
|
|
|
|
def test_approve_returns_job_status(self, client, job_with_classifications) -> None:
|
|
"""POST /approve returns updated job status."""
|
|
response = client.post(
|
|
f"/api/openapi/jobs/{job_with_classifications}/approve"
|
|
)
|
|
data = response.json()
|
|
assert "status" in data
|