Skip to content

Core: Add audit trail. - #2614

Open
cyrillkuettel wants to merge 13 commits into
masterfrom
ogc-180-audit-trail-add-versioning-insert-update-or-delete-to-ogc
Open

Core: Add audit trail.#2614
cyrillkuettel wants to merge 13 commits into
masterfrom
ogc-180-audit-trail-add-versioning-insert-update-or-delete-to-ogc

Conversation

@cyrillkuettel

@cyrillkuettel cyrillkuettel commented Aug 6, 2026

Copy link
Copy Markdown
Contributor

Please fill in the commit message below and work through the checklist. You can delete parts that are not needed, e.g. the optional description, the link to a ticket or irrelevant options of the checklist.

Commit message

Core: Add audit trail.

TYPE: Feature
LINK: OGC-180

Checklist

  • I considered adding a reviewer
  • I have updated the PO files
  • I have tested my code thoroughly by hand
  • I have added tests for my changes/features

@linear

linear Bot commented Aug 6, 2026

Copy link
Copy Markdown

OGC-180

@codecov

codecov Bot commented Aug 6, 2026

Copy link
Copy Markdown

❌ 13 Tests Failed:

Tests completed Failed Passed Skipped
2528 13 2515 16
View the top 3 failed test(s) by shortest run time
tests/onegov/org/test_cronjobs.py::test_admin_notification_signing_disabled
Stack Traces | 1.43s run time
client = <tests.onegov.org.conftest.Client object at 0x7f573791cfc0>

    def test_admin_notification_signing_disabled(
        client: Client['TestOrgApp'],
    ) -> None:
        """
        With ``enable_notification_pdf_signing`` turned off, the signing service
        is never invoked and the pdf is attached unsigned.
        """
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
    
        real_now = utcnow()
    
        transaction.begin()
        directory = _make_permit_directory(client.app.session())
        directory.enable_notification_pdf_signing = False
        directory.add(
            values=dict(
                gesuchsteller_in='Clara Meier',
                adresse='Ringstrasse 9',
                publication_start=real_now - timedelta(minutes=30),
                publication_end=real_now + timedelta(hours=2),
            )
        )
        transaction.commit()
        close_all_sessions()
    
        with patch.object(SwisscomAIS, 'sign') as sign:
            # the publication start has been crossed ...
            with freeze_time(real_now, tick=True):
                client.get(get_cronjob_url(job))
    
            # ... and later on the publication end
            with freeze_time(real_now + timedelta(hours=3), tick=True):
                client.get(get_cronjob_url(job))
    
            # signing must not be attempted at all
            assert sign.call_count == 0
    
>       assert len(os.listdir(client.app.maildir)) == 2
E       AssertionError: assert 1 == 2
E        +  where 1 = len(['0.1.1787564744.398084'])
E        +    where ['0.1.1787564744.398084'] = <built-in function listdir>('....../tmp/tmp3nn0q_3x/mails')
E        +      where <built-in function listdir> = os.listdir
E        +      and   '....../tmp/tmp3nn0q_3x/mails' = <tests.onegov.org.conftest.TestOrgApp object at 0x7f57374ba350>.maildir
E        +        where <tests.onegov.org.conftest.TestOrgApp object at 0x7f57374ba350> = <tests.onegov.org.conftest.Client object at 0x7f573791cfc0>.app

