net-kingdom/identity-provisioner/tests/test_service_groups.py

import unittest

from provisioner import LLDAPProvisioner, dispatch


class Directory(LLDAPProvisioner):
    """In-memory LLDAP: users, groups and memberships."""

    def __init__(self):
        self.users = {"tegwick", "agent-codex"}
        self.groups = {"lldap_admin": 1, "tenant:trial:one:users": 2, "forge-admins": 3}
        self.members = {1: {"tegwick"}, 2: set(), 3: set()}
        self.calls = []
        self.drop_add = False

    def _login(self):
        return "test-token"

    def _directory(self, token):
        return ([{"id": u} for u in sorted(self.users)],
                [{"id": i, "displayName": n} for n, i in self.groups.items()])

    def _user(self, token, subject):
        return {"id": subject, "groups": [{"id": i, "displayName": n} for n, i in self.groups.items()
                                           if subject in self.members[i]]}

    def _ensure_group(self, token, groups, name):
        if name not in self.groups:
            self.calls.append(("create", name))
            self.groups[name] = max(self.groups.values()) + 1
            self.members[self.groups[name]] = set()
        return self.groups[name]

    def _add_group(self, token, username, group_id):
        self.calls.append(("add", username, group_id))
        if not self.drop_add:
            self.members[group_id].add(username)

    def _remove_group(self, token, username, group_id):
        self.calls.append(("remove", username, group_id))
        self.members[group_id].discard(username)

    def _group_members(self, token, group_id):
        return sorted(self.members[group_id])

    def _gql(self, token, query, variables):
        if "deleteGroup" in query:
            self.calls.append(("delete", variables["id"]))
            name = next(n for n, i in self.groups.items() if i == variables["id"])
            del self.groups[name]
            return {"deleteGroup": {"ok": True}}
        raise AssertionError(query)


def req(**kw):
    p = {"groups": ["forge-admins", "forge-coulomb-owners", "forge-coulomb-readers"],
         "members": {"forge-admins": ["tegwick"], "forge-coulomb-owners": ["tegwick"]},
         "idempotency_key": "forgejo-netkingdom-oidc-1", "correlation_id": "test"}
    p.update(kw)
    return p


class EnsureTests(unittest.TestCase):
    def test_creates_missing_groups_and_adds_members(self):
        d = Directory()
        r = dispatch(d, "/v1/groups/ensure", req())
        self.assertEqual("groups_ensured", r.status)
        self.assertFalse(r.resumed)
        self.assertEqual({"forge-coulomb-owners", "forge-coulomb-readers"},
                         {c[1] for c in d.calls if c[0] == "create"})
        self.assertIn("tegwick", d.members[d.groups["forge-admins"]])
        self.assertIn("tegwick", d.members[d.groups["forge-coulomb-owners"]])

    def test_idempotent_and_reports_resumed(self):
        d = Directory()
        dispatch(d, "/v1/groups/ensure", req())
        d.calls.clear()
        r = dispatch(d, "/v1/groups/ensure", req())
        self.assertTrue(r.resumed)
        self.assertEqual([], [c for c in d.calls if c[0] == "create"])

    def test_never_removes_other_memberships(self):
        d = Directory()
        d.members[3].add("agent-codex")
        dispatch(d, "/v1/groups/ensure", req())
        self.assertIn("agent-codex", d.members[3])
        self.assertEqual([], [c for c in d.calls if c[0] == "remove"])

    def test_reserved_and_invalid_requests_change_nothing(self):
        bad = [req(groups=["lldap_admin"]), req(groups=["tenant-x"]), req(groups=["netkingdom-suspended"]),
               req(groups=["Forge Admins"]), req(groups=[]),
               req(members={"not-listed": ["tegwick"]}),
               req(members={"forge-admins": ["Not A User"]}),
               req(members={"forge-admins": ["nobody"]})]
        for payload in bad:
            d = Directory()
            with self.assertRaises(ValueError, msg=payload):
                dispatch(d, "/v1/groups/ensure", payload)
            self.assertEqual([], d.calls, payload)

    def test_failed_readback_never_reports_success(self):
        d = Directory()
        d.drop_add = True
        with self.assertRaises(RuntimeError):
            dispatch(d, "/v1/groups/ensure", req())


class RevokeTests(unittest.TestCase):
    def test_removes_only_listed_memberships_and_keeps_groups_by_default(self):
        d = Directory()
        dispatch(d, "/v1/groups/ensure", req())
        d.members[d.groups["forge-admins"]].add("agent-codex")
        r = dispatch(d, "/v1/groups/revoke", req())
        self.assertEqual("groups_revoked", r.status)
        self.assertEqual({"agent-codex"}, d.members[d.groups["forge-admins"]])
        self.assertIn("forge-coulomb-owners", d.groups)

    def test_deletes_only_empty_listed_groups_when_asked(self):
        d = Directory()
        dispatch(d, "/v1/groups/ensure", req())
        d.members[d.groups["forge-admins"]].add("agent-codex")
        r = dispatch(d, "/v1/groups/revoke", req(delete_if_empty=True))
        self.assertEqual("groups_revoked_and_deleted", r.status)
        self.assertIn("forge-admins", d.groups)
        self.assertNotIn("forge-coulomb-owners", d.groups)
        self.assertNotIn("forge-coulomb-readers", d.groups)
        self.assertIn("lldap_admin", d.groups)


if __name__ == "__main__":
    unittest.main()