diff --git a/backend/app/alembic/versions/93d484f5798e_refactor_project_user.py b/backend/app/alembic/versions/93d484f5798e_refactor_project_user.py new file mode 100644 index 00000000..704217cd --- /dev/null +++ b/backend/app/alembic/versions/93d484f5798e_refactor_project_user.py @@ -0,0 +1,62 @@ +"""Refactor Project user + +Revision ID: 93d484f5798e +Revises: b30727137e65 +Create Date: 2025-10-10 17:55:46.327616 + +""" +from alembic import op +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision = "93d484f5798e" +down_revision = "b30727137e65" +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("projectuser") + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "projectuser", + sa.Column("id", sa.INTEGER(), autoincrement=True, nullable=False), + sa.Column("project_id", sa.INTEGER(), autoincrement=False, nullable=False), + sa.Column("is_admin", sa.BOOLEAN(), autoincrement=False, nullable=False), + sa.Column("is_deleted", sa.BOOLEAN(), autoincrement=False, nullable=False), + sa.Column( + "inserted_at", + postgresql.TIMESTAMP(), + server_default=sa.text("now()"), + autoincrement=False, + nullable=False, + ), + sa.Column( + "updated_at", postgresql.TIMESTAMP(), autoincrement=False, nullable=False + ), + sa.Column( + "deleted_at", postgresql.TIMESTAMP(), autoincrement=False, nullable=True + ), + sa.Column("user_id", sa.INTEGER(), autoincrement=False, nullable=False), + sa.ForeignKeyConstraint( + ["project_id"], + ["project.id"], + name="projectuser_project_id_fkey", + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["user_id"], + ["user.id"], + name="projectuser_user_id_fkey", + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name="projectuser_pkey"), + ) + # ### end Alembic commands ### diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index fd946e31..7729dfdd 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -20,7 +20,6 @@ User, UserProjectOrg, UserOrganization, - ProjectUser, Project, Organization, ) @@ -148,73 +147,3 @@ def get_current_active_superuser_org(current_user: CurrentUserOrg) -> User: status_code=403, detail="The user doesn't have enough privileges" ) return current_user - - -def verify_user_project_organization( - db: SessionDep, - current_user: CurrentUserOrg, - project_id: int, - organization_id: int, -) -> UserProjectOrg: - """ - Verify that the authenticated user is part of the project - and that the project belongs to the organization. - """ - if current_user.organization_id and current_user.organization_id != organization_id: - raise HTTPException(status_code=403, detail="User is not part of organization") - - project_organization = db.exec( - select(Project, Organization) - .join(Organization, Project.organization_id == Organization.id) - .where( - Project.id == project_id, - Project.is_active == True, - Organization.id == organization_id, - Organization.is_active == True, - ) - ).first() - - if not project_organization: - # Determine the exact error based on missing data - organization = db.exec( - select(Organization).where(Organization.id == organization_id) - ).first() - if not organization: - raise HTTPException(status_code=404, detail="Organization not found") - - if not organization.is_active: - raise HTTPException( - status_code=400, detail="Organization is not active" - ) # Use 400 for inactive resources - - project = db.exec(select(Project).where(Project.id == project_id)).first() - if not project: - raise HTTPException(status_code=404, detail="Project not found") - - if not project.is_active: - raise HTTPException( - status_code=400, detail="Project is not active" - ) # Use 400 for inactive resources - - raise HTTPException( - status_code=403, detail="Project does not belong to the organization" - ) - - # Superuser bypasses all checks and If Api key request we give access to all the project in organization - if current_user.is_superuser or current_user.organization_id: - current_user.organization_id = organization_id - return UserProjectOrg(**current_user.model_dump(), project_id=project_id) - - # Check if the user is part of the project - user_in_project = db.exec( - select(ProjectUser).where( - ProjectUser.user_id == current_user.id, - ProjectUser.project_id == project_id, - ) - ).first() - - if not user_in_project: - raise HTTPException(status_code=403, detail="User is not part of the project") - - current_user.organization_id = organization_id - return UserProjectOrg(**current_user.model_dump(), project_id=project_id) diff --git a/backend/app/api/main.py b/backend/app/api/main.py index b617e4b3..01892168 100644 --- a/backend/app/api/main.py +++ b/backend/app/api/main.py @@ -10,7 +10,6 @@ organization, openai_conversation, project, - project_user, responses, private, threads, @@ -37,7 +36,6 @@ api_router.include_router(openai_conversation.router) api_router.include_router(organization.router) api_router.include_router(project.router) -api_router.include_router(project_user.router) api_router.include_router(responses.router) api_router.include_router(threads.router) api_router.include_router(users.router) diff --git a/backend/app/api/routes/project_user.py b/backend/app/api/routes/project_user.py deleted file mode 100644 index 17bd5b8a..00000000 --- a/backend/app/api/routes/project_user.py +++ /dev/null @@ -1,115 +0,0 @@ -import uuid -from fastapi import APIRouter, Depends, HTTPException, Query, Request -from sqlmodel import Session -from typing import Annotated -from app.api.deps import get_db, verify_user_project_organization -from app.crud.project_user import ( - add_user_to_project, - remove_user_from_project, - get_users_by_project, - is_project_admin, -) -from app.models import User, ProjectUserPublic, UserProjectOrg, Message -from app.utils import APIResponse - - -router = APIRouter(prefix="/project/users", tags=["project_users"]) - - -# Add a user to a project -@router.post( - "/{user_id}", response_model=APIResponse[ProjectUserPublic], include_in_schema=False -) -def add_user( - request: Request, - user_id: int, - is_admin: bool = False, - session: Session = Depends(get_db), - current_user: UserProjectOrg = Depends(verify_user_project_organization), -): - """ - Add a user to a project. - """ - project_id = current_user.project_id - - user = session.get(User, user_id) - if not user: - raise HTTPException(status_code=404, detail="User not found") - - # Only allow superusers, project admins, or API key-authenticated requests to add users - if ( - not current_user.is_superuser - and not request.headers.get("X-API-KEY") - and not is_project_admin(session, current_user.id, project_id) - ): - raise HTTPException( - status_code=403, detail="Only project admins or superusers can add users." - ) - - try: - added_user = add_user_to_project(session, project_id, user_id, is_admin) - return APIResponse.success_response(added_user) - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - - -# Get all users in a project -@router.get( - "/", response_model=APIResponse[list[ProjectUserPublic]], include_in_schema=False -) -def list_project_users( - session: Session = Depends(get_db), - current_user: UserProjectOrg = Depends(verify_user_project_organization), - skip: int = Query(0, ge=0), - limit: int = Query(100, ge=1, le=100), -): - """ - Get all users in a project. - """ - users, total_count = get_users_by_project( - session, current_user.project_id, skip, limit - ) - - metadata = {"total_count": total_count, "limit": limit, "skip": skip} - - return APIResponse.success_response(data=users, metadata=metadata) - - -# Remove a user from a project -@router.delete( - "/{user_id}", response_model=APIResponse[Message], include_in_schema=False -) -def remove_user( - request: Request, - user_id: int, - session: Session = Depends(get_db), - current_user: UserProjectOrg = Depends(verify_user_project_organization), -): - """ - Remove a user from a project. - """ - # Only allow superusers or project admins to remove user - project_id = current_user.project_id - - user = session.get(User, user_id) - if not user: - raise HTTPException(status_code=404, detail="User not found") - - # Only allow superusers, project admins, or API key-authenticated requests to remove users - if ( - not current_user.is_superuser - and not request.headers.get("X-API-KEY") - and not is_project_admin(session, current_user.id, project_id) - ): - raise HTTPException( - status_code=403, - detail="Only project admins or superusers can remove users.", - ) - - try: - remove_user_from_project(session, project_id, user_id) - return APIResponse.success_response( - {"message": "User removed from project successfully."} - ) - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) diff --git a/backend/app/crud/project_user.py b/backend/app/crud/project_user.py deleted file mode 100644 index 3b5dedd1..00000000 --- a/backend/app/crud/project_user.py +++ /dev/null @@ -1,109 +0,0 @@ -import uuid -from sqlmodel import Session, select, delete, func -from app.models import ProjectUser, ProjectUserPublic, User, Project -from datetime import datetime, timezone - -from app.core.util import now - - -def is_project_admin(session: Session, user_id: int, project_id: int) -> bool: - """ - Checks if a user is an admin of the given project. - """ - project_user = session.exec( - select(ProjectUser).where( - ProjectUser.project_id == project_id, - ProjectUser.user_id == user_id, - ProjectUser.is_deleted == False, - ) - ).first() - - return bool(project_user and project_user.is_admin) - - -# Add a user to a project -def add_user_to_project( - session: Session, project_id: int, user_id: int, is_admin: bool = False -) -> ProjectUserPublic: - """ - Adds a user to a project. - """ - existing = session.exec( - select(ProjectUser).where( - ProjectUser.project_id == project_id, ProjectUser.user_id == user_id - ) - ).first() - - if existing: - raise ValueError("User is already a member of this project.") - - project_user = ProjectUser( - project_id=project_id, user_id=user_id, is_admin=is_admin - ) - session.add(project_user) - session.commit() - session.refresh(project_user) - - return ProjectUserPublic.model_validate(project_user) - - -def remove_user_from_project(session: Session, project_id: int, user_id: int) -> None: - """ - Removes a user from a project. - """ - project_user = session.exec( - select(ProjectUser).where( - ProjectUser.project_id == project_id, - ProjectUser.user_id == user_id, - ProjectUser.is_deleted == False, # Ignore already deleted users - ) - ).first() - if not project_user: - raise ValueError("User is not a member of this project or already removed.") - - project_user.is_deleted = True - project_user.deleted_at = now() - session.add(project_user) # Required to mark as dirty for commit - session.commit() - - -def get_users_by_project( - session: Session, project_id: int, skip: int = 0, limit: int = 100 -) -> tuple[list[ProjectUserPublic], int]: - """ - Returns paginated users in a given project along with the total count. - """ - count_statement = ( - select(func.count()) - .select_from(ProjectUser) - .where(ProjectUser.project_id == project_id, ProjectUser.is_deleted == False) - ) - total_count = session.exec(count_statement).one() - - statement = ( - select(ProjectUser) - .where(ProjectUser.project_id == project_id, ProjectUser.is_deleted == False) - .offset(skip) - .limit(limit) - ) - users = session.exec(statement).all() - - return [ProjectUserPublic.model_validate(user) for user in users], total_count - - -# Check if a user belongs to an at least one project in organization -def is_user_part_of_organization(session: Session, user_id: int, org_id: int) -> bool: - """ - Checks if a user is part of at least one project within the organization. - """ - user_in_org = session.exec( - select(ProjectUser) - .join(Project, ProjectUser.project_id == Project.id) - .where( - Project.organization_id == org_id, - ProjectUser.user_id == user_id, - ProjectUser.is_deleted == False, - ) - ).first() - - return bool(user_in_org) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 537532d0..5d50cf24 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -72,11 +72,6 @@ OrganizationUpdate, ) -from .project_user import ( - ProjectUser, - ProjectUserPublic, - ProjectUsersPublic, -) from .project import ( Project, ProjectCreate, diff --git a/backend/app/models/project.py b/backend/app/models/project.py index 8f0acfa6..00549b0c 100644 --- a/backend/app/models/project.py +++ b/backend/app/models/project.py @@ -39,9 +39,6 @@ class Project(ProjectBase, table=True): inserted_at: datetime = Field(default_factory=now, nullable=False) updated_at: datetime = Field(default_factory=now, nullable=False) - users: list["ProjectUser"] = Relationship( - back_populates="project", cascade_delete=True - ) creds: list["Credential"] = Relationship( back_populates="project", cascade_delete=True ) diff --git a/backend/app/models/project_user.py b/backend/app/models/project_user.py deleted file mode 100644 index 57028436..00000000 --- a/backend/app/models/project_user.py +++ /dev/null @@ -1,41 +0,0 @@ -import uuid -from datetime import datetime -from typing import Optional, List -from sqlmodel import SQLModel, Field, Relationship - -from app.core.util import now - - -# Shared properties -class ProjectUserBase(SQLModel): - project_id: int = Field( - foreign_key="project.id", nullable=False, ondelete="CASCADE" - ) - user_id: int = Field(foreign_key="user.id", nullable=False, ondelete="CASCADE") - is_admin: bool = Field( - default=False, nullable=False - ) # Determines if user is an admin of the project - - -class ProjectUserPublic(ProjectUserBase): - id: int - inserted_at: datetime - updated_at: datetime - - -# Database model, database table inferred from class name -class ProjectUser(ProjectUserBase, table=True): - id: int = Field(default=None, primary_key=True) - inserted_at: datetime = Field(default_factory=now, nullable=False) - updated_at: datetime = Field(default_factory=now, nullable=False) - is_deleted: bool = Field(default=False, nullable=False) - deleted_at: Optional[datetime] = Field(default=None, nullable=True) - # Relationships - project: "Project" = Relationship(back_populates="users") - user: "User" = Relationship(back_populates="projects") - - -# Properties to return as a list -class ProjectUsersPublic(SQLModel): - data: List[ProjectUserPublic] - count: int diff --git a/backend/app/models/user.py b/backend/app/models/user.py index 57336e72..6328efaf 100644 --- a/backend/app/models/user.py +++ b/backend/app/models/user.py @@ -49,9 +49,6 @@ class User(UserBase, table=True): id: int = Field(default=None, primary_key=True) hashed_password: str - projects: list["ProjectUser"] = Relationship( - back_populates="user", cascade_delete=True - ) api_keys: list["APIKey"] = Relationship(back_populates="user", cascade_delete=True) diff --git a/backend/app/tests/api/routes/test_project_user.py b/backend/app/tests/api/routes/test_project_user.py deleted file mode 100644 index 85b5241a..00000000 --- a/backend/app/tests/api/routes/test_project_user.py +++ /dev/null @@ -1,214 +0,0 @@ -import uuid -import pytest -from fastapi.testclient import TestClient -from sqlmodel import Session, select -from app.core.config import settings -from app.models import User, Project, ProjectUser, Organization -from app.crud.project_user import add_user_to_project -from app.tests.utils.utils import random_email, get_non_existent_id -from app.tests.utils.user import authentication_token_from_email -from app.core.security import get_password_hash -from app.main import app - -client = TestClient(app) - - -def create_user(db: Session) -> User: - """Helper function to create a user.""" - user = User(email=random_email(), hashed_password=get_password_hash("password123")) - db.add(user) - db.commit() - db.refresh(user) - return user - - -def create_organization_and_project(db: Session) -> tuple[Organization, Project]: - """Helper function to create an organization and a project.""" - - organization = Organization( - name=f"Test Organization {uuid.uuid4()}", is_active=True - ) - db.add(organization) - db.commit() - db.refresh(organization) - - # Ensure project with unique name - project_name = f"Test Project {uuid.uuid4()}" # Ensuring unique project name - project = Project( - name=project_name, - description="A test project", - organization_id=organization.id, - is_active=True, - ) - db.add(project) - db.commit() - db.refresh(project) - - return organization, project - - -def test_add_user_to_project( - client: TestClient, db: Session, superuser_token_headers: dict[str, str] -) -> None: - """ - Test adding a user to a project successfully. - """ - user = create_user(db) - organization, project = create_organization_and_project(db) - - response = client.post( - f"{settings.API_V1_STR}/project/users/{user.id}?is_admin=true&project_id={project.id}&organization_id={organization.id}", - headers=superuser_token_headers, - ) - - assert response.status_code == 200, response.text - added_user = response.json()["data"] - assert added_user["user_id"] == user.id - assert added_user["project_id"] == project.id - assert added_user["is_admin"] is True - - -def test_add_user_not_found( - client: TestClient, db: Session, superuser_token_headers: dict[str, str] -) -> None: - """ - Test adding a non-existing user to a project (should return 404). - """ - organization, project = create_organization_and_project(db) - user_id = get_non_existent_id(db, User) - - response = client.post( - f"{settings.API_V1_STR}/project/users/{user_id}?is_admin=false&project_id={project.id}&organization_id={organization.id}", - headers=superuser_token_headers, - ) - - assert response.status_code == 404 - assert response.json()["error"] == "User not found" - - -def test_add_existing_user_to_project( - client: TestClient, db: Session, superuser_token_headers: dict[str, str] -) -> None: - """ - Test adding a user who is already in the project (should return 400). - """ - user = create_user(db) - organization, project = create_organization_and_project(db) - - # Add user to project - project_user = ProjectUser(project_id=project.id, user_id=user.id, is_admin=False) - db.add(project_user) - db.commit() - - # Try to add the same user again - response = client.post( - f"{settings.API_V1_STR}/project/users/{user.id}?is_admin=false&project_id={project.id}&organization_id={organization.id}", - headers=superuser_token_headers, - ) - - assert response.status_code == 400 - assert "User is already a member of this project" in response.json()["error"] - - -def test_remove_user_from_project( - client: TestClient, db: Session, superuser_token_headers: dict[str, str] -) -> None: - """ - Test removing a user from a project successfully. - """ - # Create organization and project - organization, project = create_organization_and_project(db) - - # Create a user - user = create_user(db) - - # Add user to project - add_user_to_project(db, project.id, user.id, is_admin=False) - - # Remove user via API - response = client.delete( - f"{settings.API_V1_STR}/project/users/{user.id}?project_id={project.id}&organization_id={organization.id}", - headers=superuser_token_headers, - ) - - # Assertions - assert response.status_code == 200, response.text - assert response.json()["data"] == { - "message": "User removed from project successfully." - } - - # Ensure user is marked as deleted in the database (Fixed) - project_user = db.exec( - select(ProjectUser).where( - ProjectUser.project_id == project.id, - ProjectUser.user_id == user.id, - ) - ).first() - - assert project_user is not None - assert project_user.is_deleted is True - assert project_user.deleted_at is not None - - -def test_normal_user_cannot_add_user( - client: TestClient, db: Session, superuser_token_headers: dict[str, str] -) -> None: - """ - Test that a normal user (not admin) cannot add a user to a project. - """ - - organization, project = create_organization_and_project(db) - - normal_user_email = random_email() - normal_user_token_headers = authentication_token_from_email( - client=client, email=normal_user_email, db=db - ) - - normal_user = db.exec(select(User).where(User.email == normal_user_email)).first() - add_user_to_project(db, project.id, normal_user.id, is_admin=False) - - target_user = create_user(db) - - # Normal user attempts to add target user to the project - response = client.post( - f"{settings.API_V1_STR}/project/users/{target_user.id}?is_admin=false&project_id={project.id}&organization_id={organization.id}", - headers=normal_user_token_headers, - ) - - assert response.status_code == 403 - assert ( - response.json()["error"] == "Only project admins or superusers can add users." - ) - - -def test_normal_user_cannot_remove_user( - client: TestClient, db: Session, superuser_token_headers: dict[str, str] -) -> None: - """ - Test that a normal user (not admin) cannot remove a user from a project. - """ - organization, project = create_organization_and_project(db) - - normal_user_email = random_email() - normal_user_token_headers = authentication_token_from_email( - client=client, email=normal_user_email, db=db - ) - - normal_user = db.exec(select(User).where(User.email == normal_user_email)).first() - add_user_to_project(db, project.id, normal_user.id, is_admin=False) - - target_user = create_user(db) - add_user_to_project(db, project.id, target_user.id, is_admin=False) - - # Normal user attempts to remove the target user - response = client.delete( - f"{settings.API_V1_STR}/project/users/{target_user.id}?project_id={project.id}&organization_id={organization.id}", - headers=normal_user_token_headers, - ) - - # Assertions - assert response.status_code == 403 - assert ( - response.json()["error"] - == "Only project admins or superusers can remove users." - ) diff --git a/backend/app/tests/api/test_deps.py b/backend/app/tests/api/test_deps.py deleted file mode 100644 index 64280ebb..00000000 --- a/backend/app/tests/api/test_deps.py +++ /dev/null @@ -1,161 +0,0 @@ -import pytest -import uuid -from sqlmodel import Session, select -from fastapi import HTTPException -from app.api.deps import verify_user_project_organization -from app.models import ( - User, - Organization, - Project, - ProjectUser, - UserProjectOrg, - UserOrganization, -) -from app.tests.utils.utils import random_email -from app.core.security import get_password_hash - - -def create_org_project( - db: Session, org_active=True, proj_active=True -) -> tuple[Organization, Project]: - """Helper function to create an organization and a project with customizable active states.""" - org = Organization(name=f"Test Org {uuid.uuid4()}", is_active=org_active) - db.add(org) - db.commit() - db.refresh(org) - - proj = Project( - name=f"Test Proj {uuid.uuid4()}", - description="A test project", - organization_id=org.id, - is_active=proj_active, - ) - db.add(proj) - db.commit() - db.refresh(proj) - - return org, proj - - -def create_user(db: Session, is_superuser=False) -> User: - """Helper function to create a user.""" - user = User( - email=random_email(), - hashed_password=get_password_hash("password123"), - is_superuser=is_superuser, - ) - db.add(user) - db.commit() - db.refresh(user) - user_org = UserOrganization(**user.model_dump(), organization_id=None) - return user_org - - -def test_verify_success(db: Session): - """Valid user in a project passes verification.""" - user = create_user(db) - org, proj = create_org_project(db) - - db.add(ProjectUser(project_id=proj.id, user_id=user.id, is_admin=False)) - db.commit() - - result = verify_user_project_organization(db, user, proj.id, org.id) - - assert isinstance(result, UserProjectOrg) - assert result.project_id == proj.id - assert result.organization_id == org.id - - -def test_verify_superuser_bypass(db: Session): - """Superuser bypasses project membership check.""" - superuser = create_user(db, is_superuser=True) - org, proj = create_org_project(db) - - result = verify_user_project_organization(db, superuser, proj.id, org.id) - - assert isinstance(result, UserProjectOrg) - assert result.project_id == proj.id - assert result.organization_id == org.id - - -def test_verify_no_org(db: Session): - """Missing organization results in a 404 error.""" - user = create_user(db) - invalid_org_id = 9999 - - assert ( - db.exec(select(Organization).where(Organization.id == invalid_org_id)).first() - is None - ) - - with pytest.raises(HTTPException) as exc_info: - verify_user_project_organization( - db, user, project_id=1, organization_id=invalid_org_id - ) - - assert exc_info.value.status_code == 404 - assert exc_info.value.detail == "Organization not found" - - -def test_verify_no_project(db: Session): - """Missing project results in a 404 error.""" - user = create_user(db) - org = Organization(name=f"Test Org {uuid.uuid4()}", is_active=True) - db.add(org) - db.commit() - db.refresh(org) - - with pytest.raises(HTTPException) as exc_info: - verify_user_project_organization(db, user, 9999, org.id) - - assert exc_info.value.status_code == 404 - assert exc_info.value.detail == "Project not found" - - -def test_verify_project_not_in_org(db: Session): - """Project not belonging to organization results in a 403 error.""" - user = create_user(db) - org1, proj1 = create_org_project(db) - org2, proj2 = create_org_project(db) - - with pytest.raises(HTTPException) as exc_info: - verify_user_project_organization(db, user, proj2.id, org1.id) - - assert exc_info.value.status_code == 403 - assert exc_info.value.detail == "Project does not belong to the organization" - - -def test_verify_user_not_in_project(db: Session): - """User not in project results in a 403 error.""" - user = create_user(db) - org, proj = create_org_project(db) - - with pytest.raises(HTTPException) as exc_info: - verify_user_project_organization(db, user, proj.id, org.id) - - assert exc_info.value.status_code == 403 - assert exc_info.value.detail == "User is not part of the project" - - -def test_verify_inactive_organization(db: Session): - """Inactive organization results in a 400 error.""" - user = create_user(db) - org, proj = create_org_project(db, org_active=False) - - with pytest.raises(HTTPException) as exc_info: - verify_user_project_organization(db, user, proj.id, org.id) - - assert exc_info.value.status_code == 400 - assert exc_info.value.detail == "Organization is not active" - - -def test_verify_inactive_project(db: Session): - """Inactive project results in a 400 error.""" - user = create_user(db) - org, proj = create_org_project(db, proj_active=False) - - with pytest.raises(HTTPException) as exc_info: - verify_user_project_organization(db, user, proj.id, org.id) - - assert exc_info.value.status_code == 400 - assert exc_info.value.detail == "Project is not active" diff --git a/backend/app/tests/crud/test_project_user.py b/backend/app/tests/crud/test_project_user.py deleted file mode 100644 index 72025ae6..00000000 --- a/backend/app/tests/crud/test_project_user.py +++ /dev/null @@ -1,141 +0,0 @@ -import uuid -from sqlmodel import Session, select -from datetime import datetime -import pytest - -from app.crud import project_user as project_user_crud -from app.models import ProjectUser, ProjectUserPublic, User, Project, Organization -from app.tests.utils.utils import random_email, get_non_existent_id -from app.core.security import get_password_hash - - -def create_organization_and_project(db: Session) -> tuple[Organization, Project]: - """Helper function to create an organization and a project.""" - - organization = Organization( - name=f"Test Organization {uuid.uuid4()}", is_active=True - ) - db.add(organization) - db.commit() - db.refresh(organization) - - # Ensure project with unique name - project_name = f"Test Project {uuid.uuid4()}" # Ensuring unique project name - project = Project( - name=project_name, - description="A test project", - organization_id=organization.id, - is_active=True, - ) - db.add(project) - db.commit() - db.refresh(project) - - return organization, project - - -def test_is_project_admin(db: Session) -> None: - organization, project = create_organization_and_project(db) - - user = User(email=random_email(), hashed_password=get_password_hash("password123")) - db.add(user) - db.commit() - db.refresh(user) - - project_user = ProjectUser(project_id=project.id, user_id=user.id, is_admin=True) - db.add(project_user) - db.commit() - db.refresh(project_user) - - assert project_user_crud.is_project_admin(db, user.id, project.id) is True - - -def test_add_user_to_project(db: Session) -> None: - organization, project = create_organization_and_project(db) - - user = User(email=random_email(), hashed_password=get_password_hash("password123")) - db.add(user) - db.commit() - db.refresh(user) - - project_user = project_user_crud.add_user_to_project( - db, project.id, user.id, is_admin=True - ) - - assert project_user.user_id == user.id - assert project_user.project_id == project.id - assert project_user.is_admin is True - - -def test_add_user_to_project_duplicate(db: Session) -> None: - organization, project = create_organization_and_project(db) - - user = User(email=random_email(), hashed_password=get_password_hash("password123")) - db.add(user) - db.commit() - db.refresh(user) - - project_user_crud.add_user_to_project(db, project.id, user.id) - - with pytest.raises(ValueError, match="User is already a member of this project"): - project_user_crud.add_user_to_project(db, project.id, user.id) - - -def test_remove_user_from_project(db: Session) -> None: - organization, project = create_organization_and_project(db) - - user = User(email=random_email(), hashed_password=get_password_hash("password123")) - db.add(user) - db.commit() - db.refresh(user) - - # Add user to project - project_user_crud.add_user_to_project(db, project.id, user.id) - - # Remove user from project - project_user_crud.remove_user_from_project(db, project.id, user.id) - - # Retrieve project user with both project_id and user_id - project_user = db.exec( - select(ProjectUser).where( - ProjectUser.project_id == project.id, ProjectUser.user_id == user.id - ) - ).first() - - assert project_user is not None # Ensure the record still exists (soft delete) - assert project_user.is_deleted is True - assert project_user.deleted_at is not None - - -def test_remove_user_from_project_not_member(db: Session) -> None: - organization, project = create_organization_and_project(db) - - project_id = project.id - user_id = get_non_existent_id(db, User) - - with pytest.raises( - ValueError, match="User is not a member of this project or already removed" - ): - project_user_crud.remove_user_from_project(db, project_id, user_id) - - -def test_get_users_by_project(db: Session) -> None: - organization, project = create_organization_and_project(db) - - user1 = User(email=random_email(), hashed_password=get_password_hash("password123")) - user2 = User(email=random_email(), hashed_password=get_password_hash("password123")) - - db.add_all([user1, user2]) - db.commit() - db.refresh(user1) - db.refresh(user2) - - project_user_crud.add_user_to_project(db, project.id, user1.id) - project_user_crud.add_user_to_project(db, project.id, user2.id) - - users, total_count = project_user_crud.get_users_by_project( - db, project.id, skip=0, limit=10 - ) - - assert total_count == 2 - assert len(users) == 2