32736 Update /search/affiliation_mappings endpoint to support migrated businesses

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

@@ -102,6 +102,7 @@ class FilingTypes(str, Enum):
         SPECIALRESOLUTION = "specialResolution"
         TRANSITION = "transition"
         TRANSPARENCY_REGISTER = "transparencyRegister"
+        TOMBSTONE = "lear_tombstone"
 
     class FilingTypesCompact(str, Enum):
         """Render enum for filing types with sub-types."""

legal-api/src/legal_api/services/search_service.py

@@ -16,13 +16,14 @@
 # pylint: disable=singleton-comparison ; pylint does not recognize sqlalchemy ==
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from operator import and_
+from operator import and_, or_
 from typing import Final, Optional
 
 from flask import current_app
 from requests import Request
 from sqlalchemy import func
 
+from legal_api.core.filing import Filing as CoreFiling
 from legal_api.models import Business, Filing, RegistrationBootstrap, db
 
 
@@ -86,7 +87,6 @@ class BusinessSearchService:  # pylint: disable=too-many-public-methods
     EXCLUDED_FILINGS_STATUS: Final = [
         Filing.Status.WITHDRAWN.value
     ]
-
     @staticmethod
     def check_and_get_respective_values(codes):
         """Check if codes belong to BUSINESS_TEMP_FILINGS_CORP_CODES and return the matching ones."""
@@ -316,19 +316,44 @@ def get_affiliation_mapping_results(identifiers):
                 nr_identifiers.append(identifier)
             else:
                 business_identifiers.append(identifier)
-        conditions = []
+        draft_conditions = []
         if business_identifiers:
-            conditions.append(Business._identifier.in_(business_identifiers))  # pylint: disable=protected-access
+            draft_conditions.append(Business._identifier.in_(business_identifiers))  # pylint: disable=protected-access
         if nr_identifiers:
-            conditions.append(Filing
+            draft_conditions.append(Filing
                               .filing_json["filing"][Filing._filing_type]  # pylint: disable=protected-access
                               ["nameRequest"]["nrNumber"]
                               .astext.in_(nr_identifiers))
         if temp_identifiers:
-            conditions.append(RegistrationBootstrap._identifier.in_(identifiers))  # pylint: disable=protected-access
-        query = query.filter(db.or_(*conditions))
-
-        rows = query.all()
-        result_list = [dict(row) for row in rows]
-
-        return result_list
+            draft_conditions.append(RegistrationBootstrap._identifier.in_(identifiers))  # pylint: disable=protected-access
+
+        if not draft_conditions:
+            return []
+
+        draft_query = query.filter(db.or_(*draft_conditions))
+
+        draft_rows = draft_query.all()
+        result_list = []
+        for row in draft_rows:
+            result_list.append({
+                "identifier": row.identifier,
+                "nrNumber": row.nrNumber,
+                "bootstrapIdentifier": row.bootstrapIdentifier
+            })
+        migrated_identifiers = db.session.query( Business._identifier.label("identifier"),  # pylint: disable=protected-access
+            Filing
+            ._filing_type  # pylint: disable=protected-access
+            .label("filing_type"),
+        ).select_from(Filing) \
+            .join(Business, Filing.business_id == Business.id).filter(Business._identifier.in_(business_identifiers))
+        migrated_rows = migrated_identifiers.filter(
+            or_(Filing._filing_type == CoreFiling.FilingTypes.TOMBSTONE.value,
+                and_(Business.legal_type.in_(["SP", "GP"]),Business.identifier.like("FM0%")))
+            ).all()
+        for row in migrated_rows:
+            result_list.append({
+                    "identifier": row.identifier,
+                    "nrNumber": None,
+                    "bootstrapIdentifier": None
+                })
+        return result_list

legal-api/tests/unit/resources/v2/test_business.py

@@ -36,7 +36,7 @@
 from legal_api.services import flags
 from legal_api.utils.datetime import datetime
 from tests import integration_affiliation
-from tests.unit.models import factory_batch, factory_batch_processing, factory_business, factory_pending_filing
+from tests.unit.models import factory_batch, factory_batch_processing, factory_business, factory_filing, factory_pending_filing
 from tests.unit.services.warnings import create_business
 from tests.unit.services.utils import create_header
 from tests.unit.models import factory_completed_filing
@@ -732,6 +732,173 @@ def test_post_affiliated_businesses_invalid(session, client, jwt):
     assert rv.status_code == HTTPStatus.BAD_REQUEST
 
 
+def _create_affiliation_mapping_draft(identifier,
+                                      filing_name='registration',
+                                      legal_type=Business.LegalTypes.SOLE_PROP.value,
+                                      nr_number=None,
+                                      legal_name='Test NR Name'):
+    """Create a draft filing with a temp registration for affiliation mapping tests."""
+    temp_reg = RegistrationBootstrap()
+    temp_reg._identifier = identifier
+    temp_reg.save()
+
+    json_data = copy.deepcopy(FILING_HEADER)
+    json_data['filing']['header']['name'] = filing_name
+    json_data['filing']['header']['identifier'] = identifier
+    del json_data['filing']['business']
+    json_data['filing'][filing_name] = {
+        'nameRequest': {
+            'legalType': legal_type
+        }
+    }
+
+    if nr_number:
+        json_data['filing'][filing_name]['nameRequest'] = {
+            **json_data['filing'][filing_name]['nameRequest'],
+            'nrNumber': nr_number,
+            'legalName': legal_name
+        }
+
+    filing = factory_pending_filing(None, json_data)
+    filing.temp_reg = identifier
+    filing.save()
+
+
+def test_post_affiliation_mappings_migrated_business_without_bootstrap(session, client, jwt):
+    """Assert that direct business identifiers return a mapping even without bootstrap-linked filings."""
+    identifier = 'BC1328381'
+    business = factory_business_model(legal_name=identifier + 'name',
+                           identifier=identifier,
+                           founding_date=datetime.utcfromtimestamp(0),
+                           last_ledger_timestamp=datetime.utcfromtimestamp(0),
+                           last_modified=datetime.utcfromtimestamp(0),
+                           fiscal_year_end_date=None,
+                           tax_id=None,
+                           dissolution_date=None,
+                           legal_type=Business.LegalTypes.BCOMP.value)
+    factory_filing(business,{'filing': {
+        'header': {
+            'name': 'lear_tombstone'
+        }
+    }}, filing_type='lear_tombstone')
+
+    rv = client.post('/api/v2/businesses/search/affiliation_mappings',
+                     json={'identifiers': [identifier]},
+                     headers=create_header(jwt, [SYSTEM_ROLE]))
+
+    assert rv.status_code == HTTPStatus.OK
+    assert rv.json['count'] == 1
+    assert rv.json['entityDetails'] == [{
+        'identifier': identifier,
+        'nrNumber': None,
+        'bootstrapIdentifier': None
+    }]
+
+
+def test_post_affiliation_mappings_temp_identifier(session, client, jwt):
+    """Assert that temp identifiers still resolve through the filing/bootstrap path."""
+    identifier = 'Tb31yQIuC1'
+    _create_affiliation_mapping_draft(identifier=identifier,
+                                      filing_name='incorporationApplication',
+                                      legal_type=Business.LegalTypes.BCOMP.value)
+
+    rv = client.post('/api/v2/businesses/search/affiliation_mappings',
+                     json={'identifiers': [identifier]},
+                     headers=create_header(jwt, [SYSTEM_ROLE]))
+
+    assert rv.status_code == HTTPStatus.OK
+    assert rv.json['count'] == 1
+    detail = rv.json['entityDetails'][0]
+    assert detail['bootstrapIdentifier'] == identifier
+    assert detail['nrNumber'] is None
+
+
+def test_post_affiliation_mappings_nr_identifier(session, client, jwt):
+    """Assert that NR identifiers still resolve through the filing/bootstrap path."""
+    bootstrap_identifier = 'Tb31yQIuC2'
+    nr_number = 'NR 1234567'
+    _create_affiliation_mapping_draft(identifier=bootstrap_identifier,
+                                      filing_name='registration',
+                                      legal_type=Business.LegalTypes.SOLE_PROP.value,
+                                      nr_number=nr_number,
+                                      legal_name='Test NR Name')
+
+    rv = client.post('/api/v2/businesses/search/affiliation_mappings',
+                     json={'identifiers': [nr_number]},
+                     headers=create_header(jwt, [SYSTEM_ROLE]))
+
+    assert rv.status_code == HTTPStatus.OK
+    assert rv.json['count'] == 1
+    detail = rv.json['entityDetails'][0]
+    assert detail['nrNumber'] == nr_number
+    assert detail['bootstrapIdentifier'] == bootstrap_identifier
+
+
+def test_post_affiliation_mappings_mixed_direct_and_nr_identifiers(session, client, jwt):
+    """Assert that direct business and draft-backed NR lookups coexist in a single request."""
+    business_identifier = 'BC7654321'
+    nr_number = 'NR 1234568'
+    bootstrap_identifier = 'Tb31yQIuC3'
+
+    business = factory_business_model(legal_name=business_identifier + 'name',
+                           identifier=business_identifier,
+                           founding_date=datetime.utcfromtimestamp(0),
+                           last_ledger_timestamp=datetime.utcfromtimestamp(0),
+                           last_modified=datetime.utcfromtimestamp(0),
+                           fiscal_year_end_date=None,
+                           tax_id=None,
+                           dissolution_date=None,
+                           legal_type=Business.LegalTypes.BCOMP.value)
+
+    factory_filing(business,{'filing': {
+        'header': {
+            'name': 'lear_tombstone'
+        }
+    }}, filing_type='lear_tombstone')
+
+    _create_affiliation_mapping_draft(identifier=bootstrap_identifier,
+                                      filing_name='registration',
+                                      legal_type=Business.LegalTypes.SOLE_PROP.value,
+                                      nr_number=nr_number,
+                                      legal_name='Mixed NR Name')
+
+    rv = client.post('/api/v2/businesses/search/affiliation_mappings',
+                     json={'identifiers': [business_identifier, nr_number]},
+                     headers=create_header(jwt, [SYSTEM_ROLE]))
+
+    assert rv.status_code == HTTPStatus.OK
+    assert rv.json['count'] == 2
+
+    business_detail = next(
+        detail for detail in rv.json['entityDetails']
+        if detail['identifier'] == business_identifier
+    )
+    nr_detail = next(
+        detail for detail in rv.json['entityDetails']
+        if detail['nrNumber'] == nr_number
+    )
+
+    assert business_detail['bootstrapIdentifier'] is None
+    assert business_detail['nrNumber'] is None
+    assert nr_detail['bootstrapIdentifier'] == bootstrap_identifier
+
+
+def test_post_affiliation_mappings_unauthorized(session, client, jwt):
+    """Assert that the affiliation mappings endpoint is unauthorized for non-system tokens."""
+    rv = client.post('/api/v2/businesses/search/affiliation_mappings',
+                     json={'identifiers': ['BC1234567']},
+                     headers=create_header(jwt, [STAFF_ROLE]))
+    assert rv.status_code == HTTPStatus.UNAUTHORIZED
+
+
+def test_post_affiliation_mappings_invalid(session, client, jwt):
+    """Assert that the affiliation mappings endpoint is bad request when identifiers are not given."""
+    rv = client.post('/api/v2/businesses/search/affiliation_mappings',
+                     json={},
+                     headers=create_header(jwt, [SYSTEM_ROLE]))
+    assert rv.status_code == HTTPStatus.BAD_REQUEST
+
+
 def test_get_could_file(session, client, jwt, monkeypatch):
     """Assert that the cold file is returned."""
     monkeypatch.setattr(