.../onegov/org/test_cronjobs.py:3624: AssertionError
tests/onegov/org/test_cronjobs.py::test_send_email_notification_for_recent_directory_entry_publications[member]
Stack Traces | 1.52s run time
client = <tests.onegov.org.conftest.Client object at 0x7f573765e750>
access = 'member'

    @pytest.mark.parametrize(
        'access',
        ('private', 'member', 'mtan', 'secret', 'secret_mtan', 'public')
    )
    def test_send_email_notification_for_recent_directory_entry_publications(
        client: Client[TestOrgApp],
        access: str
    ) -> None:
    
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
        tz = ensure_timezone('Europe/Zurich')
    
        def planauflagen() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('offentliche-planauflage'))
    
        def sport_clubs() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('sport-clubs'))
    
        def count_recipients() -> int:
            return (EntryRecipientCollection(client.app.session()).query()
                    .filter_by(directory_id=planauflagen().id)
                    .filter_by(confirmed=True).count())
    
        assert len(os.listdir(client.app.maildir)) == 0
    
        transaction.begin()
    
        directories: DirectoryCollection[ExtendedDirectory]
        directories = DirectoryCollection(client.app.session(), type='extended')
        planauflage = directories.add(
            title='Öffentliche Planauflage',
            structure="""
                Gesuchsteller/in *= ___
                Grundeigentümer/in *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Gesuchsteller/in]",
                order=['Gesuchsteller/in'],
                searchable=['title'],
            ),
            enable_update_notifications=True,
        )
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Carmine Carminio',
            grundeigentumer_in='Doris Dorinio',
            publication_start=datetime(2025, 1, 6, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 1, 30, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Emil Emilio',
            grundeigentumer_in='Franco Francinio',
            publication_start=datetime(2025, 1, 8, 6, 1, tzinfo=tz),
            publication_end=datetime(2025, 1, 31, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        sport_club = directories.add(
            title='Sport Clubs',
            structure="""
                Name *= ___
                Category *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Name]",
                order=['Name'],
                searchable=['title']
            ),
            enable_update_notifications=False,
        )
        entry = sport_club.add(values=dict(
            name='Wanderfreunde',
            category='Hiking',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 22, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = sport_club.add(values=dict(
            name='Pokerfreunde',
            category='Games',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 2, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        EntryRecipientCollection(client.app.session()).add(
            directory_id=planauflage.id,
            address='john@doe.ch',
            confirmed=True
        )
        EntryRecipientCollection(client.app.session()).add(
            directory_id=sport_club.id,
            address='john@doe.ch',
            confirmed=True
        )
    
        transaction.commit()
        close_all_sessions()
    
        assert count_recipients() == 1
        john = EntryRecipientCollection(client.app.session()).query().first()
        assert john is not None
    
        assert client.app.org.meta.get('hourly_maintenance_tasks_last_run') is None
    
        with freeze_time(datetime(2025, 1, 1, 4, 0, tzinfo=tz)):
            client.get(get_cronjob_url(job))
    
            assert len(os.listdir(client.app.maildir)) == 0
>           assert client.app.org.meta.get('hourly_maintenance_tasks_last_run')
                   ^^^^^^^^^^^^^^

.../onegov/org/test_cronjobs.py:2165: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7f5731b6acf0>
state = <sqlalchemy.orm.state.InstanceState object at 0x7f5735af3820>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7f5735af3820>, 'theme_options': {'primary-color': '#006fba'}, 'name': 'Govikon', 'created': datetime.datetime(2026, 8, 24, 9, 45, 33, 961373, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/org/test_cronjobs.py::test_send_email_notification_for_recent_directory_entry_publications[public]
Stack Traces | 1.59s run time
client = <tests.onegov.org.conftest.Client object at 0x7fc04e3e3ad0>
access = 'public'

    @pytest.mark.parametrize(
        'access',
        ('private', 'member', 'mtan', 'secret', 'secret_mtan', 'public')
    )
    def test_send_email_notification_for_recent_directory_entry_publications(
        client: Client[TestOrgApp],
        access: str
    ) -> None:
    
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
        tz = ensure_timezone('Europe/Zurich')
    
        def planauflagen() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('offentliche-planauflage'))
    
        def sport_clubs() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('sport-clubs'))
    
        def count_recipients() -> int:
            return (EntryRecipientCollection(client.app.session()).query()
                    .filter_by(directory_id=planauflagen().id)
                    .filter_by(confirmed=True).count())
    
        assert len(os.listdir(client.app.maildir)) == 0
    
        transaction.begin()
    
        directories: DirectoryCollection[ExtendedDirectory]
        directories = DirectoryCollection(client.app.session(), type='extended')
        planauflage = directories.add(
            title='Öffentliche Planauflage',
            structure="""
                Gesuchsteller/in *= ___
                Grundeigentümer/in *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Gesuchsteller/in]",
                order=['Gesuchsteller/in'],
                searchable=['title'],
            ),
            enable_update_notifications=True,
        )
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Carmine Carminio',
            grundeigentumer_in='Doris Dorinio',
            publication_start=datetime(2025, 1, 6, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 1, 30, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Emil Emilio',
            grundeigentumer_in='Franco Francinio',
            publication_start=datetime(2025, 1, 8, 6, 1, tzinfo=tz),
            publication_end=datetime(2025, 1, 31, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        sport_club = directories.add(
            title='Sport Clubs',
            structure="""
                Name *= ___
                Category *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Name]",
                order=['Name'],
                searchable=['title']
            ),
            enable_update_notifications=False,
        )
        entry = sport_club.add(values=dict(
            name='Wanderfreunde',
            category='Hiking',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 22, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = sport_club.add(values=dict(
            name='Pokerfreunde',
            category='Games',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 2, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        EntryRecipientCollection(client.app.session()).add(
            directory_id=planauflage.id,
            address='john@doe.ch',
            confirmed=True
        )
        EntryRecipientCollection(client.app.session()).add(
            directory_id=sport_club.id,
            address='john@doe.ch',
            confirmed=True
        )
    
        transaction.commit()
        close_all_sessions()
    
        assert count_recipients() == 1
        john = EntryRecipientCollection(client.app.session()).query().first()
        assert john is not None
    
        assert client.app.org.meta.get('hourly_maintenance_tasks_last_run') is None
    
        with freeze_time(datetime(2025, 1, 1, 4, 0, tzinfo=tz)):
            client.get(get_cronjob_url(job))
    
            assert len(os.listdir(client.app.maildir)) == 0
>           assert client.app.org.meta.get('hourly_maintenance_tasks_last_run')
                   ^^^^^^^^^^^^^^

.../onegov/org/test_cronjobs.py:2165: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7fc04cb2da90>
state = <sqlalchemy.orm.state.InstanceState object at 0x7fc04e769a70>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7fc04e769a70>, 'name': 'Govikon', 'theme_options': {'primary-color': '#006fba'}, 'created': datetime.datetime(2026, 8, 24, 9, 45, 37, 177730, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/org/test_cronjobs.py::test_send_email_notification_for_recent_directory_entry_publications[secret_mtan]
Stack Traces | 1.62s run time
client = <tests.onegov.org.conftest.Client object at 0x7f5736563e50>
access = 'secret_mtan'

    @pytest.mark.parametrize(
        'access',
        ('private', 'member', 'mtan', 'secret', 'secret_mtan', 'public')
    )
    def test_send_email_notification_for_recent_directory_entry_publications(
        client: Client[TestOrgApp],
        access: str
    ) -> None:
    
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
        tz = ensure_timezone('Europe/Zurich')
    
        def planauflagen() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('offentliche-planauflage'))
    
        def sport_clubs() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('sport-clubs'))
    
        def count_recipients() -> int:
            return (EntryRecipientCollection(client.app.session()).query()
                    .filter_by(directory_id=planauflagen().id)
                    .filter_by(confirmed=True).count())
    
        assert len(os.listdir(client.app.maildir)) == 0
    
        transaction.begin()
    
        directories: DirectoryCollection[ExtendedDirectory]
        directories = DirectoryCollection(client.app.session(), type='extended')
        planauflage = directories.add(
            title='Öffentliche Planauflage',
            structure="""
                Gesuchsteller/in *= ___
                Grundeigentümer/in *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Gesuchsteller/in]",
                order=['Gesuchsteller/in'],
                searchable=['title'],
            ),
            enable_update_notifications=True,
        )
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Carmine Carminio',
            grundeigentumer_in='Doris Dorinio',
            publication_start=datetime(2025, 1, 6, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 1, 30, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Emil Emilio',
            grundeigentumer_in='Franco Francinio',
            publication_start=datetime(2025, 1, 8, 6, 1, tzinfo=tz),
            publication_end=datetime(2025, 1, 31, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        sport_club = directories.add(
            title='Sport Clubs',
            structure="""
                Name *= ___
                Category *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Name]",
                order=['Name'],
                searchable=['title']
            ),
            enable_update_notifications=False,
        )
        entry = sport_club.add(values=dict(
            name='Wanderfreunde',
            category='Hiking',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 22, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = sport_club.add(values=dict(
            name='Pokerfreunde',
            category='Games',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 2, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        EntryRecipientCollection(client.app.session()).add(
            directory_id=planauflage.id,
            address='john@doe.ch',
            confirmed=True
        )
        EntryRecipientCollection(client.app.session()).add(
            directory_id=sport_club.id,
            address='john@doe.ch',
            confirmed=True
        )
    
        transaction.commit()
        close_all_sessions()
    
        assert count_recipients() == 1
        john = EntryRecipientCollection(client.app.session()).query().first()
        assert john is not None
    
        assert client.app.org.meta.get('hourly_maintenance_tasks_last_run') is None
    
        with freeze_time(datetime(2025, 1, 1, 4, 0, tzinfo=tz)):
            client.get(get_cronjob_url(job))
    
            assert len(os.listdir(client.app.maildir)) == 0
>           assert client.app.org.meta.get('hourly_maintenance_tasks_last_run')
                   ^^^^^^^^^^^^^^

.../onegov/org/test_cronjobs.py:2165: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7f5731b6a660>
state = <sqlalchemy.orm.state.InstanceState object at 0x7f5737cc7490>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7f5737cc7490>, 'theme_options': {'primary-color': '#006fba'}, 'name': 'Govikon', 'created': datetime.datetime(2026, 8, 24, 9, 45, 35, 727235, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/org/test_cronjobs.py::test_send_email_notification_for_recent_directory_entry_publications[secret]
Stack Traces | 1.64s run time
client = <tests.onegov.org.conftest.Client object at 0x7fd6756b4e50>
access = 'secret'

    @pytest.mark.parametrize(
        'access',
        ('private', 'member', 'mtan', 'secret', 'secret_mtan', 'public')
    )
    def test_send_email_notification_for_recent_directory_entry_publications(
        client: Client[TestOrgApp],
        access: str
    ) -> None:
    
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
        tz = ensure_timezone('Europe/Zurich')
    
        def planauflagen() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('offentliche-planauflage'))
    
        def sport_clubs() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('sport-clubs'))
    
        def count_recipients() -> int:
            return (EntryRecipientCollection(client.app.session()).query()
                    .filter_by(directory_id=planauflagen().id)
                    .filter_by(confirmed=True).count())
    
        assert len(os.listdir(client.app.maildir)) == 0
    
        transaction.begin()
    
        directories: DirectoryCollection[ExtendedDirectory]
        directories = DirectoryCollection(client.app.session(), type='extended')
        planauflage = directories.add(
            title='Öffentliche Planauflage',
            structure="""
                Gesuchsteller/in *= ___
                Grundeigentümer/in *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Gesuchsteller/in]",
                order=['Gesuchsteller/in'],
                searchable=['title'],
            ),
            enable_update_notifications=True,
        )
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Carmine Carminio',
            grundeigentumer_in='Doris Dorinio',
            publication_start=datetime(2025, 1, 6, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 1, 30, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Emil Emilio',
            grundeigentumer_in='Franco Francinio',
            publication_start=datetime(2025, 1, 8, 6, 1, tzinfo=tz),
            publication_end=datetime(2025, 1, 31, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        sport_club = directories.add(
            title='Sport Clubs',
            structure="""
                Name *= ___
                Category *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Name]",
                order=['Name'],
                searchable=['title']
            ),
            enable_update_notifications=False,
        )
        entry = sport_club.add(values=dict(
            name='Wanderfreunde',
            category='Hiking',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 22, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = sport_club.add(values=dict(
            name='Pokerfreunde',
            category='Games',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 2, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        EntryRecipientCollection(client.app.session()).add(
            directory_id=planauflage.id,
            address='john@doe.ch',
            confirmed=True
        )
        EntryRecipientCollection(client.app.session()).add(
            directory_id=sport_club.id,
            address='john@doe.ch',
            confirmed=True
        )
    
        transaction.commit()
        close_all_sessions()
    
        assert count_recipients() == 1
        john = EntryRecipientCollection(client.app.session()).query().first()
        assert john is not None
    
        assert client.app.org.meta.get('hourly_maintenance_tasks_last_run') is None
    
        with freeze_time(datetime(2025, 1, 1, 4, 0, tzinfo=tz)):
            client.get(get_cronjob_url(job))
    
            assert len(os.listdir(client.app.maildir)) == 0
>           assert client.app.org.meta.get('hourly_maintenance_tasks_last_run')
                   ^^^^^^^^^^^^^^

.../onegov/org/test_cronjobs.py:2165: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7fd67950f0e0>
state = <sqlalchemy.orm.state.InstanceState object at 0x7fd6743868b0>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7fd6743868b0>, 'name': 'Govikon', 'theme_options': {'primary-color': '#006fba'}, 'created': datetime.datetime(2026, 8, 24, 9, 45, 35, 897277, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/org/test_cronjobs.py::test_send_email_notification_for_recent_directory_entry_publications[private]
Stack Traces | 1.64s run time
client = <tests.onegov.org.conftest.Client object at 0x7fc04cd31fd0>
access = 'private'

    @pytest.mark.parametrize(
        'access',
        ('private', 'member', 'mtan', 'secret', 'secret_mtan', 'public')
    )
    def test_send_email_notification_for_recent_directory_entry_publications(
        client: Client[TestOrgApp],
        access: str
    ) -> None:
    
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
        tz = ensure_timezone('Europe/Zurich')
    
        def planauflagen() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('offentliche-planauflage'))
    
        def sport_clubs() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('sport-clubs'))
    
        def count_recipients() -> int:
            return (EntryRecipientCollection(client.app.session()).query()
                    .filter_by(directory_id=planauflagen().id)
                    .filter_by(confirmed=True).count())
    
        assert len(os.listdir(client.app.maildir)) == 0
    
        transaction.begin()
    
        directories: DirectoryCollection[ExtendedDirectory]
        directories = DirectoryCollection(client.app.session(), type='extended')
        planauflage = directories.add(
            title='Öffentliche Planauflage',
            structure="""
                Gesuchsteller/in *= ___
                Grundeigentümer/in *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Gesuchsteller/in]",
                order=['Gesuchsteller/in'],
                searchable=['title'],
            ),
            enable_update_notifications=True,
        )
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Carmine Carminio',
            grundeigentumer_in='Doris Dorinio',
            publication_start=datetime(2025, 1, 6, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 1, 30, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Emil Emilio',
            grundeigentumer_in='Franco Francinio',
            publication_start=datetime(2025, 1, 8, 6, 1, tzinfo=tz),
            publication_end=datetime(2025, 1, 31, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        sport_club = directories.add(
            title='Sport Clubs',
            structure="""
                Name *= ___
                Category *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Name]",
                order=['Name'],
                searchable=['title']
            ),
            enable_update_notifications=False,
        )
        entry = sport_club.add(values=dict(
            name='Wanderfreunde',
            category='Hiking',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 22, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = sport_club.add(values=dict(
            name='Pokerfreunde',
            category='Games',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 2, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        EntryRecipientCollection(client.app.session()).add(
            directory_id=planauflage.id,
            address='john@doe.ch',
            confirmed=True
        )
        EntryRecipientCollection(client.app.session()).add(
            directory_id=sport_club.id,
            address='john@doe.ch',
            confirmed=True
        )
    
        transaction.commit()
        close_all_sessions()
    
        assert count_recipients() == 1
        john = EntryRecipientCollection(client.app.session()).query().first()
        assert john is not None
    
        assert client.app.org.meta.get('hourly_maintenance_tasks_last_run') is None
    
        with freeze_time(datetime(2025, 1, 1, 4, 0, tzinfo=tz)):
            client.get(get_cronjob_url(job))
    
            assert len(os.listdir(client.app.maildir)) == 0
>           assert client.app.org.meta.get('hourly_maintenance_tasks_last_run')
                   ^^^^^^^^^^^^^^

.../onegov/org/test_cronjobs.py:2165: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7fc04b5df380>
state = <sqlalchemy.orm.state.InstanceState object at 0x7fc0513fec40>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7fc0513fec40>, 'name': 'Govikon', 'theme_options': {'primary-color': '#006fba'}, 'created': datetime.datetime(2026, 8, 24, 9, 45, 33, 543394, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/org/test_cronjobs.py::test_send_email_notification_for_recent_directory_entry_publications[mtan]
Stack Traces | 1.67s run time
client = <tests.onegov.org.conftest.Client object at 0x7fc04ef1e1a0>
access = 'mtan'

    @pytest.mark.parametrize(
        'access',
        ('private', 'member', 'mtan', 'secret', 'secret_mtan', 'public')
    )
    def test_send_email_notification_for_recent_directory_entry_publications(
        client: Client[TestOrgApp],
        access: str
    ) -> None:
    
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
        tz = ensure_timezone('Europe/Zurich')
    
        def planauflagen() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('offentliche-planauflage'))
    
        def sport_clubs() -> ExtendedDirectory:
            return (DirectoryCollection(client.app.session(), type='extended')  # type: ignore[return-value]
                    .by_name('sport-clubs'))
    
        def count_recipients() -> int:
            return (EntryRecipientCollection(client.app.session()).query()
                    .filter_by(directory_id=planauflagen().id)
                    .filter_by(confirmed=True).count())
    
        assert len(os.listdir(client.app.maildir)) == 0
    
        transaction.begin()
    
        directories: DirectoryCollection[ExtendedDirectory]
        directories = DirectoryCollection(client.app.session(), type='extended')
        planauflage = directories.add(
            title='Öffentliche Planauflage',
            structure="""
                Gesuchsteller/in *= ___
                Grundeigentümer/in *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Gesuchsteller/in]",
                order=['Gesuchsteller/in'],
                searchable=['title'],
            ),
            enable_update_notifications=True,
        )
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Carmine Carminio',
            grundeigentumer_in='Doris Dorinio',
            publication_start=datetime(2025, 1, 6, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 1, 30, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = planauflage.add(values=dict(
            gesuchsteller_in='Emil Emilio',
            grundeigentumer_in='Franco Francinio',
            publication_start=datetime(2025, 1, 8, 6, 1, tzinfo=tz),
            publication_end=datetime(2025, 1, 31, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        sport_club = directories.add(
            title='Sport Clubs',
            structure="""
                Name *= ___
                Category *= ___
            """,
            configuration=DirectoryConfiguration(
                title="[Name]",
                order=['Name'],
                searchable=['title']
            ),
            enable_update_notifications=False,
        )
        entry = sport_club.add(values=dict(
            name='Wanderfreunde',
            category='Hiking',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 22, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        entry = sport_club.add(values=dict(
            name='Pokerfreunde',
            category='Games',
            publication_start=datetime(2025, 2, 1, 2, 0, tzinfo=tz),
            publication_end=datetime(2025, 2, 2, 2, 0, tzinfo=tz),
        ))
        entry.access = access
        assert entry.access == access
    
        EntryRecipientCollection(client.app.session()).add(
            directory_id=planauflage.id,
            address='john@doe.ch',
            confirmed=True
        )
        EntryRecipientCollection(client.app.session()).add(
            directory_id=sport_club.id,
            address='john@doe.ch',
            confirmed=True
        )
    
        transaction.commit()
        close_all_sessions()
    
        assert count_recipients() == 1
        john = EntryRecipientCollection(client.app.session()).query().first()
        assert john is not None
    
        assert client.app.org.meta.get('hourly_maintenance_tasks_last_run') is None
    
        with freeze_time(datetime(2025, 1, 1, 4, 0, tzinfo=tz)):
            client.get(get_cronjob_url(job))
    
            assert len(os.listdir(client.app.maildir)) == 0
>           assert client.app.org.meta.get('hourly_maintenance_tasks_last_run')
                   ^^^^^^^^^^^^^^

.../onegov/org/test_cronjobs.py:2165: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7fc04fdd86e0>
state = <sqlalchemy.orm.state.InstanceState object at 0x7fc04dac5f30>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7fc04dac5f30>, 'name': 'Govikon', 'theme_options': {'primary-color': '#006fba'}, 'created': datetime.datetime(2026, 8, 24, 9, 45, 35, 337696, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/org/test_views_directory.py::test_delete_entry_published_before_last_maintenance_run
Stack Traces | 2.23s run time
client = <tests.onegov.org.conftest.Client object at 0x7f4f6033e840>

    def test_delete_entry_published_before_last_maintenance_run(
        client: Client,
    ) -> None:
        """An entry that just became published must not be deletable either,
        even if the hourly cronjob has not run since publication_start passed.
        It is publicly visible right now, so deleting it would remove a live
        record and bypass the proof-of-publication guarantee."""
        tz = 'Europe/Zurich'
        now_local = to_timezone(utcnow(), tz)
    
        client.login_admin()
    
        directory_page = create_notification_directory(client)
    
        # scheduled for the future (so the form's currently-published guard
        # doesn't reject it)
        page = directory_page.click('Eintrag', index=0)
        page.form['name'] = 'Permit One'
        page.form['publication_start'] = dt_for_form(now_local + timedelta(days=1))
        page.form['publication_end'] = dt_for_form(now_local + timedelta(days=30))
        entry = page.form.submit().follow()
    
        csrf_token = deletable_entry_csrf_token(directory_page)
        entry_url = entry.pyquery('a.edit-link').attr('href').replace('+edit', '')
    
        # simulate: publication has just started, but the last maintenance run
        # predates publication_start, so the cronjob has not yet observed the
        # entry as published (nor sent the published notification)
        session = client.app.session()
        permit_one = dir_query(client).filter_by(name='permit-one').one()
        permit_one.publication_start = utcnow() - timedelta(minutes=30)
        client.app.org.hourly_maintenance_tasks_last_run = (
            utcnow() - timedelta(minutes=90)
        )
        session.flush()
        transaction.commit()
    
        # the entry is published (visible) by the wall clock, but was not yet
        # published as of the last maintenance run
        permit_one = dir_query(client).filter_by(name='permit-one').one()
        assert permit_one.published is True
        assert permit_one.published_as_of(
>           client.app.org.hourly_maintenance_tasks_last_run
            ^^^^^^^^^^^^^^
        ) is False

.../onegov/org/test_views_directory.py:1610: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7f4f60d72cf0>
state = <sqlalchemy.orm.state.InstanceState object at 0x7f4f60a6eea0>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7f4f60a6eea0>, 'name': 'Govikon', 'theme_options': {'primary-color': '#006fba'}, 'created': datetime.datetime(2026, 8, 24, 9, 45, 23, 13849, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/org/test_cronjobs.py::test_admin_notification_signing_failure
Stack Traces | 2.67s run time
client = <tests.onegov.org.conftest.Client object at 0x7fd6740833e0>
caplog = <_pytest.logging.LogCaptureFixture object at 0x7fd674e374d0>

    def test_admin_notification_signing_failure(
        client: Client['TestOrgApp'],
        caplog: pytest.LogCaptureFixture,
    ) -> None:
        """
        A failing signing service does not prevent the notifications — both
        the publication and the expiry pdf are attached unsigned, and the same
        entry still gets its expiry mail after the first signing already failed.
        """
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
    
        real_now = utcnow()
    
        transaction.begin()
        directory = _make_permit_directory(client.app.session())
        directory.add(
            values=dict(
                gesuchsteller_in='Clara Meier',
                adresse='Ringstrasse 9',
                publication_start=real_now - timedelta(minutes=30),
                publication_end=real_now + timedelta(hours=2),
            )
        )
        transaction.commit()
        close_all_sessions()
    
        with (
            patch.object(SwisscomAIS, 'sign') as sign,
            caplog.at_level(logging.ERROR, logger='onegov.org'),
        ):
            sign.side_effect = RuntimeError('signing service unavailable')
    
            # the publication start has been crossed ...
            with freeze_time(real_now, tick=True):
                client.get(get_cronjob_url(job))
            assert len(os.listdir(client.app.maildir)) == 1
    
            # ... and later on the publication end
            with freeze_time(real_now + timedelta(hours=3), tick=True):
                client.get(get_cronjob_url(job))
    
>           assert sign.call_count == 2
E           AssertionError: assert 1 == 2
E            +  where 1 = <MagicMock name='sign' id='140559060790848'>.call_count

.../onegov/org/test_cronjobs.py:3562: AssertionError
tests/onegov/org/test_cronjobs.py::test_admin_notification_full_workflow
Stack Traces | 2.67s run time
client = <tests.onegov.org.conftest.Client object at 0x7fc050355bd0>

    def test_admin_notification_full_workflow(
        client: Client[TestOrgApp]
    ) -> None:
        """End-to-end, form-driven publication lifecycle for an entry in a
        directory with a notification_address, proving:
    
        """
        tz = 'Europe/Zurich'
    
        def dt(local: datetime) -> str:
            return local.strftime('%Y-%m-%dT%H:%M')
    
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
    
        base = datetime(2026, 7, 1, 10, 0, tzinfo=timezone.utc)
    
        with freeze_time(base, tick=True):
            client.login_admin()
    
            # a permit directory with an editable content field
            page = client.get('/directories').click('^Verzeichnis$')
            page.form['title'] = 'Baugesuche'
            page.form['structure'] = (
                'Name *= ___\nBeschreibung *= ___'
                '\nTermin *= YYYY.MM.DD HH:MM\nFrist *= YYYY.MM.DD'
                '\nDokument = *.pdf'
            )
            page.form['title_format'] = '[Name]'
            page.form['enable_publication'] = True
            page.form['required_publication'] = True
            page.form['notification_address'] = 'admin@example.org'
            page.form['enable_change_requests'] = False
            page = page.form.submit().follow()
    
            # a scheduled entry (start in 2h)
            now_local = to_timezone(utcnow(), tz)
            page = page.click('Eintrag', index=0)
            page.form['name'] = 'Permit One'
            page.form['beschreibung'] = 'Version A'
            page.form['termin'] = '2026-03-15T09:30'
            page.form['frist'] = '2026-08-20'
            sample_pdf = module_path('tests.onegov.org', 'fixtures/sample.pdf')
            with open(sample_pdf, 'rb') as pdf_file:
                page.form['dokument'] = Upload(
                    'Baugesuch.pdf', pdf_file.read(), 'application/pdf'
                )
            page.form['publication_start'] = dt(now_local + timedelta(hours=2))
            page.form['publication_end'] = dt(now_local + timedelta(hours=9))
            page = page.form.submit().follow()
            assert len(os.listdir(client.app.maildir)) == 0
    
            # editing while scheduled must not notify
            page = page.click('Bearbeiten')
            page.form['beschreibung'] = 'Version B'
            page = page.form.submit().follow()
            entry_url = page.request.url
            assert len(os.listdir(client.app.maildir)) == 0
    
        # cronjob before the start: still nothing (establishes last_run)
        with freeze_time(base + timedelta(hours=1), tick=True):
            client.get(get_cronjob_url(job))
            assert len(os.listdir(client.app.maildir)) == 0
    
        # cronjob after the start: one publication email carrying 'Version B'
        with freeze_time(base + timedelta(hours=3), tick=True), _ais_cassette():
            client.get(get_cronjob_url(job))
>           assert len(os.listdir(client.app.maildir)) == 1
E           AssertionError: assert 0 == 1
E            +  where 0 = len([])
E            +    where [] = <built-in function listdir>('....../tmp/tmp51n7nc22/mails')
E            +      where <built-in function listdir> = os.listdir
E            +      and   '....../tmp/tmp51n7nc22/mails' = <tests.onegov.org.conftest.TestOrgApp object at 0x7fc04de49450>.maildir
E            +        where <tests.onegov.org.conftest.TestOrgApp object at 0x7fc04de49450> = <tests.onegov.org.conftest.Client object at 0x7fc050355bd0>.app

.../onegov/org/test_cronjobs.py:3230: AssertionError
tests/onegov/org/test_views_event.py::test_view_occurrences_event_filter
Stack Traces | 3.11s run time
client = <tests.onegov.org.conftest.Client object at 0x7fc132ef6950>

    def test_view_occurrences_event_filter(client: Client) -> None:
        """
        This test switches the application settings event filter type between
        'tags', 'filters' and 'tags_and_filters'
        """
    
        def events(query: str = '') -> list[str]:
            page = client.get(f'/events/?{query}')
            return [event.text for event in page.pyquery('h3 a')]
    
        def dates(query: str = '') -> list[date]:
            page = client.get(f'/events/?{query}')
            return [
                datetime.strptime(div.text, '%d.%m.%Y').date()
                for div in page.pyquery('.occurrence-date')
            ]
    
        def set_setting_event_filter_type(
            client: Client,
            event_filter_type: str
        ) -> None:
            client.login_admin()
            settings = client.get('/event-settings')
            settings.form['event_filter_type'] = event_filter_type
            settings.form.submit()
            assert client.app.org.event_filter_type == event_filter_type
            client.logout()
    
        def setup_event_filter(client: Client) -> None:
            assert client.login_admin()
            assert client.app.org.event_filter_type in ['filters',
                                                        'tags_and_filters']
            page = client.get('/event-settings')
            page.form['event_filter_definition'] = """
                My Special Filter *=
                    [ ] A Filter
                    [ ] B Filter
            """
            page.form['keyword_fields'].value = 'My Special Filter'
            assert page.form.submit()
            client.logout()
    
        def set_filter_on_event(client: Client) -> None:
            # set single filter on one event
            client.login_admin()
            page = (client.get('/events').click('Fussballturnier').
                    click('Bearbeiten'))
            page.form['my_special_filter'] = ['B Filter']
            page.form.submit()
            client.logout()
    
        assert len(events()) == 10
        assert len(events('page=1')) == 2
        assert dates() == sorted(dates())
    
        # default: event filter type = 'tags'
        client.app.org.event_filter_type = 'tags'
        page = client.get('/events')
        assert '<h2>Schlagwort</h2>' in page
        assert 'My Special Filter' not in page
        assert 'A Filter' not in page
        assert 'B Filter' not in page
    
        # default: event filter type = 'filters'
>       set_setting_event_filter_type(client, 'filters')

.../onegov/org/test_views_event.py:238: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../onegov/org/test_views_event.py:199: in set_setting_event_filter_type
    assert client.app.org.event_filter_type == event_filter_type
           ^^^^^^^^^^^^^^
.../core/orm/cache.py:438: in wrapper
    return maybe_merge(self.session(), self.request_cache[cache_key])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.../core/orm/utils.py:39: in maybe_merge
    obj = session.merge(obj, load=False)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....../app/lib/python3.14.../sqlalchemy/orm/session.py:3967: in merge
    return self._merge(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.orm.session.Session object at 0x7fc13ba40980>
state = <sqlalchemy.orm.state.InstanceState object at 0x7fc13b5696e0>
state_dict = {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7fc13b5696e0>, 'theme_options': {'primary-color': '#006fba'}, 'name': 'Govikon', 'created': datetime.datetime(2026, 8, 24, 9, 45, 25, 823738, tzinfo=<UTC>), ...}
options = None, load = False, _recursive = {}, _resolve_conflict_map = {}

    def _merge(
        self,
        state: InstanceState[_O],
        state_dict: _InstanceDict,
        *,
        options: Optional[Sequence[ORMOption]] = None,
        load: bool,
        _recursive: Dict[Any, object],
        _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
    ) -> _O:
        mapper: Mapper[_O] = _state_mapper(state)
        if state in _recursive:
            return cast(_O, _recursive[state])
    
        new_instance = False
        key = state.key
    
        merged: Optional[_O]
    
        if key is None:
            if state in self._new:
                util.warn(
                    "Instance %s is already pending in this Session yet is "
                    "being merged again; this is probably not what you want "
                    "to do" % state_str(state)
                )
    
            if not load:
                raise sa_exc.InvalidRequestError(
                    "merge() with load=False option does not support "
                    "objects transient (i.e. unpersisted) objects.  flush() "
                    "all changes on mapped instances before merging with "
                    "load=False."
                )
            key = mapper._identity_key_from_state(state)
            key_is_persistent = LoaderCallableStatus.NEVER_SET not in key[
                1
            ] and (
                not _none_set.intersection(key[1])
                or (
                    mapper.allow_partial_pks
                    and not _none_set.issuperset(key[1])
                )
            )
        else:
            key_is_persistent = True
    
        merged = self.identity_map.get(key)
    
        if merged is None:
            if key_is_persistent and key in _resolve_conflict_map:
                merged = cast(_O, _resolve_conflict_map[key])
    
            elif not load:
                if state.modified:
>                   raise sa_exc.InvalidRequestError(
                        "merge() with load=False option does not support "
                        "objects marked as 'dirty'.  flush() all changes on "
                        "mapped instances before merging with load=False."
                    )
E                   sqlalchemy.exc.InvalidRequestError: merge() with load=False option does not support objects marked as 'dirty'.  flush() all changes on mapped instances before merging with load=False.

....../app/lib/python3.14.../sqlalchemy/orm/session.py:4033: InvalidRequestError
tests/onegov/agency/test_cli.py::test_enable_yubikey
Stack Traces | 4.2s run time
temporary_directory = '/tmp/tmpzng9mgv4'
cfg_path = '.../tmp/tmpzng9mgv4/onegov.yml'
session_manager = <onegov.core.orm.session_manager.SessionManager object at 0x7f6c7d5d5350>

    def test_enable_yubikey(
        temporary_directory: str,
        cfg_path: str,
        session_manager: SessionManager
    ) -> None:
    
        runner = CliRunner()
    
        result = runner.invoke(org_cli, [
            '--config', cfg_path,
            '--select', '/agency/govikon',
            'add', 'Govikon'
        ])
        assert result.exit_code == 0
    
        session_manager.set_current_schema('agency-govikon')
        session = session_manager.session()
        assert 'enable_yubikey' not in session.query(Organisation).one().meta
    
        result = runner.invoke(cli, [
            '--config', cfg_path,
            '--select', '/agency/govikon',
            'enable-yubikey'
        ])
        assert result.exit_code == 0
        assert "YubiKey enabled" in result.output
>       assert session.query(Organisation).one().meta['enable_yubikey'] is True
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E       KeyError: 'enable_yubikey'

.../onegov/agency/test_cli.py:70: KeyError
tests/onegov/org/test_cronjobs.py::test_admin_notification_multiple_entries
Stack Traces | 13.2s run time
client = <tests.onegov.org.conftest.Client object at 0x7fd675ef3950>

    def test_admin_notification_multiple_entries(
        client: Client[TestOrgApp]
    ) -> None:
        """
        Two entries crossing publication_start in the same window each get
        their own notification email.
        """
        job = get_cronjob_by_name(client.app, 'hourly_maintenance_tasks')
        assert job is not None
        job.app = client.app
    
        real_now = utcnow()
        pub_start = real_now - timedelta(minutes=30)
        pub_end = real_now + timedelta(days=30)
    
        transaction.begin()
        directory = _make_permit_directory(client.app.session())
        directory.add(
            values=dict(
                gesuchsteller_in='Anton Müller',
                adresse='Hauptstrasse 1',
                publication_start=pub_start,
                publication_end=pub_end,
            )
        )
        directory.add(
            values=dict(
                gesuchsteller_in='Berta Schmid',
                adresse='Seeweg 5',
                publication_start=pub_start,
                publication_end=pub_end,
            )
        )
        transaction.commit()
        close_all_sessions()
    
        # both publication_starts crossed in the same window — one email each
        with freeze_time(real_now, tick=True), _ais_cassette():
            client.get(get_cronjob_url(job))
    
        assert len(os.listdir(client.app.maildir)) == 2
        subjects = {client.get_email(i)['Subject'] for i in (0, 1)}
        assert all('Veröffentlichter Eintrag' in s for s in subjects)
        assert any('Anton Müller' in s for s in subjects)
        assert any('Berta Schmid' in s for s in subjects)
        for i in (0, 1):
            assert client.get_email(i)['To'] == 'admin@example.org'
            assert _fmt_date(pub_start) in client.get_email(i)['TextBody']
            assert _fmt_date(pub_end) in client.get_email(i)['TextBody']
            _assert_signed_pdf(client.get_email(i)['Attachments'][0])
    
        # one signing request per notification
        assert client.app.session().query(SigningRequest).count() == 2
    
        # second run — no duplicates
        with freeze_time(real_now, tick=True):
            client.get(get_cronjob_url(job))
>       assert len(os.listdir(client.app.maildir)) == 2
E       AssertionError: assert 4 == 2
E        +  where 4 = len(['0.1.1787564742.478617', '0.1.1787564742.560597', '0.1.1787564750.889623', '0.1.1787564753.574528'])
E        +    where ['0.1.1787564742.478617', '0.1.1787564742.560597', '0.1.1787564750.889623', '0.1.1787564753.574528'] = <built-in function listdir>('....../tmp/tmpoqwwiwsh/mails')
E        +      where <built-in function listdir> = os.listdir
E        +      and   '....../tmp/tmpoqwwiwsh/mails' = <tests.onegov.org.conftest.TestOrgApp object at 0x7fd6738b1d10>.maildir
E        +        where <tests.onegov.org.conftest.TestOrgApp object at 0x7fd6738b1d10> = <tests.onegov.org.conftest.Client object at 0x7fd675ef3950>.app

.../onegov/org/test_cronjobs.py:3421: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
📋 Got 3 mins? Take this short survey to help us improve Test Analytics.

@Daverball Daverball left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The overall direction looks decent to me, but we should probably rely more on the existing paradigms for attaching additional state to the current session and responding to changes, rather than rebuilding those tools in a slightly different way with slightly different semantics.

Comment thread src/onegov/core/orm/audit.py Outdated
Comment thread src/onegov/core/orm/audit.py Outdated
Comment thread src/onegov/core/request.py Outdated
Comment thread src/onegov/core/upgrades.py Outdated
Comment thread src/onegov/core/orm/session_manager.py Outdated
Comment thread src/onegov/core/framework.py Outdated
Comment thread src/onegov/core/orm/session_manager.py Outdated
…-add-versioning-insert-update-or-delete-to-ogc

# Conflicts:
#	src/onegov/org/locale/de_CH/LC_MESSAGES/onegov.org.po
#	src/onegov/org/locale/fr_CH/LC_MESSAGES/onegov.org.po
#	src/onegov/org/locale/it_CH/LC_MESSAGES/onegov.org.po
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants