diff --git a/legal-api/src/legal_api/core/filing.py b/legal-api/src/legal_api/core/filing.py index 082c553363..3c450e43cc 100644 --- a/legal-api/src/legal_api/core/filing.py +++ b/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.""" diff --git a/legal-api/src/legal_api/services/search_service.py b/legal-api/src/legal_api/services/search_service.py index 93d4e649fd..1af3c4c529 100644 --- a/legal-api/src/legal_api/services/search_service.py +++ b/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 \ No newline at end of file diff --git a/legal-api/tests/unit/resources/v2/test_business.py b/legal-api/tests/unit/resources/v2/test_business.py index cc67e6c207..ee601b5ecd 100644 --- a/legal-api/tests/unit/resources/v2/test_business.py +++ b/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(