From 0b261ee21a72eac48c802147c6731f4c03572b51 Mon Sep 17 00:00:00 2001 From: Michael Chihlas Date: Sat, 21 Mar 2026 21:04:33 -0400 Subject: [PATCH] test: add edge case tests for script builder (max sessions, max messages, slug collision, filters) Co-Authored-By: Claude Opus 4.6 (1M context) --- backend/tests/test_script_builder.py | 362 ++++++++++++++++++++++++++- 1 file changed, 359 insertions(+), 3 deletions(-) diff --git a/backend/tests/test_script_builder.py b/backend/tests/test_script_builder.py index 6d42e050..3f12cfda 100644 --- a/backend/tests/test_script_builder.py +++ b/backend/tests/test_script_builder.py @@ -1,7 +1,7 @@ # backend/tests/test_script_builder.py """Tests for the Script Builder API endpoints.""" import pytest -import uuid +import uuid as uuid_mod from unittest.mock import AsyncMock, patch, PropertyMock import sqlalchemy as sa @@ -99,7 +99,7 @@ class TestScriptBuilderSessions: @pytest.mark.asyncio async def test_get_session_not_found(self, client, auth_headers): """Getting a nonexistent session returns 404.""" - fake_id = str(uuid.uuid4()) + fake_id = str(uuid_mod.uuid4()) resp = await client.get( f"/api/v1/scripts/builder/sessions/{fake_id}", headers=auth_headers, @@ -132,7 +132,7 @@ class TestScriptBuilderSessions: @pytest.mark.asyncio async def test_delete_session_not_found(self, client, auth_headers): """Deleting a nonexistent session returns 404.""" - fake_id = str(uuid.uuid4()) + fake_id = str(uuid_mod.uuid4()) resp = await client.delete( f"/api/v1/scripts/builder/sessions/{fake_id}", headers=auth_headers, @@ -308,3 +308,359 @@ class TestScriptBuilderSaveToLibrary: 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"] + await test_db.execute( + sa.text(""" + INSERT INTO script_templates + (id, category_id, created_by, 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, + 'Test Script', 'test-script', 'echo hello', + '{"parameters": []}', '{}', '{}', '["powershell"]', + 'beginner', true, 1, 0, NOW(), NOW()) + """), + {"id": str(uuid_mod.uuid4()), "uid": user_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"] + + 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, 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, 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}, + ) + + # 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, 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, 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}, + ) + 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"] + team_id = user_resp.json().get("team_id") + + cat_id = "b0000000-0000-0000-0000-000000000001" + + # Template shared with user's team + await test_db.execute( + sa.text(""" + INSERT INTO script_templates + (id, category_id, created_by, 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, :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, "tid": team_id}, + ) + + # Template NOT shared (no team_id) + await test_db.execute( + sa.text(""" + INSERT INTO script_templates + (id, category_id, created_by, 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, 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}, + ) + 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