test_comments.py
1 import random 2 import pytest 3 from fastapi.testclient import TestClient 4 5 from restai.config import RESTAI_DEFAULT_PASSWORD 6 from restai.main import app 7 8 9 @pytest.fixture(scope="module") 10 def client(): 11 with TestClient(app) as c: 12 yield c 13 14 15 test_username = "test_comments_user_" + str(random.randint(0, 1000000)) 16 test_password = "comments_test_pass" 17 test_project_id = None 18 comment_id = None 19 20 21 def test_setup(client): 22 """Create test user and a project for comment tests.""" 23 global test_project_id 24 # Create user 25 client.post( 26 "/users", 27 json={"username": test_username, "password": test_password, "admin": False, "private": False}, 28 auth=("admin", RESTAI_DEFAULT_PASSWORD), 29 ) 30 31 # Create a test LLM 32 llm_name = f"comments_llm_{random.randint(0,999999)}" 33 client.post( 34 "/llms", 35 json={"name": llm_name, "class_name": "OpenAI", "options": {"model": "gpt-test", "api_key": "sk-fake"}, "privacy": "public"}, 36 auth=("admin", RESTAI_DEFAULT_PASSWORD), 37 ) 38 39 # Create a team and add user + LLM 40 team_name = f"comments_team_{random.randint(0,999999)}" 41 resp = client.post( 42 "/teams", 43 json={"name": team_name, "users": [test_username], "admins": [], "llms": [llm_name]}, 44 auth=("admin", RESTAI_DEFAULT_PASSWORD), 45 ) 46 team_id = resp.json()["id"] 47 48 # Create project 49 proj_name = f"comments_proj_{random.randint(0,999999)}" 50 resp = client.post( 51 "/projects", 52 json={"name": proj_name, "type": "agent", "llm": llm_name, "team_id": team_id}, 53 auth=("admin", RESTAI_DEFAULT_PASSWORD), 54 ) 55 assert resp.status_code == 201 56 test_project_id = resp.json()["project"] 57 58 # Assign test user to the project 59 client.patch( 60 f"/projects/{test_project_id}", 61 json={"users": ["admin", test_username]}, 62 auth=("admin", RESTAI_DEFAULT_PASSWORD), 63 ) 64 65 66 def test_list_comments_initial(client): 67 resp = client.get( 68 f"/projects/{test_project_id}/comments", 69 auth=("admin", RESTAI_DEFAULT_PASSWORD), 70 ) 71 assert resp.status_code == 200 72 # Clean up any leftover comments from previous runs 73 for c in resp.json(): 74 client.delete(f"/projects/{test_project_id}/comments/{c['id']}", auth=("admin", RESTAI_DEFAULT_PASSWORD)) 75 76 77 def test_create_comment(client): 78 global comment_id 79 resp = client.post( 80 f"/projects/{test_project_id}/comments", 81 json={"content": "This project works great for product questions."}, 82 auth=("admin", RESTAI_DEFAULT_PASSWORD), 83 ) 84 assert resp.status_code == 201 85 data = resp.json() 86 assert "id" in data 87 comment_id = data["id"] 88 89 90 def test_list_comments_after_create(client): 91 resp = client.get( 92 f"/projects/{test_project_id}/comments", 93 auth=("admin", RESTAI_DEFAULT_PASSWORD), 94 ) 95 assert resp.status_code == 200 96 comments = resp.json() 97 assert len(comments) == 1 98 assert comments[0]["content"] == "This project works great for product questions." 99 assert comments[0]["username"] == "admin" 100 assert "created_at" in comments[0] 101 102 103 def test_create_second_comment(client): 104 resp = client.post( 105 f"/projects/{test_project_id}/comments", 106 json={"content": "Struggles with pricing questions though."}, 107 auth=(test_username, test_password), 108 ) 109 assert resp.status_code == 201 110 111 112 def test_list_comments_order(client): 113 """Comments should be newest first.""" 114 resp = client.get( 115 f"/projects/{test_project_id}/comments", 116 auth=("admin", RESTAI_DEFAULT_PASSWORD), 117 ) 118 comments = resp.json() 119 assert len(comments) == 2 120 # Newest first 121 assert comments[0]["content"] == "Struggles with pricing questions though." 122 assert comments[1]["content"] == "This project works great for product questions." 123 124 125 def test_update_own_comment(client): 126 resp = client.patch( 127 f"/projects/{test_project_id}/comments/{comment_id}", 128 json={"content": "Updated: works great for product AND support questions."}, 129 auth=("admin", RESTAI_DEFAULT_PASSWORD), 130 ) 131 assert resp.status_code == 200 132 133 # Verify update 134 comments = client.get( 135 f"/projects/{test_project_id}/comments", 136 auth=("admin", RESTAI_DEFAULT_PASSWORD), 137 ).json() 138 updated = [c for c in comments if c["id"] == comment_id][0] 139 assert "Updated:" in updated["content"] 140 141 142 def test_non_owner_cannot_update(client): 143 resp = client.patch( 144 f"/projects/{test_project_id}/comments/{comment_id}", 145 json={"content": "Hacked!"}, 146 auth=(test_username, test_password), 147 ) 148 assert resp.status_code == 403 149 assert "own comments" in resp.json()["detail"] 150 151 152 def test_non_owner_cannot_delete(client): 153 resp = client.delete( 154 f"/projects/{test_project_id}/comments/{comment_id}", 155 auth=(test_username, test_password), 156 ) 157 assert resp.status_code == 403 158 159 160 def test_delete_own_comment(client): 161 # First create a comment as test_user, then delete it 162 create_resp = client.post( 163 f"/projects/{test_project_id}/comments", 164 json={"content": "Temporary note."}, 165 auth=(test_username, test_password), 166 ) 167 temp_id = create_resp.json()["id"] 168 169 resp = client.delete( 170 f"/projects/{test_project_id}/comments/{temp_id}", 171 auth=(test_username, test_password), 172 ) 173 assert resp.status_code == 200 174 175 176 def test_admin_can_delete_any(client): 177 """Admin can delete other users' comments.""" 178 # Get the test_user's remaining comment 179 comments = client.get( 180 f"/projects/{test_project_id}/comments", 181 auth=("admin", RESTAI_DEFAULT_PASSWORD), 182 ).json() 183 user_comment = [c for c in comments if c["username"] == test_username] 184 assert len(user_comment) > 0 185 186 resp = client.delete( 187 f"/projects/{test_project_id}/comments/{user_comment[0]['id']}", 188 auth=("admin", RESTAI_DEFAULT_PASSWORD), 189 ) 190 assert resp.status_code == 200 191 192 193 def test_comment_not_found(client): 194 resp = client.patch( 195 f"/projects/{test_project_id}/comments/999999", 196 json={"content": "nope"}, 197 auth=("admin", RESTAI_DEFAULT_PASSWORD), 198 ) 199 assert resp.status_code == 404 200 201 202 def test_empty_comment_rejected(client): 203 resp = client.post( 204 f"/projects/{test_project_id}/comments", 205 json={"content": ""}, 206 auth=("admin", RESTAI_DEFAULT_PASSWORD), 207 ) 208 # Pydantic validation: empty string with max_length should still pass, 209 # but we could also check if server rejects it 210 # At minimum it should not crash 211 assert resp.status_code in (201, 422) 212 213 214 def test_cleanup(client): 215 client.delete(f"/projects/{test_project_id}", auth=("admin", RESTAI_DEFAULT_PASSWORD)) 216 client.delete(f"/users/{test_username}", auth=("admin", RESTAI_DEFAULT_PASSWORD))