unittest.py

from __future__ import annotations
import backend.models as bmodels
import legacy.models as lmodels
from backend.models import Person, AM, Fingerprint, _build_fullname
from backend import const
from dsa.models import LDAPFields
from signon.models import Identity
from django.conf import settings
from django.utils.timezone import now
from django.contrib.auth.backends import ModelBackend
from django.test import Client
from rest_framework.test import APIClient
from collections import defaultdict
import nm2.lib.unittest
import contextlib
import datetime
import re


class NamedObjects(nm2.lib.unittest.NamedObjects):
    def create(self, _name, **kw):
        self._update_kwargs_with_defaults(_name, kw)
        self[_name] = o = self._model.objects.create(**kw)
        return o


class TestPersons(NamedObjects):
    def __init__(self, **defaults):
        defaults.setdefault("cn", lambda name, **kw: name.capitalize())
        defaults.setdefault("email", "{_name}@example.org")
        defaults.setdefault("email_ldap", "{_name}@example.org")
        super(TestPersons, self).__init__(Person, **defaults)

    def create(self, _name, **kw):
        self._update_kwargs_with_defaults(_name, kw)

        # Allow to create Person and LDAPFields in one shot
        ldap_fields = {}
        for field in ("cn", "mn", "sn", "uid"):
            if field in kw:
                ldap_fields[field] = kw.pop(field)
        if "email_ldap" in kw:
            ldap_fields["email"] = kw.pop("email_ldap")
        ldap_fields.setdefault("uid", _name)

        if "fullname" not in kw:
            kw["fullname"] = _build_fullname(ldap_fields.get("cn"), ldap_fields.get("mn"), ldap_fields.get("sn"))

        self[_name] = o = self._model.objects.create_user(audit_skip=True, **kw)
        LDAPFields.objects.create(person=o, audit_skip=True, **ldap_fields)

        return o


class TestKeys(NamedObjects):
    def __init__(self, **defaults):
        from keyring.models import Key
        super(TestKeys, self).__init__(Key, **defaults)

    def create(self, _name, **kw):
        self._update_kwargs_with_defaults(_name, kw)
        self._model.objects.test_preload(_name)
        self[_name] = o = self._model.objects.get_or_download(_name, **kw)
        return o


class TestAuthenticationBackend(ModelBackend):
    pass


class TestBase(nm2.lib.unittest.TestBase):
    TEST_AUTH_BACKEND = "backend.unittest.TestAuthenticationBackend"

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        if cls.TEST_AUTH_BACKEND not in settings.AUTHENTICATION_BACKENDS:
            settings.AUTHENTICATION_BACKENDS.append(cls.TEST_AUTH_BACKEND)

    @classmethod
    def tearDownClass(cls):
        if cls.TEST_AUTH_BACKEND in settings.AUTHENTICATION_BACKENDS:
            settings.AUTHENTICATION_BACKENDS.remove(cls.TEST_AUTH_BACKEND)
        super().tearDownClass()

    def make_test_client(self, person, signon_identities=()):
        """
        Instantiate a test client, logging in the given person.

        If person is None, visit anonymously. If person is None but
        sso_username is not None, authenticate as the given sso_username even
        if a Person record does not exist.
        """
        client = Client()
        for identity in signon_identities:
            identity = self.identities[identity]
            if identity.issuer == "salsa":
                session = client.session
                session["signon_identity_salsa"] = identity.pk
                session.save()
            elif identity.issuer == "debsso":
                client.defaults["SSL_CLIENT_S_DN_CN"] = identity.subject
            else:
                raise NotImplementedError(f"{identity.issuer} not supported as identity during testing")

        if isinstance(person, str):
            person = self.persons[person]
        if person is not None:
            client.force_login(person, backend=self.TEST_AUTH_BACKEND)
        client.visitor = person
        return client

    def make_test_apiclient(self, person, signon_identities=()):
        """
        Instantiate a test client, logging in the given person.

        If person is None, visit anonymously. If person is None but
        sso_username is not None, authenticate as the given sso_username even
        if a Person record does not exist.
        """
        client = APIClient()
        for identity in signon_identities:
            identity = self.identities[identity]
            if identity.issuer == "salsa":
                client.session[f"signon_identity_salsa"] = identity.pk
                client.session.save()
            elif identity.issuer == "debsso":
                client.defaults["SSL_CLIENT_S_DN_CN"] = identity.subject
            else:
                raise NotImplementedError(f"{identity.issuer} not supported as identity during testing")

        if isinstance(person, str):
            person = self.persons[person]
        if person is not None:
            client.force_login(person, backend=self.TEST_AUTH_BACKEND)
        client.visitor = person
        return client

    def assertPermissionDenied(self, response):
        if response.status_code == 403:
            pass
        else:
            self.fail("response has status code {} instead of a 403 Forbidden".format(response.status_code))

    def assertFormRedirectMatches(self, response, target, form_name="form"):
        if response.status_code == 200:
            form = response.context[form_name]
            if form.errors:
                self.fail("{} did not validate. Errors: {}".format(form_name, repr(form.errors)))
            self.fail("response has status code 200 instead of redirecting, and did not find any form errors")
        if response.status_code != 302:
            self.fail("response has status code {} instead of a Redirect".format(response.status_code))
        if target and not re.search(target, response["Location"]):
            self.fail("response redirects to {} which does not match {}".format(response["Location"], target))

    def assertContainsElements(self, response, elements, *names):
        """
        Check that the response contains only the elements in `names` from PageElements `elements`
        """
        want = set(names)
        extras = want - set(elements.keys())
        if extras:
            raise RuntimeError("Wanted elements not found in the list of possible ones: {}".format(", ".join(extras)))
        should_have = []
        should_not_have = []
        content = response.content.decode("utf-8")
        for name, regex in elements.items():
            if name in want:
                if not regex.search(content):
                    should_have.append(name)
            else:
                if regex.search(content):
                    should_not_have.append(name)
        if should_have or should_not_have:
            msg = []
            if should_have:
                msg.append("should have element(s) {}".format(", ".join(should_have)))
            if should_not_have:
                msg.append("should not have element(s) {}".format(", ".join(should_not_have)))
            self.fail("page " + " and ".join(msg))


