Pas: Extends permission system

src/onegov/pas/__init__.py

@@ -6,7 +6,6 @@
 log.addHandler(logging.NullHandler())
 
 from onegov.pas.i18n import _
-
 from onegov.pas.app import PasApp
 
 # Import cronjobs to register the decorators
@@ -15,5 +14,6 @@
 __all__ = (
     '_',
     'log',
-    'PasApp', 'cronjobs'
+    'PasApp',
+    'cronjobs'
 )

src/onegov/pas/app.py

@@ -4,19 +4,24 @@
 from onegov.pas.content import create_new_organisation
 from onegov.pas.custom import get_global_tools
 from onegov.pas.custom import get_top_navigation
+from onegov.pas.request import PasRequest
 from onegov.pas.theme import PasTheme
 from onegov.town6 import TownApp
 from onegov.town6.app import get_i18n_localedirs as get_i18n_localedirs_base
+from purl import URL
+from onegov.org.models import Organisation
+
 
 from typing import Any, TYPE_CHECKING
 if TYPE_CHECKING:
     from collections.abc import Callable, Iterator
     from onegov.core.types import RenderData
-    from onegov.org.models import Organisation
-    from onegov.town6.request import TownRequest
+    from morepath.authentication import NoIdentity
+    from morepath.authentication import Identity
 
 
 class PasApp(TownApp):
