23539 add batch_processing to colin_event_ids table

legal-api/migrations/versions/83a110e10979_add_batch_processing_id_to_colin_event_.py

@@ -0,0 +1,28 @@
+"""add_batch_processing_id_to_colin_event_ids
+
+Revision ID: 83a110e10979
+Revises: d55dfc5c1358
+Create Date: 2024-10-07 15:34:55.319366
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '83a110e10979'
+down_revision = 'd55dfc5c1358'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('colin_event_ids', sa.Column('batch_processing_id', sa.Integer(), nullable=True))
+    op.add_column('colin_event_ids', sa.Column('batch_processing_step', sa.Enum('WARNING_LEVEL_1', 'WARNING_LEVEL_2', 'DISSOLUTION', name='batch_processing_step'), nullable=True))
+    op.create_foreign_key('colin_event_ids_batch_processing_id_fkey', 'colin_event_ids', 'batch_processing', ['batch_processing_id'], ['id'])
+
+
+def downgrade():
+    op.drop_constraint('colin_event_ids_batch_processing_id_fkey', 'colin_event_ids', type_='foreignkey')
+    op.drop_column('colin_event_ids', 'batch_processing_id')
+    op.drop_column('colin_event_ids', 'batch_processing_step')

legal-api/src/legal_api/models/colin_event_id.py

@@ -15,6 +15,7 @@
 
 The ColinEventId class and Schema are held in this module.
 """
+from legal_api.models import BatchProcessing
 
 from .db import db
 
@@ -26,6 +27,11 @@ class ColinEventId(db.Model):  # pylint: disable=too-few-public-methods
 
     colin_event_id = db.Column('colin_event_id', db.Integer, unique=True, primary_key=True)
     filing_id = db.Column('filing_id', db.Integer, db.ForeignKey('filings.id'))
+    batch_processing_id = db.Column('batch_processing_id', db.Integer, db.ForeignKey('batch_processing.id'))
+    batch_processing_step = db.Column('batch_processing_step', db.Enum(BatchProcessing.BatchProcessingStep))
+
+    # relationships
+    batch_processing = db.relationship('BatchProcessing', lazy='select', uselist=False)
 
     def save(self):
         """Save the object to the database immediately."""
@@ -47,3 +53,13 @@ def get_by_colin_id(colin_id):
         colin_event_id_obj =\
             db.session.query(ColinEventId).filter(ColinEventId.colin_event_id == colin_id).one_or_none()
         return colin_event_id_obj
+
+    @staticmethod
+    def get_by_batch_processing_id(batch_processing_id):
+        """Get the list of colin_event_ids linked to the given batch_processing_id."""
+        colin_event_id_objs = db.session.query(ColinEventId). \
+            filter(ColinEventId.batch_processing_id == batch_processing_id).all()
+        id_list = []
+        for obj in colin_event_id_objs:
+            id_list.append(obj.colin_event_id)
+        return id_list

legal-api/tests/unit/models/test_colin_event_id.py

@@ -0,0 +1,101 @@
+# Copyright © 2024 Province of British Columbia
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests to assure the Colin Event Id Model.
+
+Test-Suite to ensure that the Colin Event Id Model is working as expected.
+"""
+from legal_api.models.colin_event_id import ColinEventId
+from registry_schemas.example_data import ANNUAL_REPORT
+
+from tests.unit.models import factory_business, factory_batch, factory_batch_processing, factory_filing
+
+
+def test_valid_colin_event_id_save(session):
+    """Assert that a valid Colin Event Id can be saved."""
+    business_identifier = 'BC1234567'
+    business = factory_business(business_identifier)
+    filing = factory_filing(business, ANNUAL_REPORT)
+    colin_event_id = ColinEventId(filing_id=filing.id)
+    colin_event_id.save()
+    assert colin_event_id.colin_event_id
+
+    # Save with batch_processing
+    batch = factory_batch()
+    batch_processing = factory_batch_processing(
+        batch_id=batch.id,
+        business_id=business.id,
+        identifier=business_identifier
+    )
+    colin_event_id.batch_processing_id = batch_processing.id
+    colin_event_id.batch_processing_step = batch_processing.step
+    colin_event_id.save()
+    assert colin_event_id.batch_processing_id
+    assert colin_event_id.batch_processing_step
+    assert colin_event_id.batch_processing
+
+
+def test_get_by_colin_id(session):
+    """Assert that the method returns correct value."""
+    business_identifier = 'BC1234567'
+    business = factory_business(business_identifier)
+    filing = factory_filing(business, ANNUAL_REPORT)
+    colin_event_id = ColinEventId(filing_id=filing.id)
+    colin_event_id.save()
+
+    res = ColinEventId.get_by_colin_id(colin_event_id.colin_event_id)
+
+    assert res
+    assert res.filing_id == colin_event_id.filing_id
+
+
+def test_get_by_filing_id(session):
+    """Assert that the method returns correct value."""
+    business_identifier = 'BC1234567'
+    business = factory_business(business_identifier)
+    filing = factory_filing(business, ANNUAL_REPORT)
+    colin_event_id = ColinEventId(filing_id=filing.id)
+    colin_event_id.save()
+
+    res = ColinEventId.get_by_filing_id(filing.id)
+
+    assert res
+    assert len(res) == 1
+    assert res[0] == colin_event_id.colin_event_id
+
+
+def test_get_by_batch_processing_id(session):
+    """Assert that the method returns correct value."""
+    business_identifier = 'BC1234567'
+    business = factory_business(business_identifier)
+    filing = factory_filing(business, ANNUAL_REPORT)
+
+    batch = factory_batch()
+    batch_processing = factory_batch_processing(
+        batch_id=batch.id,
+        business_id=business.id,
+        identifier=business_identifier
+    )
+    colin_event_id = ColinEventId(
+        filing_id=filing.id,
+        batch_processing_id = batch_processing.id,
+        batch_processing_step = batch_processing.step
+    )
+    colin_event_id.save()
+
+    res = ColinEventId.get_by_batch_processing_id(batch_processing.id)
+
+    assert res
+    assert len(res) == 1
+    assert res[0] == colin_event_id.colin_event_id