20179 get amalgamated into from version data (#2519)

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

@@ -19,6 +19,9 @@
 
 from enum import auto
 
+from sqlalchemy import or_
+from sqlalchemy_continuum import version_class
+
 from ..utils.base import BaseEnum
 from .db import db
 
@@ -74,3 +77,32 @@ def json(self):
             'identifier': business.identifier,
             'legalName': business.legal_name
         }
+
+    @classmethod
+    def get_amalgamation_revision_obj(cls, transaction_id, business_id):
+        """Get amalgamation for the given transaction id."""
+        # pylint: disable=singleton-comparison;
+        amalgamation_version = version_class(Amalgamation)
+        amalgamation = db.session.query(amalgamation_version) \
+            .filter(amalgamation_version.transaction_id <= transaction_id) \
+            .filter(amalgamation_version.operation_type != 2) \
+            .filter(amalgamation_version.business_id == business_id) \
+            .filter(or_(amalgamation_version.end_transaction_id == None,  # noqa: E711;
+                        amalgamation_version.end_transaction_id > transaction_id)) \
+            .order_by(amalgamation_version.transaction_id).one_or_none()
+        return amalgamation
+
+    @classmethod
+    def get_amalgamation_revision_json(cls, transaction_id, business_id):
+        """Get amalgamation json for the given transaction id."""
+        amalgamation = Amalgamation.get_amalgamation_revision_obj(transaction_id, business_id)
+        from .business import Business  # pylint: disable=import-outside-toplevel
+        business = Business.find_by_internal_id(amalgamation.business_id)
+
+        return {
+            'amalgamationDate': amalgamation.amalgamation_date.isoformat(),
+            'amalgamationType': amalgamation.amalgamation_type.name,
+            'courtApproval': amalgamation.court_approval,
+            'identifier': business.identifier,
+            'legalName': business.legal_name
+        }

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

@@ -437,10 +437,8 @@ def _extend_json(self, d):
         if self.fiscal_year_end_date:
             d['fiscalYearEndDate'] = datetime.date(self.fiscal_year_end_date).isoformat()
         if self.state_filing_id:
-            if (self.state == Business.State.HISTORICAL and
-                    (amalgamating_business := self.amalgamating_businesses.one_or_none())):
-                amalgamation = Amalgamation.find_by_id(amalgamating_business.amalgamation_id)
-                d['amalgamatedInto'] = amalgamation.json()
+            if (amalgamated_into := self.get_amalgamated_into()):
+                d['amalgamatedInto'] = amalgamated_into
             else:
                 d['stateFiling'] = f'{base_url}/{self.identifier}/filings/{self.state_filing_id}'
 
@@ -562,6 +560,16 @@ def get_filing_by_id(cls, business_identifier: int, filing_id: str):
             one_or_none()
         return None if not filing else filing[1]
 
+    def get_amalgamated_into(self) -> dict:
+        """Get amalgamated into if this business is part of an amalgamation."""
+        if (self.state == Business.State.HISTORICAL and
+            (state_filing := Filing.find_by_id(self.state_filing_id)) and
+                state_filing.is_amalgamation_application):
+            return Amalgamation.get_amalgamation_revision_json(state_filing.transaction_id,
+                                                               state_filing.business_id)
+
+        return None
+
     @classmethod
     def is_pending_amalgamating_business(cls, business_identifier):
         """Check if a business has a pending amalgamation with the provided business identifier."""

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

@@ -665,6 +665,11 @@ def is_correction_pending(self):
             return True
         return False
 
+    @property
+    def is_amalgamation_application(self):
+        """Is this an amalgamation application filing."""
+        return self.filing_type == Filing.FILINGS['amalgamationApplication'].get('name')
+
     @hybrid_property
     def comments_count(self):
         """Return the number of commentson this filing."""

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

@@ -21,9 +21,10 @@
 
 import datedelta
 import pytest
+from sqlalchemy_continuum import versioning_manager
 
 from legal_api.exceptions import BusinessException
-from legal_api.models import AmalgamatingBusiness, Amalgamation, Business, Filing
+from legal_api.models import AmalgamatingBusiness, Amalgamation, Business, Filing, db
 from legal_api.utils.legislation_datetime import LegislationDatetime
 from tests import EPOCH_DATETIME, TIMEZONE_OFFSET
 from tests.unit import has_expected_date_str_format
@@ -411,21 +412,26 @@ def test_continued_in_business(session):
 ])
 def test_amalgamated_into_business_json(session, test_name, existing_business_state):
     """Assert that the amalgamated into is in json."""
-    filing = Filing()
-    filing.save()
-
     existing_business = Business(
         legal_name='Test - Amalgamating Legal Name',
         legal_type='BC',
         founding_date=datetime.utcfromtimestamp(0),
         dissolution_date=datetime.now(),
         identifier='BC1234567',
-        state=existing_business_state,
-        state_filing_id=filing.id
+        state=Business.State.ACTIVE,
     )
     existing_business.save()
 
     if test_name == 'EXIST':
+
+        filing = Filing()
+        filing._filing_type = 'amalgamationApplication'
+        filing.save()
+
+        # Versioning business
+        uow = versioning_manager.unit_of_work(db.session)
+        transaction = uow.create_transaction(db.session)
+
         business = Business(
             legal_name='Test - Legal Name',
             legal_type='BC',
@@ -445,7 +451,15 @@ def test_amalgamated_into_business_json(session, test_name, existing_business_st
         amalgamation.amalgamating_businesses.append(amalgamating_business)
 
         business.amalgamation.append(amalgamation)
-        business.save()
+        db.session.add(business)
+        existing_business.state_filing_id = filing.id
+        existing_business.state = existing_business_state
+        db.session.add(existing_business)
+        db.session.commit()
+
+        filing.transaction_id = transaction.id
+        filing.business_id = business.id
+        filing.save()
 
     business_json = existing_business.json()
 
@@ -459,4 +473,3 @@ def test_amalgamated_into_business_json(session, test_name, existing_business_st
         assert business_json['amalgamatedInto']['legalName'] == business.legal_name
     else:
         assert not 'amalgamatedInto' in business_json
-        assert 'stateFiling' in business_json