class BaseFixtureMixin(TestBase):
    @classmethod
    def get_persons_defaults(cls):
        """
        Get default arguments for test persons
        """
        return {}

    @classmethod
    def setUpClass(cls):
        import process.models as pmodels
        super(BaseFixtureMixin, cls).setUpClass()
        cls.add_named_objects(
            persons=TestPersons(**cls.get_persons_defaults()),
            identities=NamedObjects(Identity),
            fingerprints=NamedObjects(Fingerprint),
            ams=NamedObjects(AM),
            processes=NamedObjects(pmodels.Process),
            keys=TestKeys()
        )

        # Preload keys
        cls.keys.create("66B4DFB68CB24EBBD8650BC4F4B4B0CC797EBFAB")
        cls.keys.create("1793D6AB75663E6BF104953A634F4BD1E7AD5568")
        cls.keys.create("0EED77DC41D760FDE44035FF5556A34E04A3610B")

    @classmethod
    def create_person(cls, name, *args, alioth=False, username=None, **kw):
        if username is None:
            if alioth:
                username = name + "-guest@users.alioth.debian.org"
            else:
                username = name + "@debian.org"
        p = cls.persons.create(name, *args, **kw)
        # cls.identities.create(
        #         f"{name}_debsso", person=p, issuer="debsso", subject=username, username=username,
        #         audit_skip=True)
        return p


class OpFixtureMixin(BaseFixtureMixin):
    def check_op(self, o, check_contents):
        # FIXME: compatibility, remove when code has been ported
        return self.assertOperationSerializes(o, check_contents)

    def assertOperationSerializes(self, o, check_contents=None):
        """
        When run with check_contents set to a function, it
        serializes/deserializes o and checks its contents with the
        check_contents function.

        When run without check_contents, it acts as a decorator, allowing a
        shorter invocation, like this:

           @self.assertOperationSerializes(operation_to_check)
           def _(o):
               self.assertEqual(o.audit_author, self.persons.fd)
               # ...more checks on o...
        """
        if check_contents is None:
            # Act as a decorator
            def run_test(check_contents):
                self.assertOperationSerializes(o, check_contents)
            return run_test
        else:
            # Actually run the test
            from backend import ops
            cls = o.__class__

            self.assertIsInstance(o.audit_time, datetime.datetime)
            check_contents(o)

            d = o.to_dict()
            o = cls(**d)
            self.assertIsInstance(o.audit_time, datetime.datetime)
            check_contents(o)

            j = o.to_json()
            o = ops.Operation.from_json(j)
            self.assertIsInstance(o, cls)
            self.assertIsInstance(o.audit_time, datetime.datetime)
            check_contents(o)

    @contextlib.contextmanager
    def collect_operations(self):
        from backend import ops
        with ops.Operation.test_collect() as ops:
            yield ops


