Delete user

src/routes/test_admin.py

@@ -149,6 +149,128 @@ def test__update_a_user(log, client, admin):
     assert response.status_code == fastapi.status.HTTP_200_OK
 
 
+def test_delete_user(db_api_session, log, client, admin):
+    """Check that we can delete a user and that it will remove all the
+    associated allows."""
+    auth=(admin["user"], admin["password"])
+
+    # FIXME fixture
+    # First, we create a normal user
+    response = client.post(
+        "/users/",
+        json={
+            "name": "normal",
+            "password": "toto",
+            "is_admin": False,
+        },
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_201_CREATED
+
+    # FIXME fixture
+    # We create a firt domain
+    response = client.post(
+        "/domains/",
+        json={
+            "name": "domaine-1",
+            "features": [ "mailbox" ],
+            "context_name": "useless",
+        },
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_201_CREATED
+
+    # FIXME fixture
+    # We create a second domain
+    response = client.post(
+        "/domains/",
+        json={
+            "name": "domaine-2",
+            "features": [ "mailbox" ],
+            "context_name": "useless",
+        },
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_201_CREATED
+
+    # FIXME fixture
+    # Then we allow the user on two domains
+    response = client.post(
+        "/allows/",
+        json={
+            "user": "normal",
+            "domain": "domaine-1",
+        },
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_201_CREATED
+
+    # FIXME fixture
+    # The second domain
+    response = client.post(
+        "/allows/",
+        json={
+            "user": "normal",
+            "domain": "domaine-2",
+        },
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_201_CREATED
+
+    # Check that we can fetch the two allows
+    response = client.get(
+        "/allows/",
+        params={
+            "username": "normal",
+        },
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_200_OK
+    items = response.json()
+    assert len(items) == 2
+    assert {"domain": "domaine-1", "user": "normal"} in items
+    assert {"domain": "domaine-2", "user": "normal"} in items
+
+    # If we try to delete a user without the prems -> forbiden
+    response = client.delete(
+        "/users/normal",
+        auth=("normal", "toto")
+    )
+    assert response.status_code == fastapi.status.HTTP_403_FORBIDDEN
+
+    # If we try to delete a user that does not exist -> not found
+    response = client.delete(
+        "/users/not-a-user",
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_404_NOT_FOUND
+
+    # If we delete the user -> ok, no content
+    response = client.delete(
+        "/users/normal",
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_204_NO_CONTENT
+
+    # And now, if we fetch the user -> not found
+    response = client.get(
+        "/users/normal",
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_404_NOT_FOUND
+
+    # And now, if we fetch the allows -> empty list
+    response = client.get(
+        "/allows",
+        params={
+            "username": "normal",
+        },
+        auth=auth,
+    )
+    assert response.status_code == fastapi.status.HTTP_200_OK
+    assert len(response.json()) == 0
+
+
 def test_domains__create_fails_no_name(db_api_session, log, client):
     """Cannot create domain with no name."""
 

src/routes/users/__init__.py

@@ -1,10 +1,12 @@
 # ruff: noqa: E402
+from .delete_user import delete_user
 from .get_user import get_user
 from .get_users import get_users
 from .patch_user import patch_user
 from .post_user import post_user
 
 __all__ = [
+    delete_user,
     get_user,
     get_users,
     patch_user,

src/routes/users/delete_user.py

@@ -0,0 +1,20 @@
+import fastapi
+
+from ... import auth, sql_api
+from .. import dependencies, routers
+
+
+@routers.users.delete("/{user_name}", status_code=fastapi.status.HTTP_204_NO_CONTENT)
+async def delete_user(
+    db: dependencies.DependsApiDb,
+    user: auth.DependsBasicAdmin,
+    user_name: str,
+) -> None:
+    user_db = sql_api.get_user(db, user_name)
+    if user_db is None:
+        raise fastapi.HTTPException(status_code=404, detail="Not found")
+
+    sql_api.delete_allows_by_user(db, user_name)
+    sql_api.delete_user(db, user_name)
+    return None
+

src/sql_api/__init__.py

@@ -1,4 +1,10 @@
-from .allow import allow_domain_for_user, deny_domain_for_user, get_allowed, get_allows
+from .allow import (
+    allow_domain_for_user,
+    delete_allows_by_user,
+    deny_domain_for_user,
+    get_allowed,
+    get_allows,
+)
 from .creds import Creds
 from .database import get_maker, init_db
 from .domain import create_domain, get_domain, get_domains
@@ -15,6 +21,7 @@
 
 __all__ = [
     allow_domain_for_user,
+    delete_allows_by_user,
     deny_domain_for_user,
     get_allowed,
     get_allows,

src/sql_api/allow.py

@@ -1,3 +1,4 @@
+import sqlalchemy as sa
 import sqlalchemy.orm as orm
 
 from . import models
@@ -35,3 +36,10 @@ def deny_domain_for_user(db: orm.Session, user: str, domain: str) -> models.DBAl
         db.delete(db_allowed)
         db.commit()
     return db_allowed
+
+
+def delete_allows_by_user(db: orm.Session, user: str) -> int:
+    """Delete all the allows for this user."""
+    res = db.execute(sa.delete(models.DBAllowed).where(models.DBAllowed.user == user))
+    db.commit()
+    return res.rowcount

src/sql_api/test_basic.py

@@ -154,6 +154,10 @@ def test_allows(db_api_session):
     assert allows[0].user == "toto"
     assert allows[0].domain == "example.net"
 
+    sql_api.delete_allows_by_user(db_api_session, user="toto")
+    allows = sql_api.get_allows(db_api_session, user="toto")
+    assert len(allows) == 0
+
 
 def test_creds(db_api_session):
     user_toto = sql_api.create_user(