684 lines
26 KiB
Python
684 lines
26 KiB
Python
# backend/tests/test_script_builder.py
|
|
"""Tests for the Script Builder API endpoints."""
|
|
import pytest
|
|
import uuid as uuid_mod
|
|
from unittest.mock import AsyncMock, patch, PropertyMock
|
|
|
|
import sqlalchemy as sa
|
|
|
|
|
|
class TestScriptBuilderSessions:
|
|
"""Test Script Builder session CRUD."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_session(self, client, auth_headers):
|
|
"""Creating a builder session returns a valid session."""
|
|
resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["language"] == "powershell"
|
|
assert data["messages"] == []
|
|
assert data["message_count"] == 0
|
|
assert data["title"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_session_invalid_language(self, client, auth_headers):
|
|
"""Invalid language is rejected."""
|
|
resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "cobol"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_session_bash(self, client, auth_headers):
|
|
"""Bash language is accepted."""
|
|
resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "bash"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["language"] == "bash"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions_empty(self, client, auth_headers):
|
|
"""Listing sessions when none exist returns empty list."""
|
|
resp = await client.get(
|
|
"/api/v1/scripts/builder/sessions",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions_returns_summaries(self, client, auth_headers):
|
|
"""Listed sessions are lightweight (no messages field)."""
|
|
# Create a session first
|
|
await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
resp = await client.get(
|
|
"/api/v1/scripts/builder/sessions",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 1
|
|
assert "messages" not in data[0]
|
|
assert "language" in data[0]
|
|
assert "title" in data[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_detail(self, client, auth_headers):
|
|
"""Getting a session by ID returns full detail with messages."""
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "python"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
resp = await client.get(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["id"] == session_id
|
|
assert "messages" in data
|
|
assert data["language"] == "python"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_not_found(self, client, auth_headers):
|
|
"""Getting a nonexistent session returns 404."""
|
|
fake_id = str(uuid_mod.uuid4())
|
|
resp = await client.get(
|
|
f"/api/v1/scripts/builder/sessions/{fake_id}",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_session(self, client, auth_headers):
|
|
"""Deleting a session removes it."""
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
resp = await client.delete(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# Verify it's gone
|
|
resp = await client.get(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_session_not_found(self, client, auth_headers):
|
|
"""Deleting a nonexistent session returns 404."""
|
|
fake_id = str(uuid_mod.uuid4())
|
|
resp = await client.delete(
|
|
f"/api/v1/scripts/builder/sessions/{fake_id}",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestScriptBuilderMessages:
|
|
"""Test sending messages (requires AI mock)."""
|
|
|
|
@pytest.fixture
|
|
def _enable_ai(self):
|
|
"""Mock AI as enabled for tests without API key."""
|
|
with patch.object(
|
|
type(__import__("app.core.config", fromlist=["settings"]).settings),
|
|
"ai_enabled",
|
|
new_callable=PropertyMock,
|
|
return_value=True,
|
|
):
|
|
yield
|
|
|
|
@pytest.fixture
|
|
def _mock_ai_response(self):
|
|
"""Mock the AI provider to return a script response."""
|
|
mock_response = (
|
|
'Here is your script:\n\n```powershell\nGet-Process | Format-Table\n```\n\nSaved as `Get-Processes.ps1`.',
|
|
100, # input_tokens
|
|
200, # output_tokens
|
|
)
|
|
with patch("app.services.script_builder_service.get_ai_provider") as mock:
|
|
provider = AsyncMock()
|
|
provider.generate_text = AsyncMock(return_value=mock_response)
|
|
mock.return_value = provider
|
|
yield
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_message(self, client, auth_headers, _enable_ai, _mock_ai_response):
|
|
"""Sending a message returns AI response with script."""
|
|
# Create session
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
# Send message
|
|
resp = await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/messages",
|
|
json={"content": "List all running processes"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["role"] == "assistant"
|
|
assert data["script"] is not None
|
|
assert "Get-Process" in data["script"]
|
|
assert data["script_filename"] == "Get-Processes.ps1"
|
|
assert data["line_count"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_message_updates_session(self, client, auth_headers, _enable_ai, _mock_ai_response):
|
|
"""Sending a message updates session state."""
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/messages",
|
|
json={"content": "List processes"},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Check session was updated
|
|
resp = await client.get(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}",
|
|
headers=auth_headers,
|
|
)
|
|
data = resp.json()
|
|
assert data["message_count"] == 1
|
|
assert data["latest_script"] is not None
|
|
assert len(data["messages"]) == 2 # user + assistant
|
|
assert data["title"] is not None
|
|
|
|
|
|
class TestScriptBuilderSaveToLibrary:
|
|
"""Test saving generated scripts to the library."""
|
|
|
|
@pytest.fixture
|
|
def _enable_ai(self):
|
|
with patch.object(
|
|
type(__import__("app.core.config", fromlist=["settings"]).settings),
|
|
"ai_enabled",
|
|
new_callable=PropertyMock,
|
|
return_value=True,
|
|
):
|
|
yield
|
|
|
|
@pytest.fixture
|
|
def _mock_ai_response(self):
|
|
mock_response = (
|
|
'Here is your script:\n\n```powershell\nGet-Process | Format-Table\n```\n\nSaved as `Get-Processes.ps1`.',
|
|
100, 200,
|
|
)
|
|
with patch("app.services.script_builder_service.get_ai_provider") as mock:
|
|
provider = AsyncMock()
|
|
provider.generate_text = AsyncMock(return_value=mock_response)
|
|
mock.return_value = provider
|
|
yield
|
|
|
|
@pytest.fixture
|
|
async def _seed_ai_category(self, test_db):
|
|
"""Seed the AI Generated category for save tests."""
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_categories (id, name, slug, description, icon, sort_order, is_active, created_at, updated_at)
|
|
VALUES (
|
|
'a0000000-0000-0000-0000-000000000001'::uuid,
|
|
'AI Generated', 'ai-generated', 'Scripts from AI', 'sparkles', 100, true, NOW(), NOW()
|
|
)
|
|
ON CONFLICT (slug) DO NOTHING
|
|
""")
|
|
)
|
|
await test_db.commit()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_to_library_no_script(self, client, auth_headers):
|
|
"""Cannot save if no script has been generated."""
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/save",
|
|
json={"name": "Test Script"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "No script" in resp.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_to_library_success(
|
|
self, client, auth_headers, _enable_ai, _mock_ai_response, _seed_ai_category
|
|
):
|
|
"""Saving a generated script creates a ScriptTemplate."""
|
|
# Create session and generate a script
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/messages",
|
|
json={"content": "List processes"},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Save to library
|
|
resp = await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/save",
|
|
json={"name": "My Process Script", "share_with_team": False},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["name"] == "My Process Script"
|
|
assert "ai-generated" in data["tags"]
|
|
|
|
|
|
class TestScriptBuilderMaxSessions:
|
|
"""Test the per-user session limit (MAX_SESSIONS_PER_USER = 5)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_max_sessions_limit(self, client, auth_headers):
|
|
"""Creating a 6th session should return 400."""
|
|
# Create 5 sessions (the maximum)
|
|
for i in range(5):
|
|
resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201, f"Session {i+1} should succeed"
|
|
|
|
# 6th session should be rejected
|
|
resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "Maximum" in resp.json()["detail"]
|
|
assert "5" in resp.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_max_sessions_after_delete_allows_new(self, client, auth_headers):
|
|
"""After deleting a session, creating a new one should succeed."""
|
|
session_ids = []
|
|
for _ in range(5):
|
|
resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_ids.append(resp.json()["id"])
|
|
|
|
# Delete one
|
|
await client.delete(
|
|
f"/api/v1/scripts/builder/sessions/{session_ids[0]}",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Now creating a new session should work
|
|
resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "bash"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
|
|
class TestScriptBuilderMaxMessages:
|
|
"""Test the per-session message limit (MAX_MESSAGES_PER_SESSION = 30)."""
|
|
|
|
@pytest.fixture
|
|
def _enable_ai(self):
|
|
with patch.object(
|
|
type(__import__("app.core.config", fromlist=["settings"]).settings),
|
|
"ai_enabled",
|
|
new_callable=PropertyMock,
|
|
return_value=True,
|
|
):
|
|
yield
|
|
|
|
@pytest.fixture
|
|
def _mock_ai_response(self):
|
|
mock_response = (
|
|
'Here is your script:\n\n```powershell\nGet-Process\n```\n\nSaved as `Get-Processes.ps1`.',
|
|
100, 200,
|
|
)
|
|
with patch("app.services.script_builder_service.get_ai_provider") as mock:
|
|
provider = AsyncMock()
|
|
provider.generate_text = AsyncMock(return_value=mock_response)
|
|
mock.return_value = provider
|
|
yield
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_max_messages_limit(
|
|
self, client, auth_headers, test_db, _enable_ai, _mock_ai_response
|
|
):
|
|
"""Sending a message when session has 30 user messages should return 400."""
|
|
# Create a session
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
# Insert 30 user message rows directly into the DB to reach the limit
|
|
for i in range(30):
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_builder_messages (id, session_id, role, content, created_at)
|
|
VALUES (:id, :sid, 'user', :content, NOW())
|
|
"""),
|
|
{
|
|
"id": str(uuid_mod.uuid4()),
|
|
"sid": session_id,
|
|
"content": f"message {i+1}",
|
|
},
|
|
)
|
|
await test_db.commit()
|
|
|
|
# Now sending a message should fail with 400
|
|
resp = await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/messages",
|
|
json={"content": "One more message"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "maximum" in resp.json()["detail"].lower()
|
|
|
|
|
|
class TestScriptBuilderSlugCollision:
|
|
"""Test that saving a script with a colliding slug generates a unique slug."""
|
|
|
|
@pytest.fixture
|
|
def _enable_ai(self):
|
|
with patch.object(
|
|
type(__import__("app.core.config", fromlist=["settings"]).settings),
|
|
"ai_enabled",
|
|
new_callable=PropertyMock,
|
|
return_value=True,
|
|
):
|
|
yield
|
|
|
|
@pytest.fixture
|
|
def _mock_ai_response(self):
|
|
mock_response = (
|
|
'Here is your script:\n\n```powershell\nGet-Process | Format-Table\n```\n\nSaved as `Get-Processes.ps1`.',
|
|
100, 200,
|
|
)
|
|
with patch("app.services.script_builder_service.get_ai_provider") as mock:
|
|
provider = AsyncMock()
|
|
provider.generate_text = AsyncMock(return_value=mock_response)
|
|
mock.return_value = provider
|
|
yield
|
|
|
|
@pytest.fixture
|
|
async def _seed_ai_category(self, test_db):
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_categories (id, name, slug, description, icon, sort_order, is_active, created_at, updated_at)
|
|
VALUES (
|
|
'a0000000-0000-0000-0000-000000000001'::uuid,
|
|
'AI Generated', 'ai-generated', 'Scripts from AI', 'sparkles', 100, true, NOW(), NOW()
|
|
)
|
|
ON CONFLICT (slug) DO NOTHING
|
|
""")
|
|
)
|
|
await test_db.commit()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slug_collision_appends_suffix(
|
|
self, client, auth_headers, test_db, _enable_ai, _mock_ai_response, _seed_ai_category
|
|
):
|
|
"""When a slug already exists, the saved template gets a UUID-suffixed slug."""
|
|
# Pre-create a template with slug "test-script" to cause collision
|
|
user_resp = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
user_id = user_resp.json()["id"]
|
|
account_id = user_resp.json()["account_id"]
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_templates
|
|
(id, category_id, created_by, account_id, name, slug, script_body,
|
|
parameters_schema, default_values, validation_rules, tags,
|
|
complexity, is_active, version, usage_count, created_at, updated_at)
|
|
VALUES
|
|
(:id, 'a0000000-0000-0000-0000-000000000001'::uuid, :uid, :account_id,
|
|
'Test Script', 'test-script', 'echo hello',
|
|
'{"parameters": []}', '{}', '{}', '["powershell"]',
|
|
'beginner', true, 1, 0, NOW(), NOW())
|
|
"""),
|
|
{"id": str(uuid_mod.uuid4()), "uid": user_id, "account_id": account_id},
|
|
)
|
|
await test_db.commit()
|
|
|
|
# Create a builder session and generate a script
|
|
create_resp = await client.post(
|
|
"/api/v1/scripts/builder/sessions",
|
|
json={"language": "powershell"},
|
|
headers=auth_headers,
|
|
)
|
|
session_id = create_resp.json()["id"]
|
|
|
|
await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/messages",
|
|
json={"content": "List processes"},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Save with a name that would generate slug "test-script"
|
|
resp = await client.post(
|
|
f"/api/v1/scripts/builder/sessions/{session_id}/save",
|
|
json={"name": "Test Script"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
# The slug should start with "test-script-" (base + UUID suffix)
|
|
assert data["slug"].startswith("test-script-")
|
|
assert data["slug"] != "test-script"
|
|
|
|
|
|
class TestScriptTemplateFilters:
|
|
"""Test mine/shared query filters on GET /scripts/templates."""
|
|
|
|
@pytest.fixture
|
|
async def _seed_category(self, test_db):
|
|
"""Seed a script category for template creation."""
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_categories (id, name, slug, description, icon, sort_order, is_active, created_at, updated_at)
|
|
VALUES (
|
|
'b0000000-0000-0000-0000-000000000001'::uuid,
|
|
'General', 'general', 'General scripts', 'code', 0, true, NOW(), NOW()
|
|
)
|
|
ON CONFLICT (slug) DO NOTHING
|
|
""")
|
|
)
|
|
await test_db.commit()
|
|
|
|
@pytest.fixture
|
|
async def second_user_headers(self, client, test_db):
|
|
"""Create a second user on the same team as the test user and return their auth headers."""
|
|
# Register second user
|
|
user_data = {
|
|
"email": "second@example.com",
|
|
"password": "TestPassword123!",
|
|
"name": "Second User",
|
|
}
|
|
resp = await client.post("/api/v1/auth/register", json=user_data)
|
|
assert resp.status_code in (200, 201)
|
|
|
|
# Login to get headers
|
|
login_resp = await client.post(
|
|
"/api/v1/auth/login/json",
|
|
json={"email": user_data["email"], "password": user_data["password"]},
|
|
)
|
|
assert login_resp.status_code == 200
|
|
token = login_resp.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mine_filter(
|
|
self, client, auth_headers, test_db, test_user, second_user_headers, _seed_category
|
|
):
|
|
"""mine=true returns only templates created by the current user."""
|
|
user_resp = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
user_id = user_resp.json()["id"]
|
|
account_id = user_resp.json()["account_id"]
|
|
|
|
second_resp = await client.get("/api/v1/auth/me", headers=second_user_headers)
|
|
second_user_id = second_resp.json()["id"]
|
|
|
|
cat_id = "b0000000-0000-0000-0000-000000000001"
|
|
|
|
# Create template owned by test user
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_templates
|
|
(id, category_id, created_by, account_id, team_id, name, slug, script_body,
|
|
parameters_schema, default_values, validation_rules, tags,
|
|
complexity, is_active, version, usage_count, created_at, updated_at)
|
|
VALUES
|
|
(:id, :cat, :uid, :account_id, NULL,
|
|
'My Script', 'my-script', 'echo mine',
|
|
'{"parameters": []}', '{}', '{}', '[]',
|
|
'beginner', true, 1, 0, NOW(), NOW())
|
|
"""),
|
|
{"id": str(uuid_mod.uuid4()), "cat": cat_id, "uid": user_id, "account_id": account_id},
|
|
)
|
|
|
|
# Create template owned by second user (no team_id, so visible to all)
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_templates
|
|
(id, category_id, created_by, account_id, team_id, name, slug, script_body,
|
|
parameters_schema, default_values, validation_rules, tags,
|
|
complexity, is_active, version, usage_count, created_at, updated_at)
|
|
VALUES
|
|
(:id, :cat, :uid, :account_id, NULL,
|
|
'Other Script', 'other-script', 'echo other',
|
|
'{"parameters": []}', '{}', '{}', '[]',
|
|
'beginner', true, 1, 0, NOW(), NOW())
|
|
"""),
|
|
{"id": str(uuid_mod.uuid4()), "cat": cat_id, "uid": second_user_id, "account_id": account_id},
|
|
)
|
|
await test_db.commit()
|
|
|
|
# mine=true should only return the test user's template
|
|
resp = await client.get(
|
|
"/api/v1/scripts/templates?mine=true",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 1
|
|
assert data[0]["name"] == "My Script"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shared_filter(
|
|
self, client, auth_headers, test_db, test_user, _seed_category
|
|
):
|
|
"""shared=true returns only templates shared with the user's team."""
|
|
user_resp = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
user_id = user_resp.json()["id"]
|
|
account_id = user_resp.json()["account_id"]
|
|
|
|
cat_id = "b0000000-0000-0000-0000-000000000001"
|
|
|
|
# Create a team and assign the test user to it
|
|
team_id = str(uuid_mod.uuid4())
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO teams (id, name, created_at)
|
|
VALUES (:tid, 'Test Team', NOW())
|
|
"""),
|
|
{"tid": team_id},
|
|
)
|
|
await test_db.execute(
|
|
sa.text("UPDATE users SET team_id = :tid WHERE id = :uid"),
|
|
{"tid": team_id, "uid": user_id},
|
|
)
|
|
await test_db.commit()
|
|
|
|
# Template shared with user's team
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_templates
|
|
(id, category_id, created_by, account_id, team_id, name, slug, script_body,
|
|
parameters_schema, default_values, validation_rules, tags,
|
|
complexity, is_active, version, usage_count, created_at, updated_at)
|
|
VALUES
|
|
(:id, :cat, :uid, :account_id, :tid,
|
|
'Team Script', 'team-script', 'echo team',
|
|
'{"parameters": []}', '{}', '{}', '[]',
|
|
'beginner', true, 1, 0, NOW(), NOW())
|
|
"""),
|
|
{"id": str(uuid_mod.uuid4()), "cat": cat_id, "uid": user_id, "account_id": account_id, "tid": team_id},
|
|
)
|
|
|
|
# Template NOT shared (no team_id)
|
|
await test_db.execute(
|
|
sa.text("""
|
|
INSERT INTO script_templates
|
|
(id, category_id, created_by, account_id, team_id, name, slug, script_body,
|
|
parameters_schema, default_values, validation_rules, tags,
|
|
complexity, is_active, version, usage_count, created_at, updated_at)
|
|
VALUES
|
|
(:id, :cat, :uid, :account_id, NULL,
|
|
'Personal Script', 'personal-script', 'echo personal',
|
|
'{"parameters": []}', '{}', '{}', '[]',
|
|
'beginner', true, 1, 0, NOW(), NOW())
|
|
"""),
|
|
{"id": str(uuid_mod.uuid4()), "cat": cat_id, "uid": user_id, "account_id": account_id},
|
|
)
|
|
await test_db.commit()
|
|
|
|
# shared=true should only return the team-shared template
|
|
resp = await client.get(
|
|
"/api/v1/scripts/templates?shared=true",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
names = [t["name"] for t in data]
|
|
assert "Team Script" in names
|
|
assert "Personal Script" not in names
|