class PersonFixtureMixin(OpFixtureMixin):
    """
    Pre-create some persons
    """
    @classmethod
    def setUpClass(cls):
        super(PersonFixtureMixin, cls).setUpClass()
        # pending account
        cls.create_person(
                "pending", status=const.STATUS_DC, expires=now() +
                datetime.timedelta(days=1), pending="12345", alioth=True)
        # debian contributor
        cls.create_person("dc", status=const.STATUS_DC, alioth=True)
        # debian contributor with guest account
        cls.create_person("dc_ga", status=const.STATUS_DC_GA, alioth=True)
        # dm
        cls.create_person("dm", status=const.STATUS_DM, alioth=True)
        # dm with guest account
        cls.create_person("dm_ga", status=const.STATUS_DM_GA, alioth=True)
        # dd, nonuploading
        cls.create_person("dd_nu", status=const.STATUS_DD_NU)
        # dd, uploading
        cls.create_person("dd_u", status=const.STATUS_DD_U)
        # dd, emeritus
        cls.create_person("dd_e", status=const.STATUS_EMERITUS_DD)
        # dd, removed
        cls.create_person("dd_r", status=const.STATUS_REMOVED_DD)
        # unrelated active am
        activeam = cls.create_person("activeam", status=const.STATUS_DD_NU)
        cls.ams.create("activeam", person=activeam)
        # inactive am
        oldam = cls.create_person("oldam", status=const.STATUS_DD_NU)
        cls.ams.create("oldam", person=oldam, is_am=False)
        # fd
        fd = cls.create_person("fd", status=const.STATUS_DD_NU)
        cls.ams.create("fd", person=fd, is_fd=True)
        # dam
        dam = cls.create_person("dam", status=const.STATUS_DD_U)
        cls.ams.create("dam", person=dam, is_fd=True, is_dam=True)


class TestSet(set):
    """
    Set of strings that can be initialized from space-separated strings, and
    changed with simple text patches.
    """
    def __init__(self, initial=""):
        if initial:
            self.update(initial.split())

    def set(self, vals):
        self.clear()
        self.update(vals.split())

    def patch(self, diff):
        for change in diff.split():
            if change[0] == "+":
                self.add(change[1:])
            elif change[0] == "-":
                self.discard(change[1:])
            else:
                raise RuntimeError("Changes {} contain {} that is nether an add nor a remove".format(
                    repr(diff), repr(change)))

    def clone(self):
        res = TestSet()
        res.update(self)
        return res


class PatchExact(object):
    def __init__(self, text):
        if text:
            self.items = set(text.split())
        else:
            self.items = set()

    def apply(self, cur):
        if self.items:
            return set(self.items)
        return None


class PatchDiff(object):
    def __init__(self, text):
        self.added = set()
        self.removed = set()
        for change in text.split():
            if change[0] == "+":
                self.added.add(change[1:])
            elif change[0] == "-":
                self.removed.add(change[1:])
            else:
                raise RuntimeError("Changes {} contain {} that is nether an add nor a remove".format(text, change))

    def apply(self, cur):
        if cur is None:
            cur = set(self.added)
        else:
            cur = (cur - self.removed) | self.added
        if not cur:
            return None
        return cur


