test: add integration tests for script template permissions and share endpoint
Also fix conftest.py to support DATABASE_TEST_URL env var for Docker test runs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Michael Chihlas
@@ -20,7 +20,13 @@ from app.core.config import settings
|
||||
settings.REQUIRE_INVITE_CODE = False
|
||||
|
||||
# Test database URL (separate from production)
|
||||
TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/patherly_test"
|
||||
# Use DATABASE_TEST_URL env var if set (e.g. inside Docker where host is 'db'),
|
||||
# otherwise fall back to localhost for local development.
|
||||
import os
|
||||
TEST_DATABASE_URL = os.environ.get(
|
||||
"DATABASE_TEST_URL",
|
||||
"postgresql+asyncpg://postgres:postgres@localhost:5432/patherly_test",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
|
||||
238
backend/tests/test_script_templates.py
Normal file
238
backend/tests/test_script_templates.py
Normal file
@@ -0,0 +1,238 @@
|
||||
"""Integration tests for Script Template Editor permissions and share endpoint."""
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.user import User
|
||||
from app.models.script_template import ScriptCategory, ScriptTemplate
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
async def _create_category(db: AsyncSession) -> ScriptCategory:
|
||||
"""Seed a script category for tests."""
|
||||
cat = ScriptCategory(name="Active Directory", slug="active-directory", sort_order=1)
|
||||
db.add(cat)
|
||||
await db.commit()
|
||||
await db.refresh(cat)
|
||||
return cat
|
||||
|
||||
|
||||
async def _make_owner(db: AsyncSession, user_id: str) -> None:
|
||||
"""Promote a user to account owner."""
|
||||
from uuid import UUID as PyUUID
|
||||
result = await db.execute(select(User).where(User.id == PyUUID(user_id)))
|
||||
user = result.scalar_one()
|
||||
user.account_role = "owner"
|
||||
await db.commit()
|
||||
|
||||
|
||||
async def _register_and_login(client: AsyncClient, email: str, password: str, name: str) -> tuple[dict, str]:
|
||||
"""Register a user, login, return (user_data, access_token)."""
|
||||
resp = await client.post("/api/v1/auth/register", json={"email": email, "password": password, "name": name})
|
||||
assert resp.status_code in (200, 201)
|
||||
user_data = resp.json()
|
||||
login_resp = await client.post("/api/v1/auth/login/json", json={"email": email, "password": password})
|
||||
assert login_resp.status_code == 200
|
||||
token = login_resp.json()["access_token"]
|
||||
return user_data, token
|
||||
|
||||
|
||||
TEMPLATE_PAYLOAD = {
|
||||
"name": "Test Template",
|
||||
"script_body": "Write-Host '{{ message }}'",
|
||||
"parameters_schema": {
|
||||
"parameters": [
|
||||
{"key": "message", "label": "Message", "type": "text", "required": True, "order": 1}
|
||||
]
|
||||
},
|
||||
"complexity": "beginner",
|
||||
}
|
||||
|
||||
|
||||
# ── Tests ────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestScriptTemplatePermissions:
|
||||
"""Test that engineers can create/edit their own templates, but not others'."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_engineer_can_create_template(self, client, auth_headers, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
resp = await client.post("/api/v1/scripts/templates", json=payload, headers=auth_headers)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "Test Template"
|
||||
assert data["created_by"] is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_engineer_can_edit_own_template(self, client, auth_headers, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=auth_headers)
|
||||
template_id = create_resp.json()["id"]
|
||||
|
||||
update_resp = await client.put(
|
||||
f"/api/v1/scripts/templates/{template_id}",
|
||||
json={"name": "Updated Template"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert update_resp.status_code == 200
|
||||
assert update_resp.json()["name"] == "Updated Template"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_engineer_cannot_edit_others_template(self, client, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
|
||||
# Engineer A creates a template
|
||||
_, token_a = await _register_and_login(client, "engineer_a@example.com", "TestPass123!", "Engineer A")
|
||||
headers_a = {"Authorization": f"Bearer {token_a}"}
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers_a)
|
||||
template_id = create_resp.json()["id"]
|
||||
|
||||
# Engineer B tries to edit it
|
||||
_, token_b = await _register_and_login(client, "engineer_b@example.com", "TestPass123!", "Engineer B")
|
||||
headers_b = {"Authorization": f"Bearer {token_b}"}
|
||||
update_resp = await client.put(
|
||||
f"/api/v1/scripts/templates/{template_id}",
|
||||
json={"name": "Hijacked!"},
|
||||
headers=headers_b,
|
||||
)
|
||||
assert update_resp.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_engineer_can_delete_own_template(self, client, auth_headers, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=auth_headers)
|
||||
template_id = create_resp.json()["id"]
|
||||
|
||||
delete_resp = await client.delete(f"/api/v1/scripts/templates/{template_id}", headers=auth_headers)
|
||||
assert delete_resp.status_code == 204
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_viewer_cannot_create_template(self, client, test_db):
|
||||
_, token = await _register_and_login(client, "viewer@example.com", "TestPass123!", "Viewer")
|
||||
# Downgrade to viewer
|
||||
result = await test_db.execute(select(User).where(User.email == "viewer@example.com"))
|
||||
user = result.scalar_one()
|
||||
user.role = "viewer"
|
||||
user.account_role = "viewer"
|
||||
await test_db.commit()
|
||||
|
||||
# Re-login to get new token with updated role
|
||||
login_resp = await client.post("/api/v1/auth/login/json", json={"email": "viewer@example.com", "password": "TestPass123!"})
|
||||
token = login_resp.json()["access_token"]
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
cat = await _create_category(test_db)
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
|
||||
assert resp.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_can_edit_others_template(self, client, test_db, admin_auth_headers):
|
||||
cat = await _create_category(test_db)
|
||||
# Create template as a regular engineer
|
||||
_, token = await _register_and_login(client, "eng@example.com", "TestPass123!", "Eng")
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
|
||||
template_id = create_resp.json()["id"]
|
||||
|
||||
# Admin edits it
|
||||
update_resp = await client.put(
|
||||
f"/api/v1/scripts/templates/{template_id}",
|
||||
json={"name": "Admin Updated"},
|
||||
headers=admin_auth_headers,
|
||||
)
|
||||
assert update_resp.status_code == 200
|
||||
assert update_resp.json()["name"] == "Admin Updated"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_managed_filter_returns_own_templates(self, client, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
|
||||
# Engineer A creates a template
|
||||
_, token_a = await _register_and_login(client, "eng_a2@example.com", "TestPass123!", "Eng A")
|
||||
headers_a = {"Authorization": f"Bearer {token_a}"}
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
await client.post("/api/v1/scripts/templates", json=payload, headers=headers_a)
|
||||
|
||||
# Engineer B should not see A's template in managed view
|
||||
_, token_b = await _register_and_login(client, "eng_b2@example.com", "TestPass123!", "Eng B")
|
||||
headers_b = {"Authorization": f"Bearer {token_b}"}
|
||||
|
||||
resp = await client.get("/api/v1/scripts/templates?managed=true", headers=headers_b)
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()) == 0
|
||||
|
||||
|
||||
class TestScriptTemplateShare:
|
||||
"""Test the share/unshare endpoint."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_owner_can_share_template(self, client, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
|
||||
# Registration auto-sets account_role="owner", so this user is already an owner
|
||||
user_data, token = await _register_and_login(client, "eng_share@example.com", "TestPass123!", "Eng")
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
|
||||
template_id = create_resp.json()["id"]
|
||||
|
||||
# Share — owner should be allowed
|
||||
share_resp = await client.patch(
|
||||
f"/api/v1/scripts/templates/{template_id}/share?shared=true",
|
||||
headers=headers,
|
||||
)
|
||||
assert share_resp.status_code == 200
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_engineer_cannot_share_template(self, client, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
|
||||
# Create a user and downgrade to engineer (registration sets owner by default)
|
||||
user_data, token = await _register_and_login(client, "eng_noshare@example.com", "TestPass123!", "Eng NoShare")
|
||||
result = await test_db.execute(select(User).where(User.email == "eng_noshare@example.com"))
|
||||
user = result.scalar_one()
|
||||
user.account_role = "engineer"
|
||||
await test_db.commit()
|
||||
|
||||
# Re-login to get fresh token with updated role
|
||||
login_resp = await client.post("/api/v1/auth/login/json", json={"email": "eng_noshare@example.com", "password": "TestPass123!"})
|
||||
eng_token = login_resp.json()["access_token"]
|
||||
eng_headers = {"Authorization": f"Bearer {eng_token}"}
|
||||
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=eng_headers)
|
||||
template_id = create_resp.json()["id"]
|
||||
|
||||
share_resp = await client.patch(
|
||||
f"/api/v1/scripts/templates/{template_id}/share?shared=true",
|
||||
headers=eng_headers,
|
||||
)
|
||||
assert share_resp.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_owner_can_unshare_template(self, client, test_db):
|
||||
cat = await _create_category(test_db)
|
||||
|
||||
# Registration auto-sets account_role="owner"
|
||||
user_data, token = await _register_and_login(client, "eng_unshare@example.com", "TestPass123!", "Eng")
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
|
||||
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
|
||||
template_id = create_resp.json()["id"]
|
||||
|
||||
# Share then unshare
|
||||
await client.patch(f"/api/v1/scripts/templates/{template_id}/share?shared=true", headers=headers)
|
||||
unshare_resp = await client.patch(
|
||||
f"/api/v1/scripts/templates/{template_id}/share?shared=false",
|
||||
headers=headers,
|
||||
)
|
||||
assert unshare_resp.status_code == 200
|
||||
assert unshare_resp.json()["team_id"] is None
|
||||
Reference in New Issue
Block a user