test_project_invitations.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 suffix = str(random.randint(0, 1000000)) 9 team_name = f"inv_team_{suffix}" 10 project_name = f"inv_project_{suffix}" 11 user_in_team = f"inv_user_{suffix}" 12 user_outside = f"inv_outsider_{suffix}" 13 14 team_id = None 15 project_id = None 16 17 18 @pytest.fixture(scope="module") 19 def client(): 20 with TestClient(app) as c: 21 yield c 22 23 24 def test_setup(client): 25 """Create team, project, and users for invite tests.""" 26 global team_id, project_id 27 auth = ("admin", RESTAI_DEFAULT_PASSWORD) 28 29 # Create team 30 resp = client.post("/teams", json={"name": team_name}, auth=auth) 31 assert resp.status_code in (200, 201) 32 team_id = resp.json()["id"] 33 34 # Create user in team 35 client.post("/users", json={"username": user_in_team, "password": "pass123", "admin": False, "private": False}, auth=auth) 36 client.post(f"/teams/{team_id}/users/{user_in_team}", auth=auth) 37 38 # Create user outside team 39 client.post("/users", json={"username": user_outside, "password": "pass123", "admin": False, "private": False}, auth=auth) 40 41 # Create project in team (block type needs no LLM) 42 resp = client.post("/projects", json={"name": project_name, "type": "block", "team_id": team_id}, auth=auth) 43 assert resp.status_code == 201 44 project_id = resp.json()["project"] 45 46 47 def test_send_invite_to_team_member(client): 48 """Sending invite to a user in the same team succeeds.""" 49 resp = client.post( 50 f"/projects/{project_id}/invitations", 51 json={"username": user_in_team}, 52 auth=("admin", RESTAI_DEFAULT_PASSWORD), 53 ) 54 assert resp.status_code == 200 55 assert "invitation" in resp.json()["message"].lower() or "user" in resp.json()["message"].lower() 56 57 58 def test_send_invite_to_outsider(client): 59 """Sending invite to user not in team returns same message (no info leak).""" 60 resp = client.post( 61 f"/projects/{project_id}/invitations", 62 json={"username": user_outside}, 63 auth=("admin", RESTAI_DEFAULT_PASSWORD), 64 ) 65 assert resp.status_code == 200 66 # Same response regardless 67 assert "message" in resp.json() 68 69 70 def test_send_invite_nonexistent_user(client): 71 """Sending invite to nonexistent user returns same message.""" 72 resp = client.post( 73 f"/projects/{project_id}/invitations", 74 json={"username": "definitely_not_a_user_xyz"}, 75 auth=("admin", RESTAI_DEFAULT_PASSWORD), 76 ) 77 assert resp.status_code == 200 78 assert "message" in resp.json() 79 80 81 def test_duplicate_invite_not_created(client): 82 """Sending the same invite twice should not create a duplicate.""" 83 auth = ("admin", RESTAI_DEFAULT_PASSWORD) 84 # First invite already sent in test_send_invite_to_team_member 85 # Send again 86 client.post(f"/projects/{project_id}/invitations", json={"username": user_in_team}, auth=auth) 87 88 # Check only 1 pending invite 89 resp = client.get("/invitations", auth=(user_in_team, "pass123")) 90 assert resp.status_code == 200 91 project_invites = [inv for inv in resp.json() if inv.get("type") == "project" and inv.get("project_id") == project_id] 92 assert len(project_invites) == 1 93 94 95 def test_invite_shows_in_invitations(client): 96 """Invited user sees the project invitation.""" 97 resp = client.get("/invitations", auth=(user_in_team, "pass123")) 98 assert resp.status_code == 200 99 project_invites = [inv for inv in resp.json() if inv.get("type") == "project"] 100 assert len(project_invites) >= 1 101 inv = project_invites[0] 102 assert inv["project_name"] == project_name 103 assert inv["invited_by"] == "admin" 104 105 106 def test_invitation_count_includes_projects(client): 107 """Invitation count includes project invitations.""" 108 resp = client.get("/invitations/count", auth=(user_in_team, "pass123")) 109 assert resp.status_code == 200 110 assert resp.json()["count"] >= 1 111 112 113 def test_accept_project_invite(client): 114 """Accepting a project invite adds the user to the project.""" 115 # Get the invitation 116 resp = client.get("/invitations", auth=(user_in_team, "pass123")) 117 project_invites = [inv for inv in resp.json() if inv.get("type") == "project" and inv.get("project_id") == project_id] 118 assert len(project_invites) >= 1 119 invite_id = project_invites[0]["id"] 120 121 # Accept 122 resp = client.post(f"/invitations/projects/{invite_id}/accept", json={}, auth=(user_in_team, "pass123")) 123 assert resp.status_code == 200 124 assert "Joined" in resp.json()["message"] 125 126 # Verify user has access 127 resp = client.get(f"/projects/{project_id}", auth=(user_in_team, "pass123")) 128 assert resp.status_code == 200 129 130 131 def test_decline_project_invite(client): 132 """Declining a project invite does not add the user.""" 133 auth_admin = ("admin", RESTAI_DEFAULT_PASSWORD) 134 135 # Create another user in team 136 decline_user = f"inv_decline_{suffix}" 137 client.post("/users", json={"username": decline_user, "password": "pass123", "admin": False, "private": False}, auth=auth_admin) 138 client.post(f"/teams/{team_id}/users/{decline_user}", auth=auth_admin) 139 140 # Send invite 141 client.post(f"/projects/{project_id}/invitations", json={"username": decline_user}, auth=auth_admin) 142 143 # Get invitation 144 resp = client.get("/invitations", auth=(decline_user, "pass123")) 145 project_invites = [inv for inv in resp.json() if inv.get("type") == "project" and inv.get("project_id") == project_id] 146 assert len(project_invites) >= 1 147 invite_id = project_invites[0]["id"] 148 149 # Decline 150 resp = client.post(f"/invitations/projects/{invite_id}/decline", json={}, auth=(decline_user, "pass123")) 151 assert resp.status_code == 200 152 153 # Verify user does NOT have access 154 resp = client.get(f"/projects/{project_id}", auth=(decline_user, "pass123")) 155 assert resp.status_code == 404 156 157 # Cleanup 158 client.delete(f"/users/{decline_user}", auth=auth_admin) 159 160 161 def test_non_member_cannot_send_invite(client): 162 """A user who is not a project member cannot send invites.""" 163 resp = client.post( 164 f"/projects/{project_id}/invitations", 165 json={"username": user_outside}, 166 auth=(user_outside, "pass123"), 167 ) 168 assert resp.status_code == 404 169 170 171 def test_cleanup(client): 172 """Clean up test resources.""" 173 auth = ("admin", RESTAI_DEFAULT_PASSWORD) 174 client.delete(f"/projects/{project_name}", auth=auth) 175 client.delete(f"/teams/{team_id}", auth=auth) 176 client.delete(f"/users/{user_in_team}", auth=auth) 177 client.delete(f"/users/{user_outside}", auth=auth)