class ExpectedSets(defaultdict):
    """
    Store the permissions expected out of a *VisitorPermissions object
    """
    def __init__(self, testcase, action_msg="{visitor}", issue_msg="{problem} {mismatch}"):
        super(ExpectedSets, self).__init__(TestSet)
        self.testcase = testcase
        self.action_msg = action_msg
        self.issue_msg = issue_msg

    @property
    def visitors(self):
        return list(self.keys())

    def __getitem__(self, key):
        if key == "-":
            key = None
        return super().__getitem__(key)

    def set(self, visitors, text):
        for v in visitors.split():
            self[v].set(text)

    def patch(self, visitors, text):
        for v in visitors.split():
            self[v].patch(text)

    def select_others(self, persons):
        other_visitors = set(persons.keys())
        other_visitors.add(None)
        other_visitors -= set(self.keys())
        return other_visitors

    def combine(self, other):
        res = ExpectedSets(self.testcase, action_msg=self.action_msg, issue_msg=self.issue_msg)
        for k, v in list(self.items()):
            res[k] = v.clone()
        for k, v in list(other.items()):
            res[k].update(v)
        return res

    def assertEqual(self, visitor, got):
        got = set(got)
        wanted = self.get(visitor, set())
        if got == wanted:
            return
        extra = got - wanted
        missing = wanted - got
        msgs = []
        if missing:
            msgs.append(self.issue_msg.format(problem="misses", mismatch=", ".join(sorted(missing))))
        if extra:
            msgs.append(self.issue_msg.format(problem="has extra", mismatch=", ".join(sorted(extra))))
        self.testcase.fail(self.action_msg.format(visitor=visitor) + " " + " and ".join(msgs))

    def assertEmpty(self, visitor, got):
        extra = set(got)
        if not extra:
            return
        self.testcase.fail(
                self.action_msg.format(visitor=visitor) + " " +
                self.issue_msg.format(problem="has", mismatch=", ".join(sorted(extra))))

    def assertMatches(self, visited):
        for visitor in self.visitors:
            visit_perms = visited.permissions_of(self.testcase.persons[visitor])
            self.assertEqual(visitor, visit_perms)
        for visitor in self.select_others(self.testcase.persons):
            visit_perms = visited.permissions_of(self.testcase.persons[visitor] if visitor else None)
            self.assertEmpty(visitor, visit_perms)


class ExpectedPerms(object):
    """
    Store the permissions expected out of a *VisitorPermissions object
    """
    def __init__(self, perms=None):
        if perms is None:
            perms = {}
        self.perms = {}
        for visitors, expected_perms in perms.items():
            for visitor in visitors.split():
                self.perms[visitor] = set(expected_perms.split())

    def _apply_diff(self, d, diff):
        for visitors, change in diff.items():
            for visitor in visitors.split():
                cur = change.apply(d.get(visitor, None))
                if not cur:
                    d.pop(visitor, None)
                else:
                    d[visitor] = cur

    def update_perms(self, diff):
        self._apply_diff(self.perms, diff)

    def set_perms(self, visitors, text):
        self.update_perms({visitors: PatchExact(text)})

    def patch_perms(self, visitors, text):
        self.update_perms({visitors: PatchDiff(text)})


class PageElements(dict):
    """
    List of all page elements possibly expected in the results of a view.

    dict matching name used to refer to the element with regexp matching the
    element.
    """
    def add_id(self, id):
        self[id] = re.compile(r"""id\s*=\s*["']{}["']""".format(re.escape(id)))

    def add_class(self, cls):
        self[cls] = re.compile(r"""class\s*=\s*["']{}["']""".format(re.escape(cls)))

    def add_href(self, name, url):
        self[name] = re.compile(r"""href\s*=\s*["']{}["']""".format(re.escape(url)))

    def add_th(self, name, title):
        self[name] = re.compile(r"""<th>\s*{}\s*</th>""".format(re.escape(title)))

    def add_string(self, name, term):
        self[name] = re.compile(r"""{}""".format(re.escape(term)))

    def clone(self):
        res = PageElements()
        res.update(self.items())
        return res


class TestOldProcesses(NamedObjects):
    def __init__(self, **defaults):
        super(TestOldProcesses, self).__init__(lmodels.Process, **defaults)
        defaults.setdefault("progress", const.PROGRESS_APP_NEW)

    def create(self, _name, advocates=[], **kw):
        self._update_kwargs_with_defaults(_name, kw)

        if "process" in kw:
            kw.setdefault("is_active", kw["process"] not in (const.PROGRESS_DONE, const.PROGRESS_CANCELLED))
        else:
            kw.setdefault("is_active", True)

        if "manager" in kw:
            try:
                am = kw["manager"].am
            except bmodels.AM.DoesNotExist:
                am = bmodels.AM.objects.create(person=kw["manager"])
            kw["manager"] = am

        self[_name] = o = self._model.objects.create(**kw)
        for a in advocates:
            o.advocates.add(a)
        return o


class OldProcessFixtureMixin(PersonFixtureMixin):
    @classmethod
    def get_processes_defaults(cls):
        """
        Get default arguments for test processes
        """
        return {}

    @classmethod
    def setUpClass(cls):
        super(OldProcessFixtureMixin, cls).setUpClass()
        cls.processes = TestOldProcesses(**cls.get_processes_defaults())