PacketStation/tests/server/fakedir.py

"""An in-memory DirectoryProtocol for tests. Passwords here are obviously fake fixture values."""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field

from packetstation.auth.groups import group_names
from packetstation.auth.ldap import AdUser, LdapUnavailable

FAKE_PASSWORD = "fake-password-1"
NESTED_GROUP = "Shop-Welders-Nested"
BASE_DN = "DC=pstest,DC=local"
OU_DN = "OU=PacketStation,OU=Groups,OU=PESCO_FARMINGTON," + BASE_DN
OTHER_DN = "OU=Groups,OU=PESCO_FARMINGTON," + BASE_DN


@dataclass
class FakeUser:
    sam: str
    display: str
    emp_no: str | None
    groups: list[str]
    password: str = FAKE_PASSWORD
    disabled: bool = False
    guid: uuid.UUID = field(default_factory=uuid.uuid4)

    @property
    def dn(self) -> str:
        return f"CN={self.display},OU=Users,OU=PESCO_FARMINGTON,DC=pstest,DC=local"

    @property
    def upn(self) -> str:
        return f"{self.sam}@pstest.local"


def default_users() -> list[FakeUser]:
    """The same people as the CI domain controller (deploy/samba)."""
    return [
        FakeUser("admin.ann", "Ann Admin", "1001", ["PS-Admins"]),
        FakeUser("clerk.clara", "Clara Clerk", "4001", ["PS-Clerical"]),
        FakeUser("clerk.leo", "Leo Lead", "4002", ["PS-Clerical-Supervisor"]),
        FakeUser("wh.wade", "Wade Warehouse", "6001", ["PS-Warehouse"]),
        FakeUser("qc.quinn", "Quinn QC", "5001", ["PS-QC"]),
        FakeUser("pm.paula", "Paula PM", "7001", ["PS-ProjectMgmt"]),
        FakeUser("weld.andy", "Andy Welder", "3003", ["PS-03-WeldParts-Worker"]),
        FakeUser("lead.lou", "Lou Lead", "3002", ["PS-03-WeldParts-Lead"]),
        FakeUser("sup.wes", "Wes Super", "2003", ["PS-03-WeldParts-Supervisor"]),
        FakeUser("multi.mia", "Mia Multi", "2006", ["PS-Clerical", "PS-06-CodeBldg2-Supervisor"]),
        FakeUser("nested.ned", "Ned Nested", "3010", [NESTED_GROUP]),
        FakeUser("both.bo", "Bo Both", "3011", ["PS-03-WeldParts-Worker", "PS-03-WeldParts-Supervisor"]),
        FakeUser("disabled.dan", "Dan Disabled", "3004", ["PS-03-WeldParts-Worker"], disabled=True),
        FakeUser("noemp.nora", "Nora NoEmp", None, ["PS-Warehouse"]),
        FakeUser("dup.dave", "Dave Dup", "9001", ["PS-03-WeldParts-Worker"]),
        FakeUser("dup.dina", "Dina Dup", "9001", ["PS-Warehouse"]),
        FakeUser("norole.nick", "Nick NoRole", "8001", []),
    ]


