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"


@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,
    ):
        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"]}
        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 _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)]