test: cover api behaviour with pytest at 100%

This commit is contained in:
ponas committed 2026-11-03 03:52:41 +01:00
1 parent 08ffd39674
commit cad499c19a
12 files changed
+644 -43

No files matched your search

-43
View File
@@ -1,43 +0,0 @@
import json
from django.contrib.auth.models import User
from django.urls import reverse
from rest_framework.test import APITestCase
from dpdapi.models import Domain, Alias
class RegexTestCase(APITestCase):
aliases = ['ops-alert@example.com', 'yolo-announce@example.com']
def setUp(self):
user = User.objects.create(username='yolo')
self.domain = Domain.objects.create(name='example.com')
for a in self.aliases:
Alias.objects.create(source=a, destination='ninja@example.com', domain=self.domain)
self.client.force_login(user)
def test_get_all(self):
res = self.client.get(reverse('alias-list'))
self.assertEqual(len(res.data), len(self.aliases))
def test_domain_filter(self):
params = {'domain__name': 'example.com'}
res = self.client.get(reverse('alias-list'), data=params)
self.assertEqual(len(res.data), len(self.aliases))
def test_domain_filter_no_hits(self):
params = {'domain__name': 'zexample.com'}
res = self.client.get(reverse('alias-list'), data=params)
self.assertEqual(len(res.data), 0)
def test_regex_filter(self):
params = {'source__regex': '^ops-|^ops@'}
res = self.client.get(reverse('alias-list'), data=params)
self.assertEqual(len(res.data), 1)
def test_delete_bulk_single(self):
data = {'destination': 'ninja@example.com', 'source': 'ding@example.com', 'domain': self.domain.pk}
Alias.objects.create(source=data['source'], destination=data['destination'], domain=self.domain)
res = self.client.delete(reverse('alias-delete-bulk'), data=[data], format='json')
self.assertEqual(res.status_code, 204, res.data)
View File
Whitespace-only changes.
+34
View File
@@ -0,0 +1,34 @@
import base64
import pytest
@pytest.fixture(autouse=True)
def static_files(settings, tmp_path):
"""The manifest storage needs collectstatic, which tests don't run."""
settings.STATIC_ROOT = tmp_path / "static"
settings.STATIC_ROOT.mkdir()
settings.STORAGES = {
**settings.STORAGES,
"staticfiles": {"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"},
}
@pytest.fixture
def admin(django_user_model):
return django_user_model.objects.create_superuser("admin", "admin@example.com", "secret")
@pytest.fixture
def api(client, admin):
"""A Django test client authenticated with HTTP Basic auth."""
token = base64.b64encode(b"admin:secret").decode()
client.defaults["HTTP_AUTHORIZATION"] = f"Basic {token}"
return client
@pytest.fixture
def domain(db):
from dpdapi.models import Domain
return Domain.objects.create(name="example.com")
+6
View File
@@ -0,0 +1,6 @@
"""Loaded with `-p` so it runs before pytest-django imports the settings."""
import os
# Never report test runs to the production Sentry project.
os.environ["SENTRY_DSN"] = ""
+160
View File
@@ -0,0 +1,160 @@
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
+48
View File
@@ -0,0 +1,48 @@
import pytest
ENDPOINTS = [
("get", "/api/aliases/"),
("post", "/api/aliases/create_bulk/"),
("delete", "/api/aliases/delete_bulk/"),
("get", "/api/domains/"),
("post", "/api/domains/"),
("get", "/api/domains/1/"),
("put", "/api/domains/1/"),
("patch", "/api/domains/1/"),
("delete", "/api/domains/1/"),
("get", "/api/users/"),
("post", "/api/users/"),
("get", "/api/users/1/"),
("put", "/api/users/1/"),
("patch", "/api/users/1/"),
("delete", "/api/users/1/"),
]
@pytest.mark.django_db
@pytest.mark.parametrize(("method", "url"), ENDPOINTS)
def test_anonymous_is_rejected(client, method, url):
res = getattr(client, method)(url, content_type="application/json")
assert res.status_code == 401
@pytest.mark.django_db
def test_wrong_password_is_rejected(client, admin):
client.defaults["HTTP_AUTHORIZATION"] = "Basic YWRtaW46d3Jvbmc=" # admin:wrong
assert client.get("/api/domains/").status_code == 401
def test_basic_auth(api):
assert api.get("/api/domains/").status_code == 200
def test_session_auth(client, admin):
client.force_login(admin)
assert client.get("/api/domains/").status_code == 200
def test_root_redirects_to_api(client, admin):
client.force_login(admin)
res = client.get("/", follow=True)
assert res.redirect_chain
assert res.status_code == 200
+98
View File
@@ -0,0 +1,98 @@
import pytest
from django.utils import timezone
from dpdapi.models import Domain
FIELDS = {"id", "created", "updated", "name"}
def test_list(api, domain):
Domain.objects.create(name="other.org")
res = api.get("/api/domains/")
assert res.status_code == 200
assert sorted(d["name"] for d in res.json()) == ["example.com", "other.org"]
assert set(res.json()[0]) == FIELDS
def test_filter_by_name(api, domain):
Domain.objects.create(name="other.org")
res = api.get("/api/domains/", {"name": "other.org"})
assert [d["name"] for d in res.json()] == ["other.org"]
def test_retrieve(api, domain):
res = api.get(f"/api/domains/{domain.pk}/")
assert res.status_code == 200
assert res.json()["name"] == "example.com"
assert res.json()["id"] == domain.pk
def test_timestamps_are_local_iso8601(api, domain):
body = api.get(f"/api/domains/{domain.pk}/").json()
assert body["created"] == timezone.localtime(domain.created).isoformat()
assert body["updated"] == timezone.localtime(domain.updated).isoformat()
def test_retrieve_missing(api, db):
assert api.get("/api/domains/999/").status_code == 404
def test_create(api, db):
res = api.post("/api/domains/", {"name": "New.ORG"}, content_type="application/json")
assert res.status_code == 201
assert res.json()["name"] == "new.org"
assert set(res.json()) == FIELDS
assert str(Domain.objects.get()) == "new.org"
@pytest.mark.parametrize(
"data", [{}, {"name": ""}, {"name": "example.com"}, {"name": "Example.COM"}, {"name": "x" * 255}]
)
def test_create_invalid(api, domain, data):
res = api.post("/api/domains/", data, content_type="application/json")
assert res.status_code == 400
assert Domain.objects.count() == 1
def test_update(api, domain):
res = api.put(f"/api/domains/{domain.pk}/", {"name": "Renamed.com"}, content_type="application/json")
assert res.status_code == 200
assert res.json()["name"] == "renamed.com"
domain.refresh_from_db()
assert domain.name == "renamed.com"
def test_update_invalid(api, domain):
res = api.put(f"/api/domains/{domain.pk}/", {}, content_type="application/json")
assert res.status_code == 400
def test_partial_update(api, domain):
res = api.patch(f"/api/domains/{domain.pk}/", {"name": "patched.com"}, content_type="application/json")
assert res.status_code == 200
assert res.json()["name"] == "patched.com"
def test_partial_update_empty(api, domain):
res = api.patch(f"/api/domains/{domain.pk}/", {}, content_type="application/json")
assert res.status_code == 200
assert res.json()["name"] == "example.com"
def test_partial_update_null(api, domain):
res = api.patch(f"/api/domains/{domain.pk}/", {"name": None}, content_type="application/json")
assert res.status_code == 400
def test_update_missing(api, db):
res = api.put("/api/domains/999/", {"name": "x.com"}, content_type="application/json")
assert res.status_code == 404
def test_delete(api, domain):
assert api.delete(f"/api/domains/{domain.pk}/").status_code == 204
assert not Domain.objects.exists()
def test_delete_missing(api, db):
assert api.delete("/api/domains/999/").status_code == 404
+61
View File
@@ -0,0 +1,61 @@
from io import StringIO
import pytest
from django.core.management import call_command
from dpdapi.models import Alias, Domain
def run(*args):
out = StringIO()
call_command("import_valiases", *args, stdout=out)
return out.getvalue()
@pytest.mark.django_db
def test_missing_argument():
assert run() == "Missing argument\n"
@pytest.mark.django_db
def test_missing_file(tmp_path):
path = tmp_path / "nope"
assert run(str(path)) == f"{path} does not exist\n"
@pytest.mark.django_db
def test_no_aliases(tmp_path):
path = tmp_path / "valiases"
path.write_text('*: catchall@example.com\nlist: "|/usr/lib/mailman/mail/mailman post list"\n')
assert run(str(path)) == "No aliases found\n"
assert not Domain.objects.exists()
@pytest.mark.django_db
def test_import(tmp_path):
path = tmp_path / "valiases"
path.write_text(
"*: catchall@example.com\n"
"ops@example.com: a@example.com,b@example.com\n"
'list@example.com: "|/usr/lib/mailman/mail/mailman post list",list-admin@example.com,c@example.com\n'
"ops@example.com: a@example.com\n"
)
out = run(str(path))
assert out == (
"Created domain example.com.\n"
"Could not import 'ops@example.com=a@example.com', already exists.\n"
"Imported 3, skipped 1 existing, total 4\n"
)
assert sorted(Alias.objects.values_list("source", "destination")) == [
("list@example.com", "c@example.com"),
("ops@example.com", "a@example.com"),
("ops@example.com", "b@example.com"),
]
@pytest.mark.django_db
def test_import_into_existing_domain(tmp_path):
Domain.objects.create(name="example.com")
path = tmp_path / "valiases"
path.write_text("ops@example.com: a@example.com\n")
assert run(str(path)) == "Imported 1, skipped 0 existing, total 1\n"
+83
View File
@@ -0,0 +1,83 @@
"""The exact calls mailadmin (https://git.neuf.no/edb/mailadmin, see mailadmin/api.py) makes, and what it reads."""
import pytest
from dpdapi.models import Alias, Domain
# mailadmin's PrefixQueryset.as_regex() for the prefixes "ops" and "yolo"
PREFIX_REGEX = "^ops-|^ops@|^yolo-|^yolo@"
@pytest.fixture
def aliases(domain):
other = Domain.objects.create(name="other.org")
Alias.objects.create(source="ops-alert@example.com", destination="ninja@example.com", domain=domain)
Alias.objects.create(source="yolo@example.com", destination="ninja@example.com", domain=domain)
Alias.objects.create(source="nope-list@example.com", destination="ninja@example.com", domain=domain)
Alias.objects.create(source="ops@other.org", destination="ninja@other.org", domain=other)
def test_list_domains(api, domain):
res = api.get("/api/domains/", {"name": "example.com"})
assert res.status_code == 200
assert [(d["id"], d["name"]) for d in res.json()] == [(domain.pk, "example.com")]
def test_list_aliases_regex(api, aliases):
res = api.get("/api/aliases/", {"domain__name": "example.com", "source__iregex": PREFIX_REGEX})
assert res.status_code == 200
assert sorted(a["source"] for a in res.json()) == ["ops-alert@example.com", "yolo@example.com"]
assert all({"id", "source", "destination"} <= set(a) for a in res.json())
def test_create_aliases(api, domain):
data = [
{"source": "ops-new@example.com", "destination": "a@example.com", "domain": domain.pk},
{"source": "ops-new@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 == 201
body = res.json()
assert [(a["source"], a["destination"]) for a in body] == [
("ops-new@example.com", "a@example.com"),
("ops-new@example.com", "b@example.com"),
]
assert [a["id"] for a in body] == list(Alias.objects.order_by("pk").values_list("pk", flat=True))
def test_create_aliases_case_insensitive_duplicate(api, aliases, domain):
data = [{"source": "OPS-alert@example.com", "destination": "Ninja@example.com", "domain": domain.pk}]
res = api.post("/api/aliases/create_bulk/", data, content_type="application/json")
assert res.status_code == 201
assert Alias.objects.count() == 4
def test_delete_aliases(api, aliases, domain):
data = [{"source": "ops-alert@example.com", "destination": "ninja@example.com", "domain": domain.pk}]
res = api.delete("/api/aliases/delete_bulk/", data, content_type="application/json")
assert res.status_code == 204
assert res.content == b"" # mailadmin turns a non-JSON body into None
def test_delete_aliases_partially_missing(api, aliases, domain):
data = [
{"source": "ops-alert@example.com", "destination": "ninja@example.com", "domain": domain.pk},
{"source": "ops-alert@example.com", "destination": "ghost@example.com", "domain": domain.pk},
]
res = api.delete("/api/aliases/delete_bulk/", data, content_type="application/json")
assert res.status_code == 200
assert "ops-alert@example.com=ghost@example.com" in res.json()["result"]
def test_errors_are_json(api, domain):
data = [{"source": "bad", "destination": "a@example.com", "domain": domain.pk}]
res = api.post("/api/aliases/create_bulk/", data, content_type="application/json")
assert res.status_code == 400
assert res.json()
def test_bad_credentials_are_json(client, db):
client.defaults["HTTP_AUTHORIZATION"] = "Basic YWRtaW46d3Jvbmc="
res = client.get("/api/domains/")
assert res.status_code == 401
assert res.json()
+24
View File
@@ -0,0 +1,24 @@
"""Postfix/Dovecot read these tables directly, so their names must never change."""
import pytest
from django.core.management import call_command
from django.db import connection
SCHEMA = {
"dpdapi_domain": {"id", "created", "updated", "name"},
"dpdapi_alias": {"id", "created", "updated", "source", "destination", "active", "domain_id"},
"dpdapi_user": {"id", "created", "updated", "email", "password", "domain_id"},
}
@pytest.mark.django_db
@pytest.mark.parametrize(("table", "columns"), SCHEMA.items())
def test_tables_and_columns(table, columns):
with connection.cursor() as cursor:
description = connection.introspection.get_table_description(cursor, table)
assert {c.name for c in description} == columns
@pytest.mark.django_db
def test_no_pending_migrations():
call_command("makemigrations", "dpdapi", "--check", "--dry-run")
+23
View File
@@ -0,0 +1,23 @@
import pytest
from dpdapi.models import Alias, Domain, User
def test_wsgi_application():
from dpdapisite.wsgi import application
assert callable(application)
@pytest.mark.parametrize("model", [Alias, Domain, User])
def test_admin_changelist(client, admin, domain, model):
client.force_login(admin)
meta = model._meta
res = client.get(f"/admin/{meta.app_label}/{meta.model_name}/", {"q": "example"})
assert res.status_code == 200
def test_sentry_disabled_in_tests():
import sentry_sdk
assert not sentry_sdk.get_client().is_active()
+107
View File
@@ -0,0 +1,107 @@
import pytest
from passlib.hash import sha512_crypt
from dpdapi.models import Domain, User
FIELDS = {"id", "created", "updated", "email", "domain"}
@pytest.fixture
def user(domain):
return User.objects.create(email="Jane@Example.com", password="x", domain=domain)
def test_list(api, user):
res = api.get("/api/users/")
assert res.status_code == 200
assert res.json() == [res.json()[0]]
assert set(res.json()[0]) == FIELDS
assert res.json()[0]["email"] == "jane@example.com"
assert res.json()[0]["domain"] == user.domain_id
def test_retrieve(api, user):
res = api.get(f"/api/users/{user.pk}/")
assert res.status_code == 200
assert set(res.json()) == FIELDS
assert str(user) == "jane@example.com"
def test_retrieve_missing(api, db):
assert api.get("/api/users/999/").status_code == 404
def test_create_hashes_password(api, domain):
data = {"email": "New@Example.com", "password": "hunter2", "domain": domain.pk}
res = api.post("/api/users/", data, content_type="application/json")
assert res.status_code == 201
assert set(res.json()) == FIELDS
assert res.json()["email"] == "new@example.com"
stored = User.objects.get().password
assert stored.startswith("$6$")
assert sha512_crypt.verify("hunter2", stored)
@pytest.mark.parametrize(
"data",
[
{"password": "p"},
{"email": "nope", "password": "p"},
{"email": "jane@example.com", "password": "p"},
{"email": "a@example.com"},
{"email": "a@example.com", "password": "p", "domain": 999},
],
)
def test_create_invalid(api, user, data):
data.setdefault("domain", user.domain_id)
res = api.post("/api/users/", data, content_type="application/json")
assert res.status_code == 400
assert User.objects.count() == 1
def test_update_hashes_password(api, user):
other = Domain.objects.create(name="other.org")
data = {"email": "jane@other.org", "password": "new-secret", "domain": other.pk}
res = api.put(f"/api/users/{user.pk}/", data, content_type="application/json")
assert res.status_code == 200
assert res.json()["email"] == "jane@other.org"
assert res.json()["domain"] == other.pk
user.refresh_from_db()
assert sha512_crypt.verify("new-secret", user.password)
def test_update_invalid(api, user):
res = api.put(f"/api/users/{user.pk}/", {"email": "x@example.com"}, content_type="application/json")
assert res.status_code == 400
def test_partial_update_keeps_password(api, user):
res = api.patch(f"/api/users/{user.pk}/", {"email": "J@example.com"}, content_type="application/json")
assert res.status_code == 200
assert res.json()["email"] == "j@example.com"
user.refresh_from_db()
assert user.password == "x"
def test_partial_update_hashes_password(api, user):
res = api.patch(f"/api/users/{user.pk}/", {"password": "p2"}, content_type="application/json")
assert res.status_code == 200
user.refresh_from_db()
assert sha512_crypt.verify("p2", user.password)
@pytest.mark.parametrize("field", ["email", "password", "domain"])
def test_partial_update_null(api, user, field):
res = api.patch(f"/api/users/{user.pk}/", {field: None}, content_type="application/json")
assert res.status_code == 400
assert field in str(res.json())
def test_delete(api, user):
assert api.delete(f"/api/users/{user.pk}/").status_code == 204
assert not User.objects.exists()
def test_deleting_domain_cascades(api, user):
assert api.delete(f"/api/domains/{user.domain_id}/").status_code == 204
assert not User.objects.exists()