✅ Fixed tests

guacamole_user_sync/ldap/ldap_client.py

@@ -5,6 +5,7 @@
 from ldap3.abstract.entry import Entry
 from ldap3.core.exceptions import (
     LDAPBindError,
+    LDAPException,
     LDAPSessionTerminatedByServerError,
     LDAPSocketOpenError,
 )
@@ -26,12 +27,14 @@ def __init__(
         self,
         hostname: str,
         *,
+        auto_bind: bool = True,
         bind_dn: str | None = None,
         bind_password: str | None = None,
     ) -> None:
+        self.auto_bind = auto_bind
         self.bind_dn = bind_dn
         self.bind_password = bind_password
-        self.hostname = hostname
+        self.server = Server(hostname, get_info=ALL)
 
     @staticmethod
     def as_list(ldap_entry: str | list[str] | None) -> list[str]:
@@ -45,20 +48,26 @@ def as_list(ldap_entry: str | list[str] | None) -> list[str]:
         raise ValueError(msg)
 
     def connect(self) -> Connection:
-        logger.info("Initialising connection to LDAP host at %s", self.hostname)
+        logger.info("Initialising connection to LDAP host at %s", self.server.host)
         try:
-            server = Server(self.hostname, get_info=ALL)
-            if self.bind_dn:
-                return Connection(
-                    server,
-                    self.bind_dn,
-                    self.bind_password,
-                    auto_bind=True,
-                )
-            return Connection(server, auto_bind=True)
+            return Connection(
+                self.server,
+                user=self.bind_dn,
+                password=self.bind_password,
+                auto_bind=self.auto_bind,
+            )
+        except LDAPSocketOpenError as exc:
+            msg = "Server could not be reached."
+            logger.exception(msg, exc_info=exc)
+            raise LDAPError(msg) from exc
         except LDAPBindError as exc:
-            logger.warning("Connection credentials were incorrect.")
-            raise LDAPError from exc
+            msg = "Connection credentials were incorrect."
+            logger.exception(msg, exc_info=exc)
+            raise LDAPError(msg) from exc
+        except LDAPException as exc:
+            msg = f"Unexpected LDAP exception of type {type(exc)}."
+            logger.exception(msg, exc_info=exc)
+            raise LDAPError(msg) from exc
 
     def search_groups(self, query: LDAPQuery) -> list[LDAPGroup]:
         output = []
@@ -96,13 +105,14 @@ def search(self, query: LDAPQuery) -> list[Entry]:
         try:
             connection = self.connect()
             connection.search(query.base_dn, query.filter, attributes=ALL_ATTRIBUTES)
-        except LDAPSocketOpenError as exc:
-            logger.warning("Server could not be reached.")
-            raise LDAPError from exc
-
         except LDAPSessionTerminatedByServerError as exc:
-            logger.warning("Server terminated LDAP request.")
-            raise LDAPError from exc
+            msg = "Server terminated LDAP request."
+            logger.exception(msg, exc_info=exc)
+            raise LDAPError(msg) from exc
+        except LDAPException as exc:
+            msg = f"Unexpected LDAP exception of type {type(exc)}."
+            logger.exception(msg, exc_info=exc)
+            raise LDAPError(msg) from exc
         else:
             results = cast(list[Entry], connection.entries)
             logger.debug("Server returned %s results.", len(results))

guacamole_user_sync/postgresql/postgresql_client.py

@@ -89,7 +89,11 @@ def assign_users_to_groups(
                 )
                 continue
             # Get the user_entity_id for each user belonging to this group
-            logger.debug("Group '%s' has %s members", group.name, len(group.member_uid))
+            logger.debug(
+                "Group '%s' has %s member(s).",
+                group.name,
+                len(group.member_uid),
+            )
             for user_uid in group.member_uid:
                 try:
                     user = next(filter(lambda u: u.uid == user_uid, users))

tests/conftest.py

@@ -2,14 +2,16 @@
 
 import pytest
 
-from guacamole_user_sync.models import LDAPGroup, LDAPQuery, LDAPSearchResult, LDAPUser
+from guacamole_user_sync.models import LDAPGroup, LDAPQuery, LDAPUser
 from guacamole_user_sync.postgresql.orm import (
     GuacamoleEntity,
     GuacamoleEntityType,
     GuacamoleUser,
     GuacamoleUserGroup,
 )
 
