[SDESK-7459] Implement ability for a Module to register new privileges (

#2813)

* Implement privilege registry for async app

SDESK-7459

* Cleanup code and add tests

SDESK-7459

* Remove not needed yield/teardown

SDESK-7459

superdesk/core/app.py

@@ -12,7 +12,9 @@
 import importlib
 
 from superdesk.core.types import WSGIApp
+
 from .auth.user_auth import UserAuthProtocol
+from .privileges import PrivilegesRegistry
 
 
 def get_app_config(key: str, default: Optional[Any] = None) -> Optional[Any]:
@@ -47,6 +49,8 @@ class SuperdeskAsyncApp:
 
     auth: UserAuthProtocol
 
+    privileges: PrivilegesRegistry
+
     def __init__(self, wsgi: WSGIApp):
         self._running = False
         self._imported_modules = {}
@@ -57,6 +61,7 @@ def __init__(self, wsgi: WSGIApp):
         self.resources = Resources(self)
         self.auth = self.load_auth_module()
         self._store_app()
+        self.privileges = PrivilegesRegistry()
 
     @property
     def running(self) -> bool:
@@ -142,11 +147,15 @@ def _load_modules(self, paths: List[str | tuple[str, dict]]):
             for endpoint in module.endpoints or []:
                 self.wsgi.register_endpoint(endpoint)
 
-        # then init all modules
+        # init all modules
         for module in self.get_module_list():
             if module.init is not None:
                 module.init(self)
 
+            # then register all module privileges
+            for privilege in module.privileges or []:
+                self.privileges.add(privilege)
+
     def start(self):
         """Start the app
 
@@ -163,6 +172,9 @@ def start(self):
         self._load_modules(self.wsgi.config.get("MODULES", []))
         self._running = True
 
+        # after app is running is longer possible to add more privileges
+        self.privileges.lock()
+
     def stop(self):
         """Stops the app
 

superdesk/core/module.py

@@ -11,9 +11,11 @@
 from typing import Callable, Optional, List, Union
 from dataclasses import dataclass
 
-from .config import ConfigModel
 from superdesk.core.types import Endpoint, EndpointGroup
 
+from .config import ConfigModel
+from .privileges import Privilege
+
 
 @dataclass
 class Module:
@@ -50,6 +52,9 @@ class Module:
     #: Optional list of HTTP endpoints to register with the system
     endpoints: Optional[List[Union[Endpoint, EndpointGroup]]] = None
 
+    # Optional list of privileges to register
+    privileges: list[Privilege] | None = None
+
 
 from .app import SuperdeskAsyncApp  # noqa: E402
 from .resources import ResourceConfig  # noqa: E402

superdesk/core/privileges.py

@@ -0,0 +1,45 @@
+from dataclasses import dataclass
+from quart_babel.speaklater import LazyString
+
+
+@dataclass(frozen=True)
+class Privilege:
+    name: str
+    label: LazyString | None = None
+    category: LazyString | None = None
+    description: LazyString | None = None
+
+
+class PrivilegesRegistry:
+    __privileges: set[Privilege] | frozenset[Privilege]
+
+    def __init__(self):
+        self.__privileges = set()
+
+    def add(self, privilege: Privilege):
+        """Add a privilege if the registry is not locked."""
+
+        if self.is_locked:
+            raise RuntimeError("Cannot add privileges after the app has started.")
+
+        self.__privileges.add(privilege)  # type: ignore[union-attr]
+
+    @property
+    def is_locked(self):
+        return isinstance(self.__privileges, frozenset)
+
+    def lock(self):
+        """Lock the registry by converting the privileges to a frozenset."""
+
+        if not self.is_locked:
+            self.__privileges = frozenset(self.__privileges)
+
+    def get_all(self) -> list[Privilege]:
+        """Retrieve all privileges."""
+
+        return list(self.__privileges)
+
+    def __contains__(self, name: str) -> bool:
+        """Check if a privilege exists."""
+
+        return any(priv.name == name for priv in self.__privileges)

tests/core/app_test.py

@@ -1,6 +1,7 @@
 from unittest import TestCase, mock
 
 from superdesk.core.app import SuperdeskAsyncApp
+from superdesk.core.privileges import Privilege
 from superdesk.tests import MockWSGI
 
 
@@ -56,3 +57,17 @@ def test_module_init(self, init):
         app = SuperdeskAsyncApp(MockWSGI(config={"MODULES": ["tests.core.modules.a"]}))
         app.start()
         init.assert_called_once_with(app)
+
+    def test_register_privileges(self):
+        app = SuperdeskAsyncApp(MockWSGI(config={"MODULES": ["tests.core.modules.module_with_privileges"]}))
+        app.start()
+
+        # check privileges are registered
+        registered_privileges = app.privileges.get_all()
+        self.assertEqual(len(registered_privileges), 2)
+        self.assertTrue(app.privileges.is_locked)
+
+        # after app is started, trying to register a privilege raise an exception
+        with self.assertRaises(RuntimeError) as exc:
+            app.privileges.add(Privilege(name="After app started"))
+            self.assertEqual(exc.msg, "Cannot add privileges after the app has started")

tests/core/modules/module_with_privileges.py

@@ -0,0 +1,12 @@
+from quart_babel import lazy_gettext
+
+from superdesk.core.module import Module
+from superdesk.core.privileges import Privilege
+
+module = Module(
+    name="tests.module_with_privileges",
+    privileges=[
+        Privilege(name="can_test", description=lazy_gettext("Test privilege can test")),
+        Privilege(name="can_play", description=lazy_gettext("Test privilege can play")),
+    ],
+)

tests/core/privileges_registry_test.py

@@ -0,0 +1,61 @@
+import pytest
+from superdesk.core.privileges import PrivilegesRegistry, Privilege
+
+
+@pytest.fixture(scope="function")
+def registry():
+    return PrivilegesRegistry()
+
+
+def test_add_privilege_before_lock(registry):
+    privilege = Privilege(name="edit_article")
+
+    registry.add(privilege)
+
+    assert "edit_article" in registry
+    assert registry.get_all() == [privilege]
+
+
+def test_add_multiple_privileges(registry):
+    privilege1 = Privilege(name="edit_article")
+    privilege2 = Privilege(name="delete_article")
+
+    registry.add(privilege1)
+    registry.add(privilege2)
+
+    assert "edit_article" in registry
+    assert "delete_article" in registry
+    assert len(registry.get_all()) == 2
+
+
+def test_lock_registry(registry):
+    privilege = Privilege(name="edit_article")
+
+    registry.add(privilege)
+    registry.lock()
+
+    assert registry.is_locked
+    assert "edit_article" in registry
+
+
+def test_cannot_add_after_lock(registry):
+    privilege = Privilege(name="edit_article")
+
+    registry.add(privilege)
+    registry.lock()
+
+    with pytest.raises(RuntimeError):
+        registry.add(Privilege(name="new_privilege"))
+
+
+def test_lock_prevents_further_additions(registry):
+    privilege = Privilege(name="edit_article")
+
+    registry.add(privilege)
+    registry.lock()
+
+    with pytest.raises(RuntimeError, match="Cannot add privileges after the app has started."):
+        registry.add(Privilege(name="new_privilege"))
+
+    assert len(registry.get_all()) == 1
+    assert registry.is_locked