117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
"""Docs Gatekeeper API tests."""
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.api.v1 import docs as docs_api
|
|
from app.main import app
|
|
from app.models.user import User
|
|
|
|
|
|
def make_user(role: str = "viewer", groups: list[str] | None = None) -> User:
|
|
user = User(
|
|
id=1,
|
|
username="docs-user",
|
|
email="docs@example.com",
|
|
password_hash="x",
|
|
role=role,
|
|
is_active=True,
|
|
)
|
|
user.gatekeeper_groups = groups or []
|
|
return user
|
|
|
|
|
|
async def get_json(path: str, user: User | None = None):
|
|
if user is not None:
|
|
async def override_user():
|
|
return user
|
|
|
|
app.dependency_overrides[docs_api.get_optional_current_user] = override_user
|
|
|
|
transport = ASGITransport(app=app)
|
|
try:
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
return await client.get(path)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_catalog_only_for_anonymous_user():
|
|
response = await get_json("/api/v1/docs/catalog")
|
|
|
|
assert response.status_code == 200
|
|
items = response.json()["items"]
|
|
assert {item["access"] for item in items} == {"public"}
|
|
assert {item["slug"] for item in items if item["lang"] == "zh"} == {
|
|
"overview",
|
|
"quickstart",
|
|
"manual",
|
|
"faq",
|
|
"location-pipeline-user",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anonymous_can_read_public_doc():
|
|
response = await get_json("/api/v1/docs/zh/quickstart")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["access"] == "public"
|
|
assert "快速开始" in response.json()["markdown"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anonymous_protected_doc_requires_authentication():
|
|
response = await get_json("/api/v1/docs/zh/backend-collectors")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_viewer_without_group_cannot_read_developer_doc():
|
|
response = await get_json(
|
|
"/api/v1/docs/zh/backend-collectors",
|
|
make_user(role="viewer"),
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_developer_group_can_read_developer_but_not_admin_doc():
|
|
user = make_user(role="viewer", groups=["docs_developer"])
|
|
|
|
developer_response = await get_json("/api/v1/docs/zh/backend-collectors", user)
|
|
admin_response = await get_json("/api/v1/docs/zh/backend-system-service-control", user)
|
|
|
|
assert developer_response.status_code == 200
|
|
assert developer_response.json()["access"] == "docs_developer"
|
|
assert admin_response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_and_super_admin_can_read_admin_docs():
|
|
admin_response = await get_json(
|
|
"/api/v1/docs/zh/backend-system-service-control",
|
|
make_user(role="admin"),
|
|
)
|
|
super_admin_response = await get_json(
|
|
"/api/v1/docs/zh/backend-system-service-control",
|
|
make_user(role="super_admin"),
|
|
)
|
|
|
|
assert admin_response.status_code == 200
|
|
assert super_admin_response.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_language_slug_and_path_traversal_do_not_read_files():
|
|
bad_lang = await get_json("/api/v1/docs/fr/quickstart")
|
|
bad_slug = await get_json("/api/v1/docs/zh/not-a-doc")
|
|
traversal = await get_json("/api/v1/docs/zh/..%2Fmanual")
|
|
|
|
assert bad_lang.status_code == 404
|
|
assert bad_slug.status_code == 404
|
|
assert traversal.status_code == 404
|