244 lines
11 KiB
Python
244 lines
11 KiB
Python
"""Integration tests for Script Template Editor permissions and share endpoint."""
|
|
from uuid import UUID as PyUUID
|
|
|
|
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
|
|
result = await test_db.execute(select(ScriptTemplate).where(ScriptTemplate.id == PyUUID(data["id"])))
|
|
template = result.scalar_one()
|
|
assert template.account_id 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
|