Enable export of non-attributable transfers

pycroft/helpers/printing/__init__.py

@@ -31,6 +31,7 @@ class BankAccount(t.Protocol):
     bank: t.Any
     iban: t.Any
     bic: t.Any
+    owner: t.Any
 
 
 class Building(t.Protocol):
@@ -289,7 +290,7 @@ def generate_user_sheet(
             six monthly contributions at once.'''.format(
             contribution / 100), style['JustifyText']))
 
-        recipient = 'Studierendenrat TUD - AG DSN'
+        recipient = bank_account.owner
 
         if user.room:
             purpose = '{id}, {name}, {dorm} {level} {room}'.format(

pycroft/lib/finance/__init__.py

@@ -41,6 +41,10 @@
     take_actions_for_payment_in_default_users,
     get_pid_csv,
 )
+from .retransfer import (
+    get_activities_to_return,
+    generate_activities_return_sepaxml,
+)
 from .transaction_crud import (
     simple_transaction,
     complex_transaction,

pycroft/lib/finance/retransfer.py

@@ -0,0 +1,46 @@
+from collections.abc import Sequence
+from datetime import datetime, timedelta
+
+from sepaxml import SepaTransfer
+from sqlalchemy import select
+from sqlalchemy.orm import joinedload, Session
+
+from pycroft import config
+from pycroft.helpers.utc import ensure_tz
+from pycroft.model.finance import BankAccountActivity
+
+
+def get_activities_to_return(session: Session) -> Sequence[BankAccountActivity]:
+    statement = (
+        select(BankAccountActivity)
+        .options(joinedload(BankAccountActivity.bank_account))
+        .filter(BankAccountActivity.transaction_id.is_(None))
+        .filter(BankAccountActivity.amount > 0)
+        .filter(BankAccountActivity.imported_at < ensure_tz(datetime.utcnow() - timedelta(days=14)))
+    )
+
+    return session.scalars(statement).all()
+
+
+def generate_activities_return_sepaxml(activities: list[BankAccountActivity]) -> bytes:
+    transfer_config: dict = {
+        "name": config.membership_fee_bank_account.owner,
+        "IBAN": config.membership_fee_bank_account.iban,
+        "BIC": config.membership_fee_bank_account.bic,
+        "batch": False,
+        "currency": "EUR",
+    }
+    sepa = SepaTransfer(transfer_config, clean=False)
+
+    for activity in activities:
+        payment = {
+            "name": activity.other_name,
+            "IBAN": activity.other_account_number,
+            "BIC": activity.other_routing_number,
+            "amount": int(activity.amount * 100),
+            "execution_date": datetime.now().date(),
+            "description": f"Rücküberweisung nicht zuordenbarer Überweisung vom {activity.posted_on} mit Referenz {activity.reference}",
+        }
+        sepa.add_payment(payment)
+
+    return sepa.export()

pycroft/model/alembic/versions/bc0e0dd480d4_add_bankaccount_owner.py

@@ -0,0 +1,26 @@
+"""add bankaccount owner
+
+Revision ID: bc0e0dd480d4
+Revises: 55e9f0d9b5f4
+Create Date: 2024-03-16 08:42:48.684471
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "bc0e0dd480d4"
+down_revision = "55e9f0d9b5f4"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "bank_account", sa.Column("owner", sa.String(length=255), nullable=False, server_default="")
+    )
+
+
+def downgrade():
+    op.drop_column("bank_account", "owner")

pycroft/model/finance.py

@@ -339,6 +339,7 @@ def check_split_on_update(mapper, connection, target):
 class BankAccount(IntegerIdModel):
     name: Mapped[str255]
     bank: Mapped[str255]
+    owner: Mapped[str255]
     account_number: Mapped[str] = mapped_column(String(10))
     routing_number: Mapped[str] = mapped_column(String(8))
     iban: Mapped[str] = mapped_column(String(34))

stubs/sepaxml/__init__.pyi

@@ -0,0 +1,4 @@
+from .debit import SepaDD as SepaDD
+from .transfer import SepaTransfer as SepaTransfer
+
+version: str

stubs/sepaxml/debit.pyi

@@ -0,0 +1,9 @@
+from .shared import SepaPaymentInitn as SepaPaymentInitn
+from .utils import ADDRESS_MAPPING as ADDRESS_MAPPING, int_to_decimal_str as int_to_decimal_str, make_id as make_id
+
+class SepaDD(SepaPaymentInitn):
+    root_el: str
+    def __init__(self, config, schema: str = 'pain.008.003.02', clean: bool = True) -> None: ...
+    def check_config(self, config): ...
+    def check_payment(self, payment): ...
+    def add_payment(self, payment) -> None: ...

stubs/sepaxml/shared.pyi

@@ -0,0 +1,10 @@
+from .utils import decimal_str_to_int as decimal_str_to_int, int_to_decimal_str as int_to_decimal_str, make_msg_id as make_msg_id
+from .validation import try_valid_xml as try_valid_xml
+from _typeshed import Incomplete
+
+class SepaPaymentInitn:
+    schema: Incomplete
+    msg_id: Incomplete
+    clean: Incomplete
+    def __init__(self, config, schema, clean: bool = True) -> None: ...
+    def export(self, validate: bool = True, pretty_print: bool = False) -> bytes: ...

stubs/sepaxml/transfer.pyi

@@ -0,0 +1,9 @@
+from .shared import SepaPaymentInitn as SepaPaymentInitn
+from .utils import ADDRESS_MAPPING as ADDRESS_MAPPING, int_to_decimal_str as int_to_decimal_str, make_id as make_id
+
+class SepaTransfer(SepaPaymentInitn):
+    root_el: str
+    def __init__(self, config, schema: str = 'pain.001.001.03', clean: bool = True) -> None: ...
+    def check_config(self, config): ...
+    def check_payment(self, payment): ...
+    def add_payment(self, payment) -> None: ...

stubs/sepaxml/utils.pyi

@@ -0,0 +1,11 @@
+from _typeshed import Incomplete
+
+using_sysrandom: bool
+
+def get_rand_string(length: int = 12, allowed_chars: str = '0123456789abcdef'): ...
+def make_msg_id(): ...
+def make_id(name): ...
+def int_to_decimal_str(integer): ...
+def decimal_str_to_int(decimal_string): ...
+
+ADDRESS_MAPPING: Incomplete

stubs/sepaxml/validation.pyi

@@ -0,0 +1,3 @@
+class ValidationError(Exception): ...
+
+def try_valid_xml(xmlout, schema) -> None: ...

tests/factories/finance.py

@@ -34,11 +34,12 @@ class BankAccountFactory(BaseFactory):
     class Meta:
         model = BankAccount
 
-    name = Faker('word')
-    bank = Faker('word')
-    account_number = Faker('random_number', digits=10)
-    routing_number = Faker('random_number', digits=8)
-    iban = Faker('iban')
+    name = Faker("word")
+    bank = Faker("word")
+    owner = Faker("word")
+    account_number = Faker("random_number", digits=10)
+    routing_number = Faker("random_number", digits=8)
+    iban = Faker("iban")
     bic = Faker("swift", length=11)
     fints_endpoint = Faker('url')
     account = SubFactory(AccountFactory, type='BANK_ASSET')
@@ -51,8 +52,8 @@ class Meta:
     bank_account = SubFactory(BankAccountFactory)
     amount = Faker('random_number', digits=5)
     reference = Sequence(lambda n: f"Reference {n}")
-    other_account_number = Faker('random_number', digits=10)
-    other_routing_number = Faker('random_number', digits=8)
+    other_account_number = Faker("iban")
+    other_routing_number = Faker("swift")
     other_name = Faker('word')
     imported_at = LazyAttribute(lambda o: session.utcnow().date() - timedelta(days=4))
     posted_on = LazyAttribute(lambda o: o.imported_at + timedelta(days=1))

tests/frontend/assertions.py

@@ -89,9 +89,7 @@ def assert_url_forbidden(self, url: str, method: str = "HEAD", **kw) -> Response
         __tracebackhide__ = True
         resp = self.open(url, method=method, **kw)
         status = resp.status_code
-        assert (
-            status == 403
-        ), f"Access to {url} expected to be forbidden, got status {status}"
+        assert status == 403, f"Access to {url} expected to be forbidden, got status {status}"
         return resp
 
     def assert_forbidden(self, endpoint: str, method: str = "HEAD", **kw) -> Response:

tests/frontend/conftest.py

@@ -72,10 +72,13 @@ def config(module_session: Session) -> Config:
 
 @pytest.fixture(scope="session")
 def blueprint_urls(app: PycroftFlask) -> BlueprintUrls:
-    def _blueprint_urls(blueprint_name: str) -> list[str]:
+    def _blueprint_urls(
+        blueprint_name: str, methods: set[str] = {"GET", "POST"}  # noqa: B006
+    ) -> list[str]:
         return [
-            _build_rule(request_ctx.url_adapter, rule)
+            (_build_rule(request_ctx.url_adapter, rule), method)
             for rule in app.url_map.iter_rules()
+            for method in rule.methods & methods
             if rule.endpoint.startswith(f"{blueprint_name}.")
         ]
     return _blueprint_urls

tests/frontend/fixture_helpers.py

@@ -26,7 +26,7 @@ def login_context(test_client: TestClient, login: str, password: str):
     test_client.get("/logout")
 
 
-BlueprintUrls: t.TypeAlias = t.Callable[[str], list[str]]
+BlueprintUrls: t.TypeAlias = t.Callable[[str], list[str, str]]
 _argument_creator_map = {
     IntegerConverter: lambda c: 1,
     UnicodeConverter: lambda c: "test",

tests/frontend/test_permissions.py

@@ -104,26 +104,26 @@ def member_logged_in(
             yield
 
     def test_access_user(self, client: TestClient, blueprint_urls: BlueprintUrls):
-        for url in blueprint_urls("user"):
-            client.assert_url_forbidden(url)
+        for url, method in blueprint_urls("user"):
+            client.assert_url_forbidden(url, method=method)
 
     def test_access_finance(self, client: TestClient, blueprint_urls: BlueprintUrls):
-        for url in blueprint_urls("finance"):
-            client.assert_url_forbidden(url)
+        for url, method in blueprint_urls("finance"):
+            client.assert_url_forbidden(url, method=method)
 
     def test_access_buildings(self, client: TestClient, blueprint_urls: BlueprintUrls):
-        for url in blueprint_urls("facilities"):
-            client.assert_url_forbidden(url)
+        for url, method in blueprint_urls("facilities"):
+            client.assert_url_forbidden(url, method=method)
 
     def test_access_infrastructure(
         self, client: TestClient, blueprint_urls: BlueprintUrls
     ):
-        for url in blueprint_urls("infrastructure"):
-            client.assert_url_forbidden(url)
+        for url, method in blueprint_urls("infrastructure"):
+            client.assert_url_forbidden(url, method=method)
 
     def test_access_properties(self, client: TestClient, blueprint_urls: BlueprintUrls):
-        for url in blueprint_urls("properties"):
-            client.assert_url_forbidden(url)
+        for url, method in blueprint_urls("properties"):
+            client.assert_url_forbidden(url, method=method)
 
     def test_access_login(self, client: TestClient, blueprint_urls: BlueprintUrls):
         # Login see Test_010_Anonymous

tests/lib/test_finance.py

@@ -18,7 +18,10 @@
     estimate_balance,
     post_transactions_for_membership_fee, get_users_with_payment_in_default,
     end_payment_in_default_memberships,
-    take_actions_for_payment_in_default_users)
+    take_actions_for_payment_in_default_users,
+    get_activities_to_return,
+    generate_activities_return_sepaxml,
+)
 from pycroft.model.finance import (
     Transaction,
     Split,
@@ -654,3 +657,65 @@ def test_last_imported_at(self, session: Session):
         assert finance.get_last_import_date(session) == datetime(
             2020, 1, 1, tzinfo=timezone.utc
         )
+
+
+class TestReturnNonAttributable:
+    @pytest.mark.parametrize(
+        "expected, set_transaction_id, amount_negative, imported_at_old",
+        [
+            (0, False, False, False),  # too young
+            (0, True, False, False),  # too young, already attributed
+            (1, False, False, True),
+            (0, True, False, True),  # already attributed
+            (0, False, True, False),  # negative amount, too young
+            (0, True, True, False),  # negative amount, too young, already attributed
+            (0, False, True, True),  # negative amount
+            (0, True, True, True),  # negative amount, already attributed
+        ],
+    )
+    def test_activities_to_return(
+        self,
+        session: Session,
+        utcnow,
+        expected,
+        set_transaction_id,
+        amount_negative,
+        imported_at_old,
+    ):
+        kwargs = {}
+
+        if amount_negative:
+            kwargs["amount"] = -1000
+        if imported_at_old:
+            kwargs["imported_at"] = utcnow.date() - timedelta(days=20)
+
+        activity = BankAccountActivityFactory.create(**kwargs)
+
+        if set_transaction_id:
+            user = UserFactory.create()
+
+            debit_account = user.account
+            credit_account = activity.bank_account.account
+            transaction = finance.simple_transaction(
+                description=activity.reference,
+                debit_account=debit_account,
+                credit_account=credit_account,
+                amount=activity.amount,
+                author=user,
+                valid_on=activity.valid_on,
+            )
+            activity.split = next(
+                split for split in transaction.splits if split.account_id == credit_account.id
+            )
+
+            session.add(activity)
+
+        activities_to_return = get_activities_to_return(session)
+
+        assert len(activities_to_return) == expected
+
+    def test_generate_sepa_xml(self, session: Session, utcnow):
+        BankAccountActivityFactory.create(imported_at=utcnow.date() - timedelta(days=20))
+        BankAccountActivityFactory.create(imported_at=utcnow.date() - timedelta(days=21))
+
+        generate_activities_return_sepaxml(get_activities_to_return(session))