class FakeDirectory:
    """Users, groups in the PacketStation OU, and nested groups outside it."""

    def __init__(
        self,
        users: list[FakeUser] | None = None,
        ou_groups: list[str] | None = None,
        nested: dict[str, list[str]] | None = None,
        other_groups: dict[str, str] | None = None,
    ):
        self.users = users if users is not None else default_users()
        if ou_groups is None:
            ou_groups = [g for g in group_names() if g != "PS-14-Electrical-Lead"] + ["PS-Extra-Legacy"]
        self.ou_groups = ou_groups
        self.nested = nested if nested is not None else {NESTED_GROUP: ["PS-10-WeldAssembly-Worker"]}
        # Groups outside the OU, by name, with a description (the nested helper group and a mail group)
        self.other_groups = (
            other_groups
            if other_groups is not None
            else {NESTED_GROUP: "Helper group for the nested membership test", "IT-Helpdesk": "IT help desk"}
        )
        self.down = False
        self.binds = 0
        self.searches = 0

    # ------------------------------------------------ helpers

    def user(self, sam: str) -> FakeUser:
        return next(u for u in self.users if u.sam == sam)

    def _check(self) -> None:
        if self.down:
            raise LdapUnavailable("fake directory is down")
        self.searches += 1

    def _ad(self, u: FakeUser) -> AdUser:
        return AdUser(u.guid, u.dn, u.sam, u.upn, u.display, u.emp_no, f"{u.sam}@example.test", u.disabled)

    def _closure(self, u: FakeUser) -> set[str]:
        seen: set[str] = set()
        todo = list(u.groups)
        while todo:
            g = todo.pop()
            if g.lower() in seen:
                continue
            seen.add(g.lower())
            todo.extend(self.nested.get(g, []))
        return seen

    def group_dn(self, name: str) -> str:
        """The DN of a group: in the PacketStation OU, or under Groups for the ones outside it."""
        if any(g.lower() == name.lower() for g in self.ou_groups):
            return f"CN={name},{OU_DN}"
        return f"CN={name},{OTHER_DN}"

    def _group_by_dn(self, dn: str) -> str | None:
        for name in [*self.ou_groups, *self.other_groups]:
            if self.group_dn(name).lower() == dn.strip().lower():
                return name
        return None

    def _in_ou(self, names: set[str]) -> list[str]:
        return sorted(g for g in self.ou_groups if g.lower() in names)

    # ------------------------------------------------ protocol

    def find_user(self, login: str) -> AdUser | None:
        self._check()
        key = login.strip().lower()
        for u in self.users:
            if key in (u.sam.lower(), u.upn.lower()):
                return self._ad(u)
        return None

    def get_user_by_guid(self, guid: uuid.UUID) -> AdUser | None:
        self._check()
        for u in self.users:
            if u.guid == guid:
                return self._ad(u)
        return None

    def verify_password(self, dn: str, password: str) -> bool:
        if self.down:
            raise LdapUnavailable("fake directory is down")
        if not password:
            return False
        self.binds += 1
        return any(u.dn == dn and u.password == password for u in self.users)

    def groups_of(self, dn: str) -> list[str]:
        self._check()
        u = next((x for x in self.users if x.dn == dn), None)
        return [] if u is None else self._in_ou(self._closure(u))

    def list_groups(self) -> list[str]:
        self._check()
        return sorted(self.ou_groups)

    def members_of(self, group_cn: str) -> list[AdUser]:
        self._check()
        if group_cn.lower() not in {g.lower() for g in self.ou_groups}:
            return []
        return [self._ad(u) for u in self.users if group_cn.lower() in self._closure(u)]

    def list_groups_detail(self) -> list[tuple[str, str]]:
        self._check()
        return sorted(((g, self.group_dn(g)) for g in self.ou_groups), key=lambda t: t[0].lower())

    def members_of_dn(self, group_dn: str) -> list[AdUser]:
        self._check()
        name = self._group_by_dn(group_dn)
        if name is None:
            return []
        return [self._ad(u) for u in self.users if name.lower() in self._closure(u)]

    def search_groups(self, q: str, limit: int = 25) -> list[dict[str, str]]:
        self._check()
        key = q.strip().lower()
        if not key:
            return []
        out = []
        for name in sorted({*self.ou_groups, *self.other_groups}, key=str.lower):
            desc = self.other_groups.get(name, f"Packet Station: {name}")
            if key in name.lower() or key in desc.lower():
                out.append({"dn": self.group_dn(name), "name": name, "description": desc})
        return out[: max(1, limit)]

    def users_by_guid(self, guids: list[uuid.UUID]) -> dict[uuid.UUID, AdUser | None]:
        self._check()
        by = {u.guid: u for u in self.users}
        return {g: (self._ad(by[g]) if g in by else None) for g in guids}