Bsr refacto admin bootstrap (#799)

* fix unit tests

* add integration tests

* Update unit coverage badge

---------

Co-authored-by: Benjamin PILIA <benjamin.pilia@protonmail.com>
Co-authored-by: benjaminpilia <benjaminpilia@users.noreply.github.com>

Author: benjaminpilia

.github/badges/coverage.json

@@ -1 +1 @@
-{"schemaVersion":1,"label":"coverage","message":"50.86%","color":"red"}
+{"schemaVersion":1,"label":"coverage","message":"50.94%","color":"red"}

api/infrastructure/fastapi/endpoints/admin/providers.py

@@ -66,15 +66,17 @@
     path=EndpointRoute.ADMIN_PROVIDERS,
     dependencies=[Security(dependency=get_current_key)],
     status_code=201,
-    responses=get_documentation_responses([
-        InconsistentModelMaxContextLengthHTTPException,
-        InconsistentModelVectorSizeHTTPException,
-        InvalidProviderTypeHTTPException,
-        ProviderNotReachableHTTPException,
-        ProviderAlreadyExistsHTTPException,
-        RouterNotFoundHTTPException,
-        NotAdminUserHTTPException,
-    ]),
+    responses=get_documentation_responses(
+        [
+            InconsistentModelMaxContextLengthHTTPException,
+            InconsistentModelVectorSizeHTTPException,
+            InvalidProviderTypeHTTPException,
+            ProviderNotReachableHTTPException,
+            ProviderAlreadyExistsHTTPException,
+            RouterNotFoundHTTPException,
+            NotAdminUserHTTPException,
+        ]
+    ),
 )
 async def create_provider(
     body: CreateProviderBody,
@@ -179,15 +181,17 @@ async def delete_provider(
     path=EndpointRoute.ADMIN_PROVIDERS + "/{provider_id}",
     dependencies=[Security(dependency=get_current_key)],
     status_code=201,
-    responses=get_documentation_responses([
-        InconsistentModelMaxContextLengthHTTPException,
-        InconsistentModelVectorSizeHTTPException,
-        InvalidProviderTypeHTTPException,
-        ProviderAlreadyExistsHTTPException,
-        RouterNotFoundHTTPException,
-        ProviderNotFoundHTTPException,
-        NotAdminUserHTTPException,
-    ]),
+    responses=get_documentation_responses(
+        [
+            InconsistentModelMaxContextLengthHTTPException,
+            InconsistentModelVectorSizeHTTPException,
+            InvalidProviderTypeHTTPException,
+            ProviderAlreadyExistsHTTPException,
+            RouterNotFoundHTTPException,
+            ProviderNotFoundHTTPException,
+            NotAdminUserHTTPException,
+        ]
+    ),
 )
 async def update_provider(
     provider_id: int = Path(description="The ID of the provider to update."),

api/infrastructure/fastapi/endpoints/admin/routers.py

@@ -52,11 +52,13 @@
     path=EndpointRoute.ADMIN_ROUTERS,
     dependencies=[Security(dependency=get_current_key)],
     status_code=201,
-    responses=get_documentation_responses([
-        RouterAliasAlreadyExistsHTTPException,
-        RouterAlreadyExistsHTTPException,
-        NotAdminUserHTTPException,
-    ]),
+    responses=get_documentation_responses(
+        [
+            RouterAliasAlreadyExistsHTTPException,
+            RouterAlreadyExistsHTTPException,
+            NotAdminUserHTTPException,
+        ]
+    ),
 )
 async def create_router(
     body: CreateRouterBody = Body(description="The router creation request."),
@@ -216,12 +218,14 @@ async def delete_router(
 @router.patch(
     path=EndpointRoute.ADMIN_ROUTERS + "/{router_id}",
     dependencies=[Security(dependency=get_current_key)],
-    responses=get_documentation_responses([
-        RouterNotFoundHTTPException,
-        NotAdminUserHTTPException,
-        RouterAliasAlreadyExistsHTTPException,
-        RouterAlreadyExistsHTTPException,
-    ]),
+    responses=get_documentation_responses(
+        [
+            RouterNotFoundHTTPException,
+            NotAdminUserHTTPException,
+            RouterAliasAlreadyExistsHTTPException,
+            RouterAlreadyExistsHTTPException,
+        ]
+    ),
     status_code=200,
 )
 async def update_router(

api/infrastructure/postgres/_postgresrolesrepository.py

@@ -58,7 +58,6 @@ async def get_roles(
         order_by: Literal["id", "name", "created", "updated"] = "id",
         order_direction: Literal["asc", "desc"] = "asc",
     ) -> list[Role]:
-
         if role_id is None:
             # get the unique role IDs with pagination
             statement = select(RoleTable.id).offset(offset=offset).limit(limit=limit).order_by(text(f"{order_by} {order_direction}"))  # nosemgrep

api/tests/integration/app/__init__.py

@@ -0,0 +1,70 @@
+import pytest
+from sqlalchemy import select
+
+from api.domain.role.entities import PermissionType
+from api.schemas.core.configuration import Configuration, Dependencies, Settings
+from api.sql.models import Permission, User
+from api.tests.integration.factories import RoleSQLFactory, UserSQLFactory
+from api.utils.lifespan import bootstrap_default_admin
+
+ADMIN_USERNAME = "admin"
+ADMIN_PASSWORD = "s3cr3t"
+
+
+@pytest.fixture
+def bootstrap_configuration() -> Configuration:
+    return Configuration.model_construct(
+        settings=Settings.model_construct(
+            auth_default_username=ADMIN_USERNAME,
+            auth_default_password=ADMIN_PASSWORD,
+        ),
+        dependencies=Dependencies.model_construct(sentry=None),
+    )
+
+
+@pytest.mark.asyncio(loop_scope="session")
+class TestBootstrapDefaultAdmin:
+    async def test_creates_admin_user_and_role_when_no_admin_exists(self, db_session, bootstrap_configuration):
+        # Act
+        await bootstrap_default_admin(configuration=bootstrap_configuration, postgres_session=db_session)
+        await db_session.flush()
+
+        # Assert
+        user = (await db_session.execute(select(User).where(User.email == ADMIN_USERNAME))).scalar_one_or_none()
+        assert user is not None
+        assert user.email == ADMIN_USERNAME
+
+        permission = (
+            await db_session.execute(select(Permission).where(Permission.role_id == user.role_id, Permission.permission == PermissionType.ADMIN))
+        ).scalar_one_or_none()
+        assert permission is not None
+
+    async def test_skips_when_admin_user_already_exists(self, db_session, bootstrap_configuration):
+        # Arrange
+        UserSQLFactory(admin_user=True)
+        await db_session.flush()
+
+        # Act
+        await bootstrap_default_admin(configuration=bootstrap_configuration, postgres_session=db_session)
+
+    async def test_raises_runtime_error_when_role_name_already_taken(self, db_session, bootstrap_configuration):
+        # Arrange
+        role = RoleSQLFactory(name=ADMIN_USERNAME)
+        await db_session.flush()
+
+        # Act & Assert
+        with pytest.raises(RuntimeError, match=f"'{role.name}' already exists"):
+            await bootstrap_default_admin(configuration=bootstrap_configuration, postgres_session=db_session)
+
+    async def test_raises_runtime_error_when_user_email_already_taken(self, db_session, bootstrap_configuration):
+        # Arrange
+        user = UserSQLFactory(regular_user=True, email=ADMIN_USERNAME)
+        await db_session.flush()
+
+        # Act & Assert
+        with pytest.raises(RuntimeError, match=f"'{user.email}' already exists"):
+            await bootstrap_default_admin(configuration=bootstrap_configuration, postgres_session=db_session)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

api/tests/integration/app/test_bootstrapadmin.py

@@ -267,8 +267,7 @@ async def test_update_provider_should_persist_all_updatable_fields(self, reposit
 
         # Act
         result = await repository.update_provider(
-            domain_provider
-            .with_router_id(router_2.id)
+            domain_provider.with_router_id(router_2.id)
             .with_timeout(120)
             .with_model_hosting_zone(ProviderCarbonFootprintZone.USA)
             .with_model_total_params(2_000_000)

api/tests/integration/test_createapp.py …/tests/integration/app/test_createapp.pyapi/tests/integration/test_createapp.py renamed to api/tests/integration/app/test_createapp.py api/tests/integration/test_createapp.py renamed to api/tests/integration/app/test_createapp.py

@@ -47,10 +47,12 @@ def test_get_carbon_footprint_return_footprint(self, mocker):
         mocked_electricity_mix = mocker.patch("api.utils.carbon.electricity_mixes.find_electricity_mix")
         mocked_electricity_mix.return_value = SimpleNamespace(adpe=1, pe=2, gwp=3, wue=4)
         mocked_compute_llm_impacts = mocker.patch("api.utils.carbon.compute_llm_impacts")
-        mocked_compute_llm_impacts.return_value = dict_to_namespace({
-            "energy": {"value": 1},
-            "gwp": {"value": 3},
-        })
+        mocked_compute_llm_impacts.return_value = dict_to_namespace(
+            {
+                "energy": {"value": 1},
+                "gwp": {"value": 3},
+            }
+        )
         active_params = 1
         total_params = 1
         model_zone = ProviderCarbonFootprintZone.WOR

api/tests/integration/postgres/test_postgresproviderrepository.py

@@ -24,58 +24,45 @@ def use_case(role_repository, user_info_repository):
     return CreateRoleUseCase(role_repository=role_repository, user_info_repository=user_info_repository)
 
 
-@pytest.mark.asyncio
-async def test_happy_path_returns_success_instance(use_case, role_repository, user_info_repository):
-    # Arrange
-    user_info_repository.get_user_info.return_value = UserInfoFactory(admin=True)
-    role_repository.create_role.return_value = RoleFactory()
-    command = CreateRoleCommand(user_id=1, name="new_role", permissions=[], limits=[])
-
-    # Act
-    result = await use_case.execute(command)
-
-    # Assert
-    assert isinstance(result, CreateRoleUseCaseSuccess)
-
-
-@pytest.mark.asyncio
-async def test_happy_path_result_contains_role_with_correct_name_and_permissions(use_case, role_repository, user_info_repository):
-    # Arrange
-    user_info_repository.get_user_info.return_value = UserInfoFactory(admin=True)
-    role_repository.create_role.return_value = RoleFactory(name="created_role", permissions=[PermissionType.READ_METRIC])
-    command = CreateRoleCommand(user_id=1, name="created_role", permissions=[PermissionType.READ_METRIC], limits=[])
-
-    # Act
-    result = await use_case.execute(command)
-
-    # Assert
-    assert result.role.name == "created_role"
-    assert PermissionType.READ_METRIC in result.role.permissions
-
-
-@pytest.mark.asyncio
-async def test_returns_user_is_not_admin_error_when_user_is_not_admin(use_case, user_info_repository):
-    # Arrange
-    user_info_repository.get_user_info.return_value = UserInfoFactory(without_permission=True, limits=[])
-    command = CreateRoleCommand(user_id=1, name="new_role", permissions=[], limits=[])
-
-    # Act
-    result = await use_case.execute(command)
-
-    # Assert
-    assert isinstance(result, UserIsNotAdminError)
-
-
-@pytest.mark.asyncio
-async def test_returns_role_already_exists_error_when_name_conflicts(use_case, role_repository, user_info_repository):
-    # Arrange
-    user_info_repository.get_user_info.return_value = UserInfoFactory(admin=True)
-    role_repository.create_role.return_value = RoleAlreadyExistsError(name="existing_role")
-    command = CreateRoleCommand(user_id=1, name="existing_role", permissions=[], limits=[])
-
-    # Act
-    result = await use_case.execute(command)
-
-    # Assert
-    assert isinstance(result, RoleAlreadyExistsError)
-    assert result.name == "existing_role"
+class TestCreateRoleUseCase:
+    @pytest.mark.asyncio
+    async def test_should_returns_success_instance(self, use_case, role_repository, user_info_repository):
+        # Arrange
+        user_info_repository.get_user_info.return_value = UserInfoFactory(admin=True)
+        role_repository.create_role.return_value = RoleFactory(name="created_role", permissions=[PermissionType.READ_METRIC])
+        command = CreateRoleCommand(user_id=1, name="created_role", permissions=[PermissionType.READ_METRIC], limits=[])
+
+        # Act
+        result = await use_case.execute(command)
+
+        # Assert
+        assert isinstance(result, CreateRoleUseCaseSuccess)
+
+        assert result.role.name == "created_role"
+        assert PermissionType.READ_METRIC in result.role.permissions
+
+    @pytest.mark.asyncio
+    async def test_should_return_user_is_not_admin_error_when_user_is_not_admin(self, use_case, user_info_repository):
+        # Arrange
+        user_info_repository.get_user_info.return_value = UserInfoFactory(without_permission=True, limits=[])
+        command = CreateRoleCommand(user_id=1, name="new_role", permissions=[], limits=[])
+
+        # Act
+        result = await use_case.execute(command)
+
+        # Assert
+        assert isinstance(result, UserIsNotAdminError)
+
+    @pytest.mark.asyncio
+    async def test_should_return_role_already_exists_error_when_name_conflicts(self, use_case, role_repository, user_info_repository):
+        # Arrange
+        user_info_repository.get_user_info.return_value = UserInfoFactory(admin=True)
+        role_repository.create_role.return_value = RoleAlreadyExistsError(name="existing_role")
+        command = CreateRoleCommand(user_id=1, name="existing_role", permissions=[], limits=[])
+
+        # Act
+        result = await use_case.execute(command)
+
+        # Assert
+        assert isinstance(result, RoleAlreadyExistsError)
+        assert result.name == "existing_role"