forked from ungleich-public/ungleich-user
Sajeyks
187dd8d6cc
LDAPSERVER was documented as a space separated list, but the whole string went into a single ldap3.Server. Logins worked (python-ldap accepts lists), while registration and every write would have failed with more than one URI. settings.LDAP_SERVERS is now the split list, used through one ldap3 ServerPool per process: requests go to the first reachable server, a server that is down is skipped for 60s, connections time out after 5s, and with all servers down the pool gives up after two passes instead of retrying forever.
127 lines
4.5 KiB
Python
127 lines
4.5 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 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="customers", email=None):
|
|
"""Create a user directly in LDAP, the same way the app does."""
|
|
uid = uid or unique_uid()
|
|
dn = "uid={},ou={},dc=ungleich,dc=ch".format(uid, ou)
|
|
conn = admin_conn()
|
|
ok = conn.add(dn, ["inetOrgPerson", "posixAccount", "ldapPublicKey"], {
|
|
"uid": uid, "cn": uid, "sn": "Tester", "givenName": "Tina",
|
|
"mail": email or "{}@example.com".format(uid),
|
|
"uidNumber": str(30000 + int(uuid.uuid4().int % 100000)),
|
|
"gidNumber": "2000", "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="customers"):
|
|
"""Register a user created by the app for cleanup."""
|
|
self._created_dns.append("uid={},ou={},dc=ungleich,dc=ch".format(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="customers"):
|
|
server = ldap3.Server(TEST_LDAP_URI)
|
|
conn = ldap3.Connection(
|
|
server, "uid={},ou={},dc=ungleich,dc=ch".format(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]
|