+    request_class = PasRequest
 
     def configure_kub_api(
         self,
@@ -33,21 +38,32 @@ def configure_kub_api(
         self.kub_api_token = kub_api_token
         self.kub_base_url = kub_base_url
 
+    def redirect_after_login(
+        self,
+        identity: Identity | NoIdentity,
+        request: PasRequest,  # type:ignore[override]
+        default: str
+    ) -> str | None:
+
+        if default != '/' and '/auth/login' not in str(default):
+            return None
+        return URL(request.class_link(Organisation)).path()
+
 
 @PasApp.setting(section='org', name='create_new_organisation')
 def get_create_new_organisation_factory(
 ) -> Callable[[TownApp, str], Organisation]:
     return create_new_organisation
 
 
-# NOTE: Feriennet doesn't need a citizen login
+# NOTE: PAS doesn't need a citizen login
 @PasApp.setting(section='org', name='citizen_login_enabled')
 def get_citizen_login_enabled() -> bool:
     return False
 
 
 @PasApp.template_variables()
-def get_template_variables(request: TownRequest) -> RenderData:
+def get_template_variables(request: PasRequest) -> RenderData:
     return {
         'global_tools': tuple(get_global_tools(request)),
         'top_navigation': tuple(get_top_navigation(request)),

src/onegov/pas/cli.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import click
+import transaction
 import logging
 import requests
 import urllib3
 import warnings
 from onegov.core.cli import command_group
+from onegov.pas.collections.parliamentarian import PASParliamentarianCollection
 from onegov.pas.excel_header_constants import (
     commission_expected_headers_variant_1,
     commission_expected_headers_variant_2,
@@ -22,10 +24,10 @@
 if TYPE_CHECKING:
     from collections.abc import Callable
     from onegov.pas.app import PasApp
-    from onegov.pas.app import TownRequest
+    from onegov.pas.request import PasRequest
     from typing import TypeAlias
 
-    Processor: TypeAlias = Callable[[TownRequest, PasApp], None]
+    Processor: TypeAlias = Callable[[PasRequest, PasApp], None]
 
 
 log = logging.getLogger('onegov.org.cli')
@@ -45,7 +47,10 @@
 def import_commission_data(
     excel_file: str,
 ) -> Processor:
-    """Import commission data from an Excel or csv file.
+    """
+    Note: This is deprecated, not really used, as we have the JSON import.
+
+    Import commission data from an Excel or csv file.
 
     Assumes that the name of the commission is the filename.
 
@@ -57,7 +62,7 @@ def import_commission_data(
             "Kommission_Gesundheit_und_Soziales.xlsx"
     """
 
-    def import_data(request: TownRequest, app: PasApp) -> None:
+    def import_data(request: PasRequest, app: PasApp) -> None:
 
         try:
             import_commissions_excel(
@@ -80,6 +85,30 @@ def import_data(request: TownRequest, app: PasApp) -> None:
     return import_data
 
 
+@cli.command(name='update-accounts', context_settings={'singular': True})
+@click.option('--dry-run/-no-dry-run', default=False)
+def update_accounts_cli(dry_run: bool) -> Processor:
+    """ Updates user accounts for parliamentarians. """
+
+    def do_update_accounts(request: PasRequest, app: PasApp) -> None:
+
+        parliamentarians = PASParliamentarianCollection(app)
+        for parliamentarian in parliamentarians.query():
+            if not parliamentarian.email_primary:
+                click.echo(
+                    f'Skipping {parliamentarian.title}, no primary email.'
+                )
+                continue
+            parliamentarians.update_user(
+                parliamentarian, parliamentarian.email_primary
+            )
+
+        if dry_run:
+            transaction.abort()
+
+    return do_update_accounts
+
+
 @cli.command('check-api')
 @click.option('--url', default='',
               help='API endpoint to check')
@@ -147,7 +176,7 @@ def import_kub_data(
             --token "your-token-here" --max-workers 5
     """
 
-    def cli_wrapper(request: TownRequest, app: PasApp) -> None:
+    def cli_wrapper(request: PasRequest, app: PasApp) -> None:
         """CLI wrapper that calls the orchestrator."""
         # Create composite output handler for both CLI and database
         click_handler = ClickOutputHandler()
@@ -221,7 +250,7 @@ def update_custom_data(
             --max-workers 5
     """
 
-    def update_data(request: TownRequest, app: PasApp) -> None:
+    def update_data(request: PasRequest, app: PasApp) -> None:
         # Create composite output handler for both CLI and database
         click_handler = ClickOutputHandler()
         db_handler = DatabaseOutputHandler()

src/onegov/pas/collections/attendence.py

@@ -16,6 +16,7 @@
 if TYPE_CHECKING:
     from datetime import date
     from sqlalchemy.orm import Query, Session
+    from onegov.pas.request import PasRequest
 
 
 class AttendenceCollection(GenericCollection[Attendence]):
@@ -138,3 +139,69 @@ def by_party(
             commission_id=self.commission_id,
             party_id=party_id,
         )
+
+    def for_parliamentarian(self, parliamentarian_id: str) -> Self:
+        """Returns attendances for a specific parliamentarian only."""
+        return self.for_filter(parliamentarian_id=parliamentarian_id)
+
+    def for_commission_president(
+        self,
+        parliamentarian_id: str,
+        active_commission_ids: list[str]
+    ) -> Query[Attendence]:
+        """
+        Returns attendances for a commission president:
+        - Their own attendances
+        - Attendances of members in commissions they preside over
+        """
+        query = self.query()
+        return query.filter(
+            (Attendence.parliamentarian_id == parliamentarian_id) |
+            (Attendence.commission_id.in_(active_commission_ids))
+        )
+
+    def view_for_parliamentarian(
+        self, request: PasRequest
+    ) -> list[Attendence]:
+        """
+        Returns filtered attendances based on user role and permissions.
+        This encapsulates the filtering logic previously in the view.
+        """
+        user = request.current_user
+
+        if not request.is_parliamentarian:
+            # Admins see all attendances
+            return self.query().all()
+
+        if not (user and hasattr(user, 'parliamentarian') and
+                user.parliamentarian):
+            return []
+
+        parliamentarian = user.parliamentarian
+
+        if user.role == 'commission_president':
+            from datetime import date
+            # Commission presidents see own + commission members' attendances
+            # We have to check all but usually president of just one
+            active_presidencies = [
+                cm.commission_id
+                for cm in parliamentarian.commission_memberships
+                if (cm.role == 'president' and
+                    (cm.end is None or cm.end >= date.today()))
+            ]
+
+            if active_presidencies:
+                return self.for_commission_president(
+                    str(parliamentarian.id),
+                    active_presidencies
+                ).all()
+            else:
+                # Fallback to own attendances only
+                return self.for_parliamentarian(
+                    str(parliamentarian.id)
+                ).query().all()
+        else:
+            # Regular parliamentarians see only their own attendances
+            return self.for_parliamentarian(
+                str(parliamentarian.id)
+            ).query().all()

src/onegov/pas/collections/parliamentarian.py

@@ -1,13 +1,137 @@
 from __future__ import annotations
 
+import logging
+from onegov.core.utils import toggle
+from onegov.core.crypto import random_password
 from onegov.parliament.collections import ParliamentarianCollection
 from onegov.pas.models import PASParliamentarian
+from onegov.user import UserCollection
+
+log = logging.getLogger('onegov.pas.collections.parliamentarian')
+
+
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from typing import Any, Self
+    from onegov.core import Framework
 
 
 class PASParliamentarianCollection(
     ParliamentarianCollection[PASParliamentarian]
 ):
 
+    def __init__(self, app: Framework, **kwargs: Any) -> None:
+        super().__init__(app.session(), **kwargs)
+        self.app = app
+
+    def for_filter(
+        self,
+        active: bool | None = None,
+        party: str | None = None,
+    ) -> Self:
+        active_ = toggle(self.active, active)
+        party_ = toggle(self.party, party)
+
+        return self.__class__(
+            self.app,
+            active=active_,
+            party=party_
+        )
+
     @property
     def model_class(self) -> type[PASParliamentarian]:
         return PASParliamentarian
+
+    def add(self, **kwargs: Any) -> PASParliamentarian:
+        item = super().add(**kwargs)
+        if not item.email_primary:
+            log.warning(
+                f'Creating parliamentarian {item.title} without'
+                'email_primary. This will prevent user account'
+                'creation and may cause permission-related failures.'
+            )
+        self.update_user(item, item.email_primary)
+        self.session.flush()
+        return item
+
+    def delete(self, item: PASParliamentarian) -> None:
+        self.update_user(item, None)
+        self.session.delete(item)
+        self.session.flush()
+
+    def update_user(
+        self,
+        item: PASParliamentarian,
+        new_email: str | None
+    ) -> None:
+        """ Keep the parliamentarian and its user account in sync.
+
+        * Creates a new user account if an email address is set (if not already
+          existing).
+        * Disable user accounts if an email has been deleted.
+        * Change usernames if an email has changed.
+        * Make sure used user accounts have the right role.
+        * Make sure used user accounts are activated.
+        * Make sure the password is changed if activated or disabled.
+
+        """
+
+        old_email = item.email_primary
+        users = UserCollection(self.session)
+        old_user = users.by_username(old_email) if old_email else None
+        new_user = users.by_username(new_email) if new_email else None
+        create = False
+        enable = None
+        disable = []
+
+        if not new_email:
+            # email has been unset: disable obsolete users
+            disable.extend([old_user, new_user])
+        else:
+            if new_email == old_email:
+                # email has not changed, old_user == new_user
+                if not old_user:
+                    create = True
+                else:
+                    enable = old_user
+            else:
+                # email has changed: ensure user exist
+                if old_user and new_user:
+                    disable.append(old_user)
+                    enable = new_user
+                elif not old_user and not new_user:
+                    create = True
+                else:
+                    enable = old_user if old_user else new_user
+
+        if create:
+            assert new_email is not None
+            log.info(f'Creating user {new_email}')
+            users.add(
+                new_email, random_password(16), role='parliamentarian',
+                realname=item.title
+            )
+
+        if enable:
+            corrections = {
+                'username': new_email,
+                'role': 'parliamentarian',
+                'active': True,
+                'source': None,
+                'source_id': None
+            }
+            corrections = {
+                attribute: value for attribute, value in corrections.items()
+                if getattr(enable, attribute) != value
+            }
+            if corrections:
+                log.info(f'Correcting user {enable.username} to {corrections}')
+                for attribute, value in corrections.items():
+                    setattr(enable, attribute, value)
+                enable.logout_all_sessions(self.app)
+
+        for user in disable:
+            if user:
+                log.info(f'Deactivating user {user.username}')
+                user.active = False
+                user.logout_all_sessions(self.app)