19331 - eft short name linking audit columns (#1398)

pay-api/migrations/versions/2024_02_02_bbd1189e4e75_.py

@@ -0,0 +1,38 @@
+"""EFT Short names add audit fields for linking
+
+Revision ID: bbd1189e4e75
+Revises: 4f87e62bc81e
+Create Date: 2024-02-02 09:41:11.363804
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'bbd1189e4e75'
+down_revision = '4f87e62bc81e'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('eft_short_names', sa.Column('linked_by', sa.String(length=100), nullable=True))
+    op.add_column('eft_short_names', sa.Column('linked_by_name', sa.String(length=100), nullable=True))
+    op.add_column('eft_short_names', sa.Column('linked_on', sa.DateTime(), nullable=True))
+    op.add_column('eft_short_names_version', sa.Column('linked_by', sa.String(length=100), autoincrement=False, nullable=True))
+    op.add_column('eft_short_names_version', sa.Column('linked_by_name', sa.String(length=100), autoincrement=False, nullable=True))
+    op.add_column('eft_short_names_version', sa.Column('linked_on', sa.DateTime(), autoincrement=False, nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('eft_short_names_version', 'linked_on')
+    op.drop_column('eft_short_names_version', 'linked_by_name')
+    op.drop_column('eft_short_names_version', 'linked_by')
+    op.drop_column('eft_short_names', 'linked_on')
+    op.drop_column('eft_short_names', 'linked_by_name')
+    op.drop_column('eft_short_names', 'linked_by')
+    # ### end Alembic commands ###

pay-api/src/pay_api/models/eft_short_names.py

@@ -49,6 +49,9 @@ class EFTShortnames(VersionedModel):  # pylint: disable=too-many-instance-attrib
     auth_account_id = db.Column('auth_account_id', db.String(50), nullable=True, index=True)
     created_on = db.Column('created_on', db.DateTime, nullable=False, default=datetime.now)
     short_name = db.Column('short_name', db.String, nullable=False, index=True)
+    linked_by = db.Column('linked_by', db.String(100), nullable=True)
+    linked_by_name = db.Column('linked_by_name', db.String(100), nullable=True)
+    linked_on = db.Column('linked_on', db.DateTime, nullable=True)
 
     @classmethod
     def find_by_short_name(cls, short_name: str):
@@ -70,6 +73,9 @@ class EFTShortnameSchema:  # pylint: disable=too-few-public-methods
     transaction_date: datetime
     deposit_date: datetime
     deposit_amount: Decimal
+    linked_by: str
+    linked_by_name: str
+    linked_on: datetime
 
     @classmethod
     def from_row(cls, row: EFTShortnames):
@@ -86,4 +92,8 @@ def from_row(cls, row: EFTShortnames):
                    transaction_id=getattr(row, 'transaction_id', None),
                    transaction_date=getattr(row, 'transaction_date', None),
                    deposit_date=getattr(row, 'deposit_date', None),
-                   deposit_amount=cents_to_decimal(getattr(row, 'deposit_amount', None)))
+                   deposit_amount=cents_to_decimal(getattr(row, 'deposit_amount', None)),
+                   linked_by=row.linked_by,
+                   linked_by_name=row.linked_by_name,
+                   linked_on=row.linked_on
+                   )

pay-api/src/pay_api/services/eft_short_names.py

@@ -35,6 +35,7 @@
 from pay_api.utils.converter import Converter
 from pay_api.utils.enums import EFTFileLineType, EFTProcessStatus, EFTShortnameState, InvoiceStatus, PaymentMethod
 from pay_api.utils.errors import Error
+from pay_api.utils.user_context import user_context
 
 
 @dataclass
@@ -151,7 +152,8 @@ def update(cls, short_name_id: str, short_name_request: Dict[str, Any]) -> EFTSh
         return cls.find_by_short_name_id(short_name.id)
 
     @classmethod
-    def patch(cls, short_name_id: int, auth_account_id: str) -> EFTShortnames:
+    @user_context
+    def patch(cls, short_name_id: int, auth_account_id: str, **kwargs) -> EFTShortnames:
         """Patch eft short name auth account mapping."""
         current_app.logger.debug('<patch eft short name mapping')
 
@@ -165,6 +167,9 @@ def patch(cls, short_name_id: int, auth_account_id: str) -> EFTShortnames:
             raise BusinessException(Error.EFT_SHORT_NAME_ALREADY_MAPPED)
 
         short_name.auth_account_id = auth_account_id
+        short_name.linked_by = kwargs['user'].user_name
+        short_name.linked_by_name = kwargs['user'].name
+        short_name.linked_on = datetime.now()
         short_name.save()
 
         # Update any existing credit mappings with the payment account
@@ -279,6 +284,9 @@ def get_search_query(cls, search_criteria: EFTShortnamesSearch, is_count: bool =
                                  EFTShortnameModel.short_name,
                                  EFTShortnameModel.auth_account_id,
                                  EFTShortnameModel.created_on,
+                                 EFTShortnameModel.linked_by,
+                                 EFTShortnameModel.linked_by_name,
+                                 EFTShortnameModel.linked_on,
                                  case(
                                      [(PaymentAccountModel.name.like('%-' + PaymentAccountModel.branch_name),
                                        func.replace(PaymentAccountModel.name, '-' + PaymentAccountModel.branch_name, '')
@@ -303,7 +311,7 @@ def get_search_query(cls, search_criteria: EFTShortnamesSearch, is_count: bool =
             query = query.filter_conditionally(search_criteria.deposit_date, sub_query.c.deposit_date)
             query = query.filter_conditionally(search_criteria.deposit_amount, sub_query.c.deposit_amount_cents)
             query = query.filter_conditionally(search_criteria.account_id,
-                                               PaymentAccountModel.auth_account_id, is_like=True)
+                                               EFTShortnameModel.auth_account_id, is_like=True)
             query = query.filter_conditionally(search_criteria.account_name, PaymentAccountModel.name, is_like=True)
             query = query.filter_conditionally(search_criteria.account_branch, PaymentAccountModel.branch_name,
                                                is_like=True)
@@ -320,7 +328,18 @@ def get_search_query(cls, search_criteria: EFTShortnamesSearch, is_count: bool =
 
         # Short name free text search
         query = query.filter_conditionally(search_criteria.short_name, EFTShortnameModel.short_name, is_like=True)
-        query = query.order_by(sub_query.c.transaction_date) if sub_query is not None else query
+        query = cls.get_order_by(search_criteria, query, sub_query)
+
+        return query
+
+    @classmethod
+    def get_order_by(cls, search_criteria, query, sub_query):
+        """Get the order by for search query."""
+        if search_criteria.state == EFTShortnameState.LINKED.value:
+            return query.order_by(EFTShortnameModel.linked_on.desc())
+
+        if search_criteria.state == EFTShortnameState.UNLINKED.value and sub_query is not None:
+            return query.order_by(sub_query.c.transaction_date.desc())
 
         return query
 

pay-api/tests/unit/api/test_eft_short_names.py

@@ -36,7 +36,8 @@
 
 def test_patch_eft_short_name(session, client, jwt, app):
     """Assert that an EFT short name account id can be patched."""
-    token = jwt.create_jwt(get_claims(roles=[Role.STAFF.value]), token_header)
+    token = jwt.create_jwt(get_claims(roles=[Role.STAFF.value],
+                                      username='IDIR/JSMITH'), token_header)
     headers = {'Authorization': f'Bearer {token}', 'content-type': 'application/json'}
     factory_payment_account(payment_method_code=PaymentMethod.EFT.value,
                             auth_account_id='1234').save()
@@ -51,6 +52,10 @@ def test_patch_eft_short_name(session, client, jwt, app):
     assert shortname_dict['id'] is not None
     assert shortname_dict['shortName'] == 'TESTSHORTNAME'
     assert shortname_dict['accountId'] == '1234'
+    assert shortname_dict['linkedBy'] == 'IDIR/JSMITH'
+
+    date_format = '%Y-%m-%dT%H:%M:%S.%f'
+    assert datetime.strptime(shortname_dict['linkedOn'], date_format).date() == datetime.now().date()
 
 
 def test_patch_eft_short_name_validation(session, client, jwt, app):