Adding Mock LDAP Functionality for improved testing

docs/source/configuration.rst

@@ -13,10 +13,21 @@ Core
                                          connecting to LDAP. Defaults to 
                                          ``389``.
 
-``LDAP_HOST``                            Speficies the address of the server to
-                                         connect to by default. ``None``.
+``LDAP_HOST``                            Specifies the address of the default
+                                         server to use when connecting to LDAP.
                                          Additional servers can be added via the
                                          ``add_server`` method.
+                                         Defaults to ``None``.
+
+``LDAP_MOCK_DATA``                       If specified, configures
+                                         ``ldap3.Connection`` with
+                                         ``client_strategy=MOCK_SYNC`` to setup
+                                         testing with a mock LDAP connection.
+                                         [#ldap3mock]_
+                                         Useful for running unit tests.
+                                         The value is used to point to a json
+                                         entries file to load data to the DIT.
+                                         Defaults to ``None`` (no mocking).
 
 ``LDAP_USE_SSL``                         Specifies whether the default server
                                          connection should use SSL. Defaults to
@@ -160,3 +171,7 @@ Filters/Searching
                                      Defaults to ``ldap3.ALL_ATTRIBUTES``       
 
 ==================================== ================================================
+
+
+.. [#ldap3mock] For details about mocking the ldap3.Connection,
+   see `Mocking -- ldap3 documentation <https://ldap3.readthedocs.io/en/latest/mocking.html>`_

flask_ldap3_login/__init__.py

@@ -47,6 +47,7 @@
     ("LDAP_GROUP_MEMBERS_ATTR", "uniqueMember"),
     ("LDAP_GET_GROUP_ATTRIBUTES", ldap3.ALL_ATTRIBUTES),
     ("LDAP_ADD_SERVER", True),
+    ("LDAP_MOCK_DATA", None),
 ]
 
 
@@ -784,21 +785,46 @@ def _make_connection(
                 ldap3, current_app.config.get("LDAP_BIND_AUTHENTICATION_TYPE")
             )
 
+        if current_app.config.get("LDAP_MOCK_DATA") is None:
+            strategy = ldap3.SYNC
+            server_arg = app.ldap3_login_manager_server_pool
+        else:
+            log.info("Using MOCK_SYNC")
+            strategy = ldap3.MOCK_SYNC
+            # This is needed because using MOCK with ServerPool is broken,
+            # see https://github.com/cannatag/ldap3/issues/1007
+            server_arg = ldap3.Server("fake_server")
+
         log.debug(
-            "Opening connection with bind user '{}'".format(bind_user or "Anonymous")
+            "Opening connection with bind user '{}' [{}]".format(
+                bind_user or "Anonymous", strategy
+            )
         )
         connection = ldap3.Connection(
-            server=app.ldap3_login_manager_server_pool,
+            server=server_arg,
             read_only=current_app.config.get("LDAP_READONLY"),
             user=bind_user,
             password=bind_password,
-            client_strategy=ldap3.SYNC,
+            client_strategy=strategy,
             authentication=authentication,
             check_names=current_app.config["LDAP_CHECK_NAMES"],
             raise_exceptions=True,
             **kwargs
         )
 
+        if current_app.config.get("LDAP_MOCK_DATA") is not None:
+            # TODO: Should use current_app.instance_path relative path
+            #  or app.open_instance_resource to open file, but entries_from_json
+            #  expects a filename to open, not file data.
+            log.info(
+                "Loading LDAP_MOCK_DATA from: {}".format(
+                    current_app.config.get("LDAP_MOCK_DATA")
+                )
+            )
+            connection.strategy.entries_from_json(
+                current_app.config.get("LDAP_MOCK_DATA")
+            )
+
         if contextualise:
             self._contextualise_connection(connection)
         return connection

flask_ldap3_login_tests/Directory.py

@@ -1,3 +1,5 @@
+import json
+
 import ldap3
 
 DIRECTORY = {
@@ -9,6 +11,7 @@
                 "objectclass": ["person"],
                 "dn": "cn=Bind,dc=mydomain,dc=com",
                 "password": "bind123",
+                "userPassword": "bind123",
             },
             "ou=users": {
                 "cn=Nick Whyte": {
@@ -90,3 +93,32 @@ def get_directory_base_recurse(location, directory):
 def get_directory_base(dn):
     location = list(reversed(dn.split(",")))
     return get_directory_base_recurse(location, directory=DIRECTORY)
+
+
+def key_path_recurse(d, path=None):
+    """Used by `dump_directory_to_file` to flatten DIRECTORY"""
+    keys = d.keys()
+    if any("=" in k for k in d):  # If any keys have "=", assume they are paths.
+        result = list()
+        for k in keys:
+            new_path = ",".join([k, path]) if path else k
+            kres = key_path_recurse(d[k], path=new_path)
+            if isinstance(kres, list):
+                result.extend(kres)
+            elif isinstance(kres, dict):
+                result.append(kres)
+            else:
+                raise ValueError("Unexpected type for key result: {}".format(kres))
+        return result
+    else:  # Otherwise, assume it's the attributes.
+        return {"dn": path, "raw": d}
+
+
+def dump_directory_to_file(filename):
+    """
+    Reformat the test directory data to a format used
+    by ldap3.MockBaseStrategy.entries_from_json and save it to filename
+    """
+    entries = key_path_recurse(DIRECTORY)
+    with open(filename, "w") as outfile:
+        json.dump({"entries": entries}, outfile, indent=2)

flask_ldap3_login_tests/test_ldap3_login.py

@@ -1,4 +1,6 @@
 import logging
+import os
+import tempfile
 import unittest
 
 import flask
@@ -9,7 +11,7 @@
 import flask_ldap3_login as ldap3_login
 from flask_ldap3_login.forms import LDAPLoginForm
 
-from .Directory import DIRECTORY
+from .Directory import DIRECTORY, dump_directory_to_file
 from .MockTypes import Connection, Server, ServerPool
 
 try:
@@ -644,3 +646,35 @@ def test_check_names_false(self, connection):
         self.manager.authenticate("janecitizen", "fake321")
         connection.assert_called_once()
         self.assertEqual(connection.call_args[1]["check_names"], False)
+
+
+class MockConnectionTestCase(unittest.TestCase):
+    def setUp(self):
+        # Create Temporary Dump of Directory Data
+        self.ldap_dump_fd, self.ldap_dump_path = tempfile.mkstemp()
+        dump_directory_to_file(self.ldap_dump_path)
+        # Initialize App with updated config
+        app = flask.Flask(__name__)
+        app.config.update(BASE_CONFIG)
+        app.config.update({"LDAP_MOCK_DATA": self.ldap_dump_path})
+        self.app = app
+        # Setup LDAP3LoginManager using mock data
+        ldap3_manager = ldap3_login.LDAP3LoginManager(app)
+        self.manager = ldap3_manager
+        self.app.app_context().push()
+
+    def tearDown(self):
+        stack.top.pop()
+        super().tearDown()
+        os.close(self.ldap_dump_fd)
+        os.unlink(self.ldap_dump_path)
+
+    def test_get_user_info_for_username(self):
+        user = self.manager.get_user_info_for_username("[email protected]")
+        # The MOCK ldap3 Connection returns the "password" attribute as a list.
+        DIRECTORY["dc=com"]["dc=mydomain"]["ou=users"]["cn=Nick Whyte"]["password"] = [
+            DIRECTORY["dc=com"]["dc=mydomain"]["ou=users"]["cn=Nick Whyte"]["password"]
+        ]
+        self.assertEqual(
+            user, DIRECTORY["dc=com"]["dc=mydomain"]["ou=users"]["cn=Nick Whyte"]
+        )