forked from ungleich-public/ungleich-user
Sajeyks
1a1036e60b
The test helpers hardcoded dc=ungleich,dc=ch. They now build DNs from ENTIRE_SEARCH_BASE, so the suite can run against ungleich's test server (dc=my-domain,dc=com) by overriding settings through the environment. The TLS tests depend on the local dev LDAP and its CA, so they skip themselves when LDAPSERVER points elsewhere.
132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
"""
|
|
Shared helpers for the end-to-end tests.
|
|
|
|
These tests drive the real views over HTTP (Django test client) against a
|
|
real OpenLDAP server, started with:
|
|
|
|
docker compose -f docker-compose.dev.yml up -d --build
|
|
|
|
They are deliberately black-box: they assert on what a user sees (templates,
|
|
emails, status codes) and on what ends up in LDAP, so they keep passing while
|
|
the internals are rewritten during the upgrade.
|
|
"""
|
|
import re
|
|
import uuid
|
|
from unittest import mock
|
|
|
|
import ldap3
|
|
from django.conf import settings
|
|
from django.core import mail
|
|
from django.test import TestCase
|
|
|
|
from dal.ungleich_ldap import LdapManager
|
|
|
|
_ldap_checked = False
|
|
|
|
# The real test server; tests may override settings.LDAP_SERVERS.
|
|
TEST_LDAP_URI = settings.LDAP_SERVERS[0]
|
|
|
|
|
|
def admin_conn():
|
|
server = ldap3.Server(TEST_LDAP_URI)
|
|
return ldap3.Connection(
|
|
server, settings.LDAP_ADMIN_DN, settings.LDAP_ADMIN_PASSWORD,
|
|
auto_bind=True, raise_exceptions=False,
|
|
)
|
|
|
|
|
|
def ensure_ldap_up():
|
|
global _ldap_checked
|
|
if _ldap_checked:
|
|
return
|
|
try:
|
|
admin_conn().unbind()
|
|
except Exception as exc: # pragma: no cover - environment problem
|
|
raise RuntimeError(
|
|
"Test LDAP not reachable at {}. Start it with: docker compose -f "
|
|
"docker-compose.dev.yml up -d --build ({})".format(
|
|
TEST_LDAP_URI, exc)
|
|
)
|
|
_ldap_checked = True
|
|
|
|
|
|
def user_dn(uid, ou="customer"):
|
|
"""DN of a user entry; the tree comes from settings (ENTIRE_SEARCH_BASE)."""
|
|
return "uid={},ou={},{}".format(uid, ou, settings.ENTIRE_SEARCH_BASE)
|
|
|
|
|
|
def unique_uid(prefix="t"):
|
|
# Matches the app's username rule: lowercase letters, digits, - and _
|
|
return "{}-{}".format(prefix, uuid.uuid4().hex[:10])
|
|
|
|
|
|
class LdapTestCase(TestCase):
|
|
"""Base class: unique users per test, cleaned up afterwards."""
|
|
|
|
def setUp(self):
|
|
ensure_ldap_up()
|
|
self._created_dns = []
|
|
# OTP server is external; never hit the network from tests.
|
|
self.otp_post = mock.patch("dal.views.requests.post").start()
|
|
self.otp_post.return_value = mock.Mock(status_code=201, text="{}")
|
|
self.otp_get = mock.patch("dal.views.requests.get").start()
|
|
self.otp_get.return_value = mock.Mock(status_code=200, text="[]")
|
|
self.addCleanup(mock.patch.stopall)
|
|
|
|
def tearDown(self):
|
|
conn = admin_conn()
|
|
for dn in self._created_dns:
|
|
conn.delete(dn)
|
|
conn.unbind()
|
|
|
|
# LDAP helpers ---------------------------------------------------------
|
|
|
|
def make_ldap_user(self, uid=None, password="initial-pass-123",
|
|
ou="customer", email=None):
|
|
"""Create a user directly in LDAP, the same way the app does."""
|
|
uid = uid or unique_uid()
|
|
dn = user_dn(uid, ou)
|
|
conn = admin_conn()
|
|
ok = conn.add(dn, settings.LDAP_USER_OBJECT_CLASSES, {
|
|
"uid": uid, "cn": uid, "sn": "Tester", "givenName": "Tina",
|
|
"mail": email or "{}@example.com".format(uid),
|
|
"uidNumber": str(30000 + int(uuid.uuid4().int % 100000)),
|
|
"gidNumber": str(settings.LDAP_CUSTOMER_GROUP_ID), "homeDirectory": "/home/" + uid,
|
|
"userPassword": LdapManager()._ssha_password(password.encode()),
|
|
})
|
|
conn.unbind()
|
|
self.assertTrue(ok, "could not seed LDAP user {}".format(dn))
|
|
self._created_dns.append(dn)
|
|
return uid
|
|
|
|
def track(self, uid, ou="customer"):
|
|
"""Register a user created by the app for cleanup."""
|
|
self._created_dns.append(user_dn(uid, ou))
|
|
|
|
def ldap_entry(self, uid, attributes=("*",)):
|
|
conn = admin_conn()
|
|
conn.search(settings.ENTIRE_SEARCH_BASE, "(uid={})".format(
|
|
ldap3.utils.conv.escape_filter_chars(uid)), attributes=list(attributes))
|
|
entries = list(conn.entries)
|
|
conn.unbind()
|
|
return entries[0] if entries else None
|
|
|
|
def can_bind(self, uid, password, ou="customer"):
|
|
server = ldap3.Server(TEST_LDAP_URI)
|
|
conn = ldap3.Connection(
|
|
server, user_dn(uid, ou), password)
|
|
ok = conn.bind()
|
|
conn.unbind()
|
|
return ok
|
|
|
|
# HTTP helpers ---------------------------------------------------------
|
|
|
|
def login(self, uid, password):
|
|
self.assertTrue(self.client.login(username=uid, password=password),
|
|
"login failed for {}".format(uid))
|
|
|
|
def last_mail_link(self):
|
|
self.assertTrue(mail.outbox, "no email was sent")
|
|
urls = re.findall(r"https?://testserver(/[^\s\"'<>]+)", mail.outbox[-1].body)
|
|
self.assertTrue(urls, "no link in email: " + mail.outbox[-1].body)
|
|
return urls[-1]
|