+from .mocks import MockLDAPGroupEntry, MockLDAPUserEntry
+
 
 @pytest.fixture
 def ldap_model_groups_fixture() -> list[LDAPGroup]:
@@ -69,70 +71,45 @@ def ldap_query_users_fixture() -> LDAPQuery:
 
 
 @pytest.fixture
-def ldap_response_groups_fixture() -> LDAPSearchResult:
+def ldap_response_groups_fixture() -> list[MockLDAPGroupEntry]:
     return [
-        (
-            0,
-            (
-                "CN=plaintiffs,OU=groups,DC=rome,DC=la",
-                {
-                    "cn": [b"plaintiffs"],
-                    "memberOf": [],
-                    "memberUid": [b"aulus.agerius"],
-                },
-            ),
-        ),
-        (
-            1,
-            (
-                "CN=defendants,OU=groups,DC=rome,DC=la",
-                {
-                    "cn": [b"defendants"],
-                    "memberOf": [],
-                    "memberUid": [b"numerius.negidius"],
-                },
-            ),
-        ),
-        (
-            2,
-            (
-                "CN=everyone,OU=groups,DC=rome,DC=la",
-                {
-                    "cn": [b"everyone"],
-                    "memberOf": [],
-                    "memberUid": [b"aulus.agerius", b"numerius.negidius"],
-                },
-            ),
+        MockLDAPGroupEntry(
+            dn="CN=plaintiffs,OU=groups,DC=rome,DC=la",
+            cn="plaintiffs",
+            memberOf=[],
+            memberUid=["aulus.agerius"],
+        ),
+        MockLDAPGroupEntry(
+            dn="CN=defendants,OU=groups,DC=rome,DC=la",
+            cn="defendants",
+            memberOf=[],
+            memberUid=["numerius.negidius"],
+        ),
+        MockLDAPGroupEntry(
+            dn="CN=everyone,OU=groups,DC=rome,DC=la",
+            cn="everyone",
+            memberOf=[],
+            memberUid=["aulus.agerius", "numerius.negidius"],
         ),
     ]
 
 
 @pytest.fixture
-def ldap_response_users_fixture() -> LDAPSearchResult:
+def ldap_response_users_fixture() -> list[MockLDAPUserEntry]:
     return [
-        (
-            0,
-            (
-                "CN=aulus.agerius,OU=users,DC=rome,DC=la",
-                {
-                    "displayName": [b"Aulus Agerius"],
-                    "memberOf": [b"CN=plaintiffs,OU=groups,DC=rome,DC=la"],
-                    "uid": [b"aulus.agerius"],
-                    "userName": [b"[email protected]"],
-                },
-            ),
-        ),
-        (
-            1,
-            (
-                "CN=numerius.negidius,OU=users,DC=rome,DC=la",
-                {
-                    "displayName": [b"Numerius Negidius"],
-                    "memberOf": [b"CN=defendants,OU=groups,DC=rome,DC=la"],
-                    "uid": [b"numerius.negidius"],
-                    "userName": [b"[email protected]"],
-                },
-            ),
+        MockLDAPUserEntry(
+            dn="CN=aulus.agerius,OU=users,DC=rome,DC=la",
+            displayName="Aulus Agerius",
+            memberOf=["CN=plaintiffs,OU=groups,DC=rome,DC=la"],
+            uid="aulus.agerius",
+            userName="[email protected]",
+        ),
+        MockLDAPUserEntry(
+            dn="CN=numerius.negidius,OU=users,DC=rome,DC=la",
+            displayName="Numerius Negidius",
+            memberOf=["CN=defendants,OU=groups,DC=rome,DC=la"],
+            uid="numerius.negidius",
+            userName="[email protected]",
         ),
     ]
 

tests/mocks.py

@@ -1,67 +1,88 @@
 from typing import Any
 
-import ldap
+from ldap3.core.exceptions import LDAPBindError
 from sqlalchemy import TextClause
 
-from guacamole_user_sync.models import LDAPSearchResult
 from guacamole_user_sync.postgresql.orm import GuacamoleBase
 
 
-class MockLDAPObject:
-    """Mock LDAPObject."""
+class MockLDAPAttribute:
+    """Mock LDAP value."""
 
-    def __init__(self, uri: str) -> None:
-        self.uri = uri
-        self.bind_dn = ""
-        self.bind_password = ""
+    def __init__(self, value: str | float | list[str]) -> None:
+        self.value = value
 
-    def simple_bind_s(self, bind_dn: str, bind_password: str) -> None:
-        if bind_password == "incorrect-password":  # noqa: S105
-            raise ldap.INVALID_CREDENTIALS
-        self.bind_dn = bind_dn
-        self.bind_password = bind_password
 
-
-class MockAsyncSearchList:
-    """Mock AsyncSearchList."""
+class MockLDAPGroupEntry:
+    """Mock LDAP group entry."""
 
     def __init__(
         self,
-        partial: bool,  # noqa: FBT001
-        results: LDAPSearchResult,
-        *args: Any,  # noqa: ANN401, ARG002
-        **kwargs: Any,  # noqa: ANN401, ARG002
+        dn: str,
+        cn: str,
+        memberOf: list[str],  # noqa: N803
+        memberUid: list[str],  # noqa: N803
     ) -> None:
-        self.allResults = results
-        self.partial = partial
+        self.dn = MockLDAPAttribute(dn)
+        self.cn = MockLDAPAttribute(cn)
+        self.memberOf = MockLDAPAttribute(memberOf)
+        self.memberUid = MockLDAPAttribute(memberUid)
 
-    def startSearch(  # noqa: N802
-        self,
-        *args: Any,  # noqa: ANN401
-        **kwargs: Any,  # noqa: ANN401
-    ) -> None:
-        pass
 
-    def processResults(  # noqa: N802
+class MockLDAPUserEntry:
+    """Mock LDAP user entry."""
+
+    def __init__(
         self,
-        *args: Any,  # noqa: ANN401, ARG002
-        **kwargs: Any,  # noqa: ANN401, ARG002
-    ) -> bool:
-        return self.partial
+        dn: str,
+        displayName: str,  # noqa: N803
+        memberOf: list[str],  # noqa: N803
+        uid: str,
+        userName: str,  # noqa: N803
+    ) -> None:
+        self.dn = MockLDAPAttribute(dn)
+        self.displayName = MockLDAPAttribute(displayName)
+        self.memberOf = MockLDAPAttribute(memberOf)
+        self.uid = MockLDAPAttribute(uid)
+        self.userName = MockLDAPAttribute(userName)
 
 
-class MockAsyncSearchListFullResults(MockAsyncSearchList):
-    """Mock AsyncSearchList with full results."""
+class MockLDAPServer:
+    """Mock LDAP server."""
 
-    def __init__(self, results: LDAPSearchResult) -> None:
-        super().__init__(results=results, partial=False)
+    def __init__(
+        self,
+        entries: list[MockLDAPGroupEntry] | list[MockLDAPUserEntry],
+    ) -> None:
+        self.entries = entries
 
 
-class MockAsyncSearchListPartialResults(MockAsyncSearchList):
-    """Mock AsyncSearchList with partial results."""
+class MockLDAPConnection:
+    """Mock LDAP connection."""
 
-    def __init__(self, results: LDAPSearchResult) -> None:
-        super().__init__(results=results, partial=True)
+    def __init__(
+        self,
+        server: MockLDAPServer | None = None,
+        user: str | None = None,
+        password: str | None = None,
+        *,
+        auto_bind: bool = False,
+    ) -> None:
+        self.auto_bind = auto_bind
+        self.password = password
+        if password == "incorrect-password":  # noqa: S105
+            raise LDAPBindError
+        self.server = server
+        self.user = user
+
+    def search(
+        self,
+        base_dn: str,  # noqa: ARG002
+        ldap_filter: str,  # noqa: ARG002
+        attributes: str,  # noqa: ARG002
+    ) -> None:
+        if self.server:
+            self.entries = self.server.entries
 
 
 class MockPostgreSQLBackend: