@@ -51,8 +50,11 @@
{% trans %}Organisation Type{% endtrans %}
diff --git a/tests/api/routes/test_organizations.py b/tests/api/routes/test_organizations.py
index ee372521..5f80245d 100644
--- a/tests/api/routes/test_organizations.py
+++ b/tests/api/routes/test_organizations.py
@@ -1,9 +1,10 @@
import pytest
+from amt.models import Organization
from amt.schema.organization import OrganizationNew
from httpx import AsyncClient
from pytest_mock import MockFixture
-from tests.constants import default_auth_user, default_user
+from tests.constants import default_auth_user, default_user, default_user_without_default_organization
from tests.database_test_utils import DatabaseTestUtils
@@ -184,3 +185,77 @@ async def test_get_users(client: AsyncClient, mocker: MockFixture, db: DatabaseT
assert response.status_code == 200
assert response.headers["content-type"] == "application/json"
assert b"[]" in response.content
+
+
+@pytest.mark.asyncio
+async def test_get_members(client: AsyncClient, mocker: MockFixture, db: DatabaseTestUtils) -> None:
+ # given
+ await db.given([default_user()])
+ client.cookies["fastapi-csrf-token"] = "1"
+
+ mocker.patch("amt.api.routes.organizations.get_user", return_value=default_auth_user())
+ mocker.patch("fastapi_csrf_protect.CsrfProtect.validate_csrf", new_callable=mocker.AsyncMock)
+
+ # when
+ response = await client.get("/organizations/default-organization/members")
+ # then
+ assert response.status_code == 200
+ assert response.headers["content-type"] == "text/html; charset=utf-8"
+ assert b"/organizations/default-organization/members/92714be3-f798-4461-ba83-55d6cfd889a6" in response.content
+
+ # when
+ response = await client.get("/organizations/default-organization/members", headers={"HX-Request": "true"})
+ # then
+ assert response.status_code == 200
+ assert response.headers["content-type"] == "text/html; charset=utf-8"
+ assert b"/organizations/default-organization/members/92714be3-f798-4461-ba83-55d6cfd889a6" in response.content
+
+
+@pytest.mark.asyncio
+async def test_delete_member(client: AsyncClient, mocker: MockFixture, db: DatabaseTestUtils) -> None:
+ # given
+ await db.given([default_user()])
+
+ client.cookies["fastapi-csrf-token"] = "1"
+
+ mocker.patch("amt.api.routes.organizations.get_user", return_value=default_auth_user())
+ mocker.patch("fastapi_csrf_protect.CsrfProtect.validate_csrf", new_callable=mocker.AsyncMock)
+ # when
+ response = await client.delete(
+ f"/organizations/default-organization/members/{default_user().id!s}", headers={"X-CSRF-Token": "1"}
+ )
+
+ # then
+ assert response.status_code == 200
+ assert response.headers["content-type"] == "text/html; charset=utf-8"
+ assert response.headers["HX-Redirect"] == "/organizations/default-organization/members"
+
+
+@pytest.mark.asyncio
+async def test_add_member(client: AsyncClient, mocker: MockFixture, db: DatabaseTestUtils) -> None:
+ # given
+ await db.given(
+ [
+ default_user(),
+ default_user_without_default_organization(
+ id="910edc25-135c-4a86-9cfb-58935c47db90", name="Another user", organizations=None
+ ),
+ ]
+ )
+
+ client.cookies["fastapi-csrf-token"] = "1"
+
+ mocker.patch("amt.api.routes.organizations.get_user", return_value=default_auth_user())
+ mocker.patch("fastapi_csrf_protect.CsrfProtect.validate_csrf", new_callable=mocker.AsyncMock)
+
+ client.cookies["fastapi-csrf-token"] = "1"
+ response = await client.put(
+ "/organizations/default-organization/members",
+ json={"user_ids": ["910edc25-135c-4a86-9cfb-58935c47db90"]},
+ headers={"X-CSRF-Token": "1"},
+ )
+ updated_organization: Organization = (await db.get(Organization, "id", 1))[0]
+ assert len(updated_organization.users) == 2 # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
+ assert response.status_code == 200
+ assert response.headers["content-type"] == "text/html; charset=utf-8"
+ assert response.headers["HX-Redirect"] == "/organizations/default-organization/members"
diff --git a/tests/constants.py b/tests/constants.py
index 1025a9c7..894a486d 100644
--- a/tests/constants.py
+++ b/tests/constants.py
@@ -52,8 +52,8 @@ def default_algorithm(name: str = "default algorithm", organization_id: int = 1)
return Algorithm(name=name, organization_id=organization_id)
-def default_organization(name: str = "default organization") -> Organization:
- return Organization(name=name, slug="default-organization", created_by_id=UUID(default_auth_user()["sub"]))
+def default_organization(name: str = "default organization", slug: str = "default-organization") -> Organization:
+ return Organization(name=name, slug=slug, created_by_id=UUID(default_auth_user()["sub"]))
def default_user(
@@ -75,6 +75,25 @@ def default_user(
)
+def default_user_without_default_organization(
+ id: str | UUID | None = None,
+ name: str | None = None,
+ organizations: list[Organization] | None = None,
+) -> User:
+ user_name = name if name else default_auth_user()["name"]
+ organizations = [] if organizations is None else organizations
+ user_id = UUID(default_auth_user()["sub"]) if id is None else UUID(id) if isinstance(id, str) else id
+
+ return User(
+ id=user_id,
+ name=user_name,
+ email=default_auth_user()["email"],
+ name_encoded=quote_plus(user_name.strip().lower()),
+ email_hash=default_auth_user()["email_hash"],
+ organizations=organizations,
+ )
+
+
def default_algorithm_with_system_card(name: str = "default algorithm") -> Algorithm:
with open("resources/system_card_templates/AMT_Template_1.json") as f:
system_card_from_template = json.load(f)
diff --git a/tests/database_test_utils.py b/tests/database_test_utils.py
index 769c1f52..5a502cff 100644
--- a/tests/database_test_utils.py
+++ b/tests/database_test_utils.py
@@ -1,5 +1,6 @@
import logging
from pathlib import Path
+from typing import TypeVar
from amt.models.base import Base
from sqlalchemy import select
@@ -37,7 +38,9 @@ async def exists(self, model: type[Base], model_field: str, field_value: str | i
except AttributeError as err:
raise ValueError(f"Field '{model_field}' does not exist in model {model.__name__}") from err
- async def get(self, model: type[Base], model_field: str, field_value: str | int) -> list[Base]:
+ BaseType = TypeVar("BaseType", bound=Base)
+
+ async def get(self, model: type[BaseType], model_field: str, field_value: str | int) -> list[BaseType]:
try:
query = select(model).where(getattr(model, model_field) == field_value)
result = await self.get_session().execute(query)
diff --git a/tests/repositories/test_organizations.py b/tests/repositories/test_organizations.py
new file mode 100644
index 00000000..45cc0720
--- /dev/null
+++ b/tests/repositories/test_organizations.py
@@ -0,0 +1,90 @@
+import logging
+
+import pytest
+from amt.api.organization_filter_options import OrganizationFilterOptions
+from amt.models import Organization
+from amt.repositories.organizations import OrganizationsRepository
+from tests.constants import default_organization, default_user
+from tests.database_test_utils import DatabaseTestUtils
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio
+async def test_find_by(db: DatabaseTestUtils):
+ await db.given([default_user(), default_organization(name="ZZZZZ", slug="zzzzz")])
+ organization_repository = OrganizationsRepository(db.get_session())
+
+ organizations = await organization_repository.find_by(sort={"name": "ascending"})
+ assert organizations[0].name == "default organization"
+ assert organizations[1].name == "ZZZZZ"
+
+ organizations = await organization_repository.find_by(sort={"name": "descending"})
+ assert organizations[0].name == "ZZZZZ"
+ assert organizations[1].name == "default organization"
+
+ organizations = await organization_repository.find_by(
+ filters={"organization-type": OrganizationFilterOptions.MY_ORGANIZATIONS.value}, user_id=default_user().id
+ )
+ assert organizations[0].name == "default organization"
+
+ organizations = await organization_repository.find_by(search="default")
+ assert organizations[0].name == "default organization"
+
+ organizations = await organization_repository.find_by(search="no results")
+ assert len(organizations) == 0
+
+ organizations = await organization_repository.find_by(search="default", skip=10, limit=100)
+ assert len(organizations) == 0
+
+
+@pytest.mark.asyncio
+async def test_find_by_as_count(db: DatabaseTestUtils):
+ await db.given([default_user(), default_organization(name="ZZZZZ", slug="zzzzz")])
+ organization_repository = OrganizationsRepository(db.get_session())
+ organizations_count = await organization_repository.find_by_as_count(search="default")
+ assert organizations_count == 1
+
+
+@pytest.mark.asyncio
+async def test_save(db: DatabaseTestUtils):
+ organization_repository = OrganizationsRepository(db.get_session())
+ new_organization = Organization(name="New organization", slug="new-organization", created_by_id=default_user().id)
+ organization = await organization_repository.save(new_organization)
+ assert organization.id == 1
+
+
+@pytest.mark.asyncio
+async def test_find_by_slug(db: DatabaseTestUtils):
+ await db.given([default_user()])
+ organization_repository = OrganizationsRepository(db.get_session())
+ organization = await organization_repository.find_by_slug("default-organization")
+ assert organization.name == "default organization"
+
+
+@pytest.mark.asyncio
+async def test_find_by_id(db: DatabaseTestUtils):
+ await db.given([default_user()])
+ organization_repository = OrganizationsRepository(db.get_session())
+ organization = await organization_repository.find_by_id(1)
+ assert organization.name == "default organization"
+
+
+@pytest.mark.asyncio
+async def test_find_by_id_and_user_id(db: DatabaseTestUtils):
+ await db.given([default_user()])
+ organization_repository = OrganizationsRepository(db.get_session())
+ organization = await organization_repository.find_by_id_and_user_id(user_id=default_user().id, organization_id=1)
+ assert organization.name == "default organization"
+
+
+@pytest.mark.asyncio
+async def test_remove_user(db: DatabaseTestUtils):
+ await db.given([default_user()])
+ organization_repository = OrganizationsRepository(db.get_session())
+ organization = await organization_repository.find_by_id_and_user_id(user_id=default_user().id, organization_id=1)
+ organization_updated = await organization_repository.remove_user(
+ organization=organization,
+ user=organization.users[0], # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
+ )
+ assert len(organization_updated.users) == 0 # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]