Delete alias

src/routes/aliases/__init__.py

@@ -1,8 +1,10 @@
 # ruff: noqa: E402
 from .get_alias import get_alias
 from .post_alias import post_alias
+from .delete_alias import delete_alias
 
 __all__ = [
+    delete_alias,
     get_alias,
     post_alias,
 ]

src/routes/aliases/delete_alias.py

@@ -0,0 +1,44 @@
+import logging
+
+import fastapi
+
+from ... import auth, sql_postfix, web_models
+from .. import dependencies, routers
+
+
+@routers.aliases.delete(
+    "/{user_name}/{destination}",
+    description="Deletes an alias. "
+        "When destination=all, will remove all the destinations.",
+    status_code=204
+)
+async def delete_alias(
+    domain_name: str,
+    user_name: str,
+    destination: str,
+    user: auth.DependsTokenUser,
+    db: dependencies.DependsPostfixDb,
+) -> None:
+    log = logging.getLogger(__name__)
+
+    perms = user.get_creds()
+    if not perms.can_read(domain_name):
+        log.info(f"Cet utilisateur n'a pas les droits sur le domaine {domain_name}")
+        raise fastapi.HTTPException(status_code=403, detail="Permission denied")
+
+    name = user_name + "@" + domain_name
+    if destination == "all":
+        log.info(f"On demande la suppression de toutes les destinations pour {name}")
+        count = sql_postfix.delete_aliases_by_name(db, name)
+        if count == 0:
+            raise fastapi.HTTPException(status_code=404, detail="Not found")
+        return None
+
+    log.info("On supprime un alias exact")
+    count = sql_postfix.delete_alias(db, name, destination)
+    if count == 0:
+        log.info("Cet alias n'existe pas")
+        raise fastapi.HTTPException(status_code=404, detail="Not found")
+
+    return None
+

src/routes/test_alias.py

@@ -157,7 +157,7 @@ def test_alias__creates_and_fetch_an_alias(
     assert response.status_code == fastapi.status.HTTP_200_OK
     assert len(response.json()) == 3
 
-    # We feth the alias having 2 destinations and check the destinations
+    # We fetch the alias having 2 destinations and check the destinations
     # are correct
     response = client.get(
         f"/domains/{domain_name}/aliases/",
@@ -173,4 +173,40 @@ def test_alias__creates_and_fetch_an_alias(
         assert item["username"] == "from"
         assert item["destination"] in ["[email protected]", "[email protected]"]
 
+    # We remove a destination from an alias, first from an alias that
+    # does not exist
+    response = client.delete(
+        f"/domains/{domain_name}/aliases/pas-un-alias/destination",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert response.status_code == fastapi.status.HTTP_404_NOT_FOUND
+
+    # We remove all the destinations from an alias, for an alias that
+    # does not exist
+    response = client.delete(
+        f"/domains/{domain_name}/aliases/pas-un-alias/all",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert response.status_code == fastapi.status.HTTP_404_NOT_FOUND
+
+    # We remove a destination from an alias which exists
+    response = client.delete(
+        f"/domains/{domain_name}/aliases/old.chap/[email protected]",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert response.status_code == fastapi.status.HTTP_204_NO_CONTENT
+
+    # We remove all the destinations from an alias which exists
+    response = client.delete(
+        f"/domains/{domain_name}/aliases/from/all",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert response.status_code == fastapi.status.HTTP_204_NO_CONTENT
+
+    # We check that the virgin user cannot delete an alias
+    response = client.delete(
+        f"/domains/{domain_name}/aliases/from/all",
+        headers={"Authorization": f"Bearer {virgin_token}"},
+    )
+    assert response.status_code == fastapi.status.HTTP_403_FORBIDDEN
 

src/sql_postfix/__init__.py

@@ -1,9 +1,11 @@
-from .crud import create_alias, get_alias, get_aliases_by_domain, get_aliases_by_name
+from .crud import create_alias, delete_alias, delete_aliases_by_name, get_alias, get_aliases_by_domain, get_aliases_by_name
 from .database import get_maker, init_db
 from .models import PostfixAlias
 
 __all__ = [
     create_alias,
+    delete_alias,
+    delete_aliases_by_name,
     get_alias,
     get_aliases_by_domain,
     get_aliases_by_name,

src/sql_postfix/crud.py

@@ -1,3 +1,4 @@
+import sqlalchemy as sa
 import sqlalchemy.orm as orm
 
 from . import models
@@ -34,3 +35,18 @@ def create_alias(
         return None
     db.refresh(db_alias)
     return db_alias
+
+
+def delete_alias(db: orm.Session, alias: str, destination: str) -> int:
+    db_alias = get_alias(db, alias, destination)
+    if db_alias is not None:
+        db.delete(db_alias)
+        db.commit()
+        return 1
+    return 0
+
+
+def delete_aliases_by_name(db: orm.Session, name: str):
+    res = db.execute(sa.delete(models.PostfixAlias).where(models.PostfixAlias.alias == name))
+    db.commit()
+    return res.rowcount

src/sql_postfix/test_basic.py

@@ -20,3 +20,28 @@ def test_alias__create_and_get_an_alias(db_postfix_session):
     assert alias.alias == "[email protected]"
     assert alias.destination == "[email protected]"
     assert alias.domain == "example.com"
+
+    alias = sql_postfix.create_alias(
+        db_postfix_session, "example.com", "from", "[email protected]"
+    )
+    assert isinstance(alias, sql_postfix.PostfixAlias)
+
+    count = sql_postfix.delete_alias(
+        db_postfix_session, "[email protected]", "[email protected]"
+    )
+    assert count == 1
+
+    alias = sql_postfix.create_alias(
+        db_postfix_session, "example.com", "from", "[email protected]"
+    )
+    assert isinstance(alias, sql_postfix.PostfixAlias)
+
+    count = sql_postfix.delete_aliases_by_name(
+        db_postfix_session, "[email protected]"
+    )
+    assert count == 2
+
+    count = sql_postfix.delete_aliases_by_name(
+        db_postfix_session, "[email protected]"
+    )
+    assert count == 0