django-postfix-dovecot-api/tests/test_aliases.py

import pytest

from dpdapi.models import Alias, Domain

FIELDS = {"id", "created", "updated", "source", "destination", "domain", "active"}


@pytest.fixture
def aliases(domain):
    other = Domain.objects.create(name="other.org")
    return [
        Alias.objects.create(source="ops-alert@example.com", destination="ninja@example.com", domain=domain),
        Alias.objects.create(source="yolo-announce@example.com", destination="ninja@example.com", domain=domain),
        Alias.objects.create(source="Ops@Other.org", destination="Boss@Other.org", domain=other),
    ]


def sources(res):
    assert res.status_code == 200
    return sorted(a["source"] for a in res.json())


def test_list(api, aliases):
    res = api.get("/api/aliases/")
    assert res.status_code == 200
    body = res.json()
    assert len(body) == 3
    assert set(body[0]) == FIELDS
    first = next(a for a in body if a["id"] == aliases[0].pk)
    assert first["source"] == "ops-alert@example.com"
    assert first["destination"] == "ninja@example.com"
    assert first["domain"] == aliases[0].domain_id
    assert first["active"] is True


def test_addresses_are_lowercased(aliases):
    assert aliases[2].source == "ops@other.org"
    assert aliases[2].destination == "boss@other.org"
    assert str(aliases[2]) == "ops@other.org=boss@other.org"


@pytest.mark.parametrize(
    ("params", "expected"),
    [
        ({"domain__name": "example.com"}, ["ops-alert@example.com", "yolo-announce@example.com"]),
        ({"domain__name": "nope.com"}, []),
        ({"source__regex": "^ops-|^ops@"}, ["ops-alert@example.com", "ops@other.org"]),
        ({"source__regex": "^OPS"}, []),
        ({"source__iregex": "^OPS"}, ["ops-alert@example.com", "ops@other.org"]),
        ({"destination__regex": "^boss"}, ["ops@other.org"]),
        ({"destination__iregex": "^NINJA"}, ["ops-alert@example.com", "yolo-announce@example.com"]),
        ({"domain__name": "example.com", "source__regex": "^yolo"}, ["yolo-announce@example.com"]),
    ],
)
def test_filters(api, aliases, params, expected):
    assert sources(api.get("/api/aliases/", params)) == expected


def test_filter_by_domain_id(api, aliases):
    res = api.get("/api/aliases/", {"domain": aliases[2].domain_id})
    assert sources(res) == ["ops@other.org"]


def test_create_bulk(api, domain):
    data = [
        {"source": "A@example.com", "destination": "b@example.com", "domain": domain.pk},
        {"source": "c@example.com", "destination": "d@example.com", "domain": domain.pk, "active": False},
    ]
    res = api.post("/api/aliases/create_bulk/", data, content_type="application/json")
    assert res.status_code == 201
    body = res.json()
    assert [(a["source"], a["destination"], a["active"]) for a in body] == [
        ("a@example.com", "b@example.com", True),
        ("c@example.com", "d@example.com", False),
    ]
    assert all(set(a) == FIELDS for a in body)
    assert Alias.objects.count() == 2


def test_create_bulk_accepts_existing(api, aliases):
    existing = aliases[0]
    data = [{"source": existing.source, "destination": existing.destination, "domain": existing.domain_id}]
    res = api.post("/api/aliases/create_bulk/", data, content_type="application/json")
    assert res.status_code == 201
    assert res.json()[0]["id"] == existing.pk
    assert Alias.objects.count() == 3


@pytest.mark.parametrize(
    "item",
    [
        {"source": "not-an-email", "destination": "b@example.com"},
        {"source": "a@example.com"},
        {"source": "a@example.com", "destination": "b@example.com", "domain": 999},
    ],
)
def test_create_bulk_invalid(api, domain, item):
    item.setdefault("domain", domain.pk)
    res = api.post("/api/aliases/create_bulk/", [item], content_type="application/json")
    assert res.status_code == 400
    assert Alias.objects.count() == 0


def test_create_bulk_requires_list(api, domain):
    data = {"source": "a@example.com", "destination": "b@example.com", "domain": domain.pk}
    res = api.post("/api/aliases/create_bulk/", data, content_type="application/json")
    assert res.status_code == 400


def test_delete_bulk(api, aliases):
    data = [{"source": a.source, "destination": a.destination, "domain": a.domain_id} for a in aliases[:2]]
    res = api.delete("/api/aliases/delete_bulk/", data, content_type="application/json")
    assert res.status_code == 204
    assert list(Alias.objects.all()) == [aliases[2]]


def test_delete_bulk_empty(api, aliases):
    res = api.delete("/api/aliases/delete_bulk/", [], content_type="application/json")
    assert res.status_code == 204
    assert Alias.objects.count() == 3


def test_delete_bulk_reports_missing(api, aliases):
    data = [
        {"source": aliases[0].source, "destination": aliases[0].destination, "domain": aliases[0].domain_id},
        {"source": "ghost@example.com", "destination": "ninja@example.com", "domain": aliases[0].domain_id},
    ]
    res = api.delete("/api/aliases/delete_bulk/", data, content_type="application/json")
    assert res.status_code == 200
    assert res.json() == {
        "result": "Aliases 'ghost@example.com=ninja@example.com' could not be found and were not deleted."
    }
    assert Alias.objects.count() == 2


def test_delete_bulk_only_matches_exact_alias(api, aliases):
    """Extra keys must not widen the match to other aliases."""
    data = [
        {
            "source": "ghost@example.com",
            "destination": "ninja@example.com",
            "domain": aliases[0].domain_id,
            "source__regex": ".*",
        }
    ]
    res = api.delete("/api/aliases/delete_bulk/", data, content_type="application/json")
    assert res.status_code == 200
    assert Alias.objects.count() == 3


def test_delete_bulk_invalid(api, aliases):
    res = api.delete("/api/aliases/delete_bulk/", [{"source": "x"}], content_type="application/json")
    assert res.status_code == 400
    assert Alias.objects.count() == 3


@pytest.mark.parametrize("method", ["post", "put", "patch"])
def test_alias_detail_routes_do_not_exist(api, aliases, method):
    res = getattr(api, method)("/api/aliases/", {}, content_type="application/json")
    assert res.status_code == 405