ungleich-user/dal/tests/base.py

"""
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 os
import re
import tempfile
import uuid
from unittest import mock

import ldap3
from django.conf import settings
from django.core import mail
from django.test import TestCase, override_settings

from dal.ungleich_ldap import LdapManager

_ldap_checked = False


def admin_conn():
    server = ldap3.Server(settings.AUTH_LDAP_SERVER)
    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(
                settings.AUTH_LDAP_SERVER, 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 = []
        fd, self.uid_file = tempfile.mkstemp(prefix="ldap_max_uid_")
        os.close(fd)
        os.unlink(self.uid_file)  # app must cope with a missing file
        self._settings = override_settings(LDAP_MAX_UID_FILE_PATH=self.uid_file)
        self._settings.enable()
        # 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()
        self._settings.disable()
        if os.path.exists(self.uid_file):
            os.unlink(self.uid_file)

    # 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(settings.AUTH_LDAP_SERVER)
        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]