diff --git a/pycroft/lib/user_deletion.py b/pycroft/lib/user_deletion.py index 32627c877..c77e82292 100644 --- a/pycroft/lib/user_deletion.py +++ b/pycroft/lib/user_deletion.py @@ -5,14 +5,14 @@ This module contains methods concerning user archival and deletion. """ from __future__ import annotations -from datetime import timedelta, datetime +from datetime import datetime from typing import Protocol, Sequence, cast +from sqlalchemy import func from sqlalchemy.future import select from sqlalchemy.orm import joinedload, Session from sqlalchemy.sql import Select from sqlalchemy.sql.elements import and_, not_ -from sqlalchemy.sql.functions import current_timestamp from pycroft.model.property import CurrentProperty from pycroft.model.user import User @@ -25,7 +25,9 @@ class ArchivableMemberInfo(Protocol): mem_end: datetime -def select_archivable_members(delta: timedelta) -> Select: # Select[Tuple[User, int, datetime]] +def select_archivable_members( + current_year: int, +) -> Select: # Select[Tuple[User, int, datetime]] # last_mem: (user_id, mem_id, mem_end) last_mem = select_user_and_last_mem().cte("last_mem") return ( @@ -40,7 +42,7 @@ def select_archivable_members(delta: timedelta) -> Select: # Select[Tuple[User, # …and use that to filter out the `do-not-archive` occurrences. .filter(CurrentProperty.property_name.is_(None)) .join(User, User.id == last_mem.c.user_id) - .filter(last_mem.c.mem_end < current_timestamp() - delta) + .filter(func.extract("year", last_mem.c.mem_end) + 2 <= current_year) .order_by(last_mem.c.mem_end) .add_columns( User, @@ -51,7 +53,8 @@ def select_archivable_members(delta: timedelta) -> Select: # Select[Tuple[User, def get_archivable_members( - session: Session, delta: timedelta = timedelta(days=14) + session: Session, + current_year: int | None = None, ) -> Sequence[ArchivableMemberInfo]: """Return all the users that qualify for being archived right now. @@ -66,13 +69,18 @@ def get_archivable_members( - current_properties_maybe_denied :param session: - :param delta: how far back the end of membership has to lie (positive timedelta). + :param current_year: dependency injection of the current year. + defaults to the current year. """ return cast( list[ArchivableMemberInfo], session.execute( - select_archivable_members(delta) - .options( + select_archivable_members( + # I know we're sloppy with time zones, + # but ±2h around new year's eve don't matter. + current_year=current_year + or datetime.now().year + ).options( joinedload(User.hosts), # joinedload(User.current_memberships), joinedload(User.account, innerjoin=True), diff --git a/tests/lib/user/test_deletion.py b/tests/lib/user/test_deletion.py index 30ae7e647..dae88b8d1 100644 --- a/tests/lib/user/test_deletion.py +++ b/tests/lib/user/test_deletion.py @@ -2,11 +2,18 @@ # This file is part of the Pycroft project and licensed under the terms of # the Apache License, Version 2.0. See the LICENSE file for details from datetime import datetime, date +from typing import Sequence import pytest from pycroft.helpers.interval import closed, closedopen -from pycroft.lib.user_deletion import get_archivable_members, archive_users +from pycroft.helpers.utc import with_min_time +from pycroft.lib.user_deletion import ( + get_archivable_members, + archive_users, + ArchivableMemberInfo, +) +from pycroft.model.user import User from tests.factories import UserFactory, ConfigFactory, MembershipFactory, \ PropertyGroupFactory, \ HostFactory @@ -26,27 +33,49 @@ def test_users_without_membership_not_in_list(session): assert get_archivable_members(session) == [] -def assert_archivable_members(members, expected_user, expected_end_date): - match members: - case [(user, mem_id, mem_end)]: - assert user == expected_user - assert mem_id is not None - assert mem_end.date() == expected_end_date - case _: - pytest.fail() +def filter_members(members, user): + return [(u, id, end) for u, id, end in members if u == user] -class TestUserDeletion: +def assert_member_present( + members: Sequence[ArchivableMemberInfo], + expected_user: User, + expected_end_date: date, +): + relevant_members = filter_members(members, expected_user) + assert len(relevant_members) == 1 + [(_, mem_id, mem_end)] = relevant_members + assert mem_id is not None + assert mem_end.date() == expected_end_date + + +def assert_member_absent( + members: Sequence[ArchivableMemberInfo], + expected_absent_user: User, +): + assert not filter_members(members, expected_absent_user) + + +class TestArchivableUserSelection: + @pytest.fixture( + scope="class", + params=[date(2020, 3, 1), date(2020, 1, 1), date(2020, 12, 15)], + ) + def end_date(self, request): + return request.param + @pytest.fixture(scope='class') def do_not_archive_group(self, class_session): return PropertyGroupFactory(granted={'do-not-archive'}) - @pytest.fixture(scope='class') - def old_user(self, class_session, config, do_not_archive_group): + @pytest.fixture(scope="class") + def old_user(self, class_session, config, do_not_archive_group, end_date) -> User: user = UserFactory.create( registered_at=datetime(2000, 1, 1), with_membership=True, - membership__active_during=closed(datetime(2020, 1, 1), datetime(2020, 3, 1)), + membership__active_during=closed( + with_min_time(date(2020, 1, 1)), with_min_time(end_date) + ), membership__group=config.member_group, ) MembershipFactory.create( @@ -66,31 +95,38 @@ def do_not_archive_membership(self, session, old_user, do_not_archive_group): active_during=closedopen(datetime(2020, 1, 1), None), ) - def test_old_users_in_deletion_list(self, session, old_user): - members = get_archivable_members(session) - assert_archivable_members(members, old_user, date(2020, 3, 1)) + @pytest.mark.parametrize("year", [2022, 2023, 2024]) + def test_old_users_in_deletion_list_after(self, session, old_user, year, end_date): + members = get_archivable_members(session, current_year=year) + assert_member_present(members, old_user, end_date) - def test_old_user_not_in_list_with_long_delta(self, session, old_user): - delta = date.today() - date(2020, 1, 1) # before 2020-03-01 - assert get_archivable_members(session, delta) == [] + @pytest.mark.parametrize("year", [2019, 2020, 2021]) + def test_old_user_not_in_list_before(self, session, old_user, year): + assert_member_absent( + get_archivable_members(session, current_year=year), old_user + ) - def test_user_with_do_not_archive_not_in_list(self, session, old_user, - do_not_archive_membership): - assert get_archivable_members(session) == [] + @pytest.mark.parametrize("year", list(range(2018, 2023))) + def test_user_with_do_not_archive_not_in_list( + self, session, old_user, do_not_archive_membership, year + ): + assert_member_absent( + get_archivable_members(session, current_year=year), old_user + ) @pytest.mark.parametrize('num_hosts', [0, 2]) - def test_user_with_host_in_list(self, session, old_user, num_hosts): + def test_user_with_host_in_list(self, session, old_user, num_hosts, end_date): if num_hosts: HostFactory.create_batch(num_hosts, owner=old_user) members = get_archivable_members(session) - assert_archivable_members(members, old_user, date(2020, 3, 1)) + assert_member_present(members, old_user, end_date) - def test_user_with_room_in_list(self, session, old_user): + def test_user_with_room_in_list(self, session, old_user, end_date): with session.begin_nested(): old_user.room = None session.add(old_user) members = get_archivable_members(session) - assert_archivable_members(members, old_user, date(2020, 3, 1)) + assert_member_present(members, old_user, end_date) class TestUserArchival: