diff --git a/legal-api/pyproject.toml b/legal-api/pyproject.toml index 40525ee5bf..b151dcd4b1 100644 --- a/legal-api/pyproject.toml +++ b/legal-api/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "legal-api" -version = "3.1.42" +version = "3.1.44" description = "" authors = [ {name = "thor",email = "1042854+thorwolpert@users.noreply.github.com"} diff --git a/legal-api/src/legal_api/resources/v2/business/business_tasks.py b/legal-api/src/legal_api/resources/v2/business/business_tasks.py index 8e25f191d8..c93ef5b3d9 100644 --- a/legal-api/src/legal_api/resources/v2/business/business_tasks.py +++ b/legal-api/src/legal_api/resources/v2/business/business_tasks.py @@ -28,6 +28,7 @@ from business_common.utils.legislation_datetime import LegislationDatetime from business_model.models import Business, Filing from legal_api.services import check_warnings, namex +from legal_api.services.authz import BusinessBlocker, business_blocker_check from legal_api.services.request_context import add_account_linking_key_header from legal_api.services.warnings.business.business_checks import BusinessWarningCodes, WarningType from legal_api.utils.auth import jwt @@ -80,7 +81,7 @@ def get_tasks(identifier): return jsonify(tasks=rv) -def construct_task_list(business: Business): # noqa: PLR0915 +def construct_task_list(business: Business): """ Return all current pending tasks to do. @@ -97,32 +98,11 @@ def construct_task_list(business: Business): # noqa: PLR0915 - Corporations must file one AR per year, on or after the anniversary of the founding date """ - # Legal types that never get AR todos. SP/GP don't file annual reports at all. The rest are - # legacy/mainframe migrated legal types where annualReport filing is disabled/greyed out in the - # UI - some of these were founded decades ago (for example,1940s), so without this exclusion they'd - # generate a large backlog of AR todos that can never actually be filed. - entity_types_no_ar = [ - "SP", "GP", - "RLY", # Railways - "LIB", # Public Library Association - "CEM", # Cemetery - "TMY", # Tramways - "PFS", # Pension Funded Society - "SB", # Society Branch - ] tasks = [] order = 1 warnings = check_warnings(business) - if any(x["warningType"] == WarningType.MISSING_REQUIRED_BUSINESS_INFO for x in warnings): - # TODO remove compliance warning line when UI has been integrated to use warnings instead of complianceWarnings - business.compliance_warnings = warnings - business.warnings = warnings - - # Checking for draft or pending conversion - if not Filing.get_incomplete_filings_by_type(business.id, "conversion"): - tasks.append(create_conversion_filing_todo(business, order, True)) - order += 1 + tasks, order = _add_missing_info_task(business, warnings, tasks, order) # Retrieve filings that are either incomplete, or drafts pending_filings = Filing.get_filings_by_status(business.id, [Filing.Status.DRAFT.value, @@ -135,92 +115,156 @@ def construct_task_list(business: Business): # noqa: PLR0915 if filing.filing_type == "transparencyRegister": pending_tr_type = filing.filing_sub_type - filing_json = filing.json - if filing.payment_status_code == "CREATED" and filing.payment_token: - # get current pay details from pay-api - try: - headers = { - "Authorization": f"Bearer {jwt.get_token_auth_header()}", - "Content-Type": "application/json" - } - add_account_linking_key_header(headers) - pay_response = requests.get( - url=f'{current_app.config.get("PAYMENT_SVC_URL")}/{filing.payment_token}', - headers=headers - ) - pay_details = { - "isPaymentActionRequired": pay_response.json().get("isPaymentActionRequired", False), - "paymentMethod": pay_response.json().get("paymentMethod", "") - } - filing_json["filing"]["header"].update(pay_details) + if not (task := _create_pending_filing_task(business, filing, order)): + return "pay_connection_error" + tasks.append(task) + order += 1 - except (exceptions.ConnectionError, exceptions.Timeout) as err: - current_app.logger.error( - f"Payment connection failure for {business.identifier} task list. ", err) - return "pay_connection_error" + tasks, order = _add_ar_tasks(business, tasks, order, pending_filings) - task = {"task": filing_json, "order": order, "enabled": True} - tasks.append(task) + tasks, order = add_tr_tasks(business, tasks, order, pending_tr_type) + + tasks, order = _add_transition_task(business, warnings, tasks, order, pending_filings) + + return tasks + + +def _add_missing_info_task(business: Business, warnings: list, tasks: list, order: int): + """Add the todo that lets the business fix missing required info (firms: conversion, corps: reviewImportedData).""" + if not any(x["warningType"] == WarningType.MISSING_REQUIRED_BUSINESS_INFO for x in warnings): + return tasks, order + + # TODO remove compliance warning line when UI has been integrated to use warnings instead of complianceWarnings + business.compliance_warnings = warnings + business.warnings = warnings + + if business.legal_type in (Business.LegalTypes.SOLE_PROP.value, Business.LegalTypes.PARTNERSHIP.value): + # Checking for draft or pending conversion + if not Filing.get_incomplete_filings_by_type(business.id, "conversion"): + tasks.append(create_conversion_filing_todo(business, order, True)) + order += 1 + elif ( + business.legal_type in Business.CORPS + and not Filing.get_incomplete_filings_by_type(business.id, "reviewImportedData") + ): + # reviewImportedData is gated by BusinessBlocker.DEFAULT in authz, so disable the todo when the API would reject it + enabled = not business_blocker_check(business)[BusinessBlocker.DEFAULT] + tasks.append(create_review_imported_data_todo(business, order, enabled)) order += 1 + return tasks, order + + +def _create_pending_filing_task(business: Business, filing: Filing, order: int) -> dict | None: + """Return the task for a draft/pending filing, or None when pay-api could not be reached.""" + filing_json = filing.json + if filing.payment_status_code == "CREATED" and filing.payment_token: + # get current pay details from pay-api + try: + headers = { + "Authorization": f"Bearer {jwt.get_token_auth_header()}", + "Content-Type": "application/json" + } + add_account_linking_key_header(headers) + pay_response = requests.get( + url=f'{current_app.config.get("PAYMENT_SVC_URL")}/{filing.payment_token}', + headers=headers + ) + pay_details = { + "isPaymentActionRequired": pay_response.json().get("isPaymentActionRequired", False), + "paymentMethod": pay_response.json().get("paymentMethod", "") + } + filing_json["filing"]["header"].update(pay_details) + + except (exceptions.ConnectionError, exceptions.Timeout) as err: + current_app.logger.error( + f"Payment connection failure for {business.identifier} task list. ", err) + return None + + return {"task": filing_json, "order": order, "enabled": True} + + +def _add_ar_tasks(business: Business, tasks: list, order: int, pending_filings: list): + """Add the Annual Report todos for every AR year that is due and not yet started.""" + # Legal types that never get AR todos. SP/GP don't file annual reports at all. The rest are + # legacy/mainframe migrated legal types where annualReport filing is disabled/greyed out in the + # UI - some of these were founded decades ago (for example,1940s), so without this exclusion they'd + # generate a large backlog of AR todos that can never actually be filed. + entity_types_no_ar = [ + "SP", "GP", + "RLY", # Railways + "LIB", # Public Library Association + "CEM", # Cemetery + "TMY", # Tramways + "PFS", # Pension Funded Society + "SB", # Society Branch + ] # Also skip AR todos when founding date is the COLIN year-1 sentinel/unknown - without this, a # sentinel founding_date of 0001-01-01 generates ~2000 AR todos and a multi-MB /tasks payload. unknown_founding_date = ( not business.founding_date or business.founding_date.year <= 1 ) if ( - business.legal_type not in entity_types_no_ar - and not business.in_liquidation - and not unknown_founding_date + business.legal_type in entity_types_no_ar + or business.in_liquidation + or unknown_founding_date ): - # If this is the first calendar year since incorporation, there is no previous ar year. - next_ar_year = (business.last_ar_year if business.last_ar_year else business.founding_date.year) + 1 + return tasks, order + + # If this is the first calendar year since incorporation, there is no previous ar year. + next_ar_year = (business.last_ar_year if business.last_ar_year else business.founding_date.year) + 1 + + # Checking for pending ar + annual_report_filings = Filing.get_incomplete_filings_by_type(business.id, "annualReport") + if annual_report_filings: + # Consider each filing as each year and add to find next ar year + next_ar_year += len(annual_report_filings) - # Checking for pending ar - annual_report_filings = Filing.get_incomplete_filings_by_type(business.id, "annualReport") - if annual_report_filings: - # Consider each filing as each year and add to find next ar year - next_ar_year += len(annual_report_filings) + ar_min_date, ar_max_date = business.get_ar_dates(next_ar_year) + start_year = next_ar_year + while next_ar_year <= datetime.now(UTC).year and ar_min_date <= datetime.now(UTC).date(): + enabled = not pending_filings and ar_min_date.year == start_year + tasks.append(create_todo(business, next_ar_year, ar_min_date, ar_max_date, order, enabled)) + + # Include all ar's to todo from last ar filing + next_ar_year += 1 ar_min_date, ar_max_date = business.get_ar_dates(next_ar_year) + order += 1 - start_year = next_ar_year - while next_ar_year <= datetime.now(UTC).year and ar_min_date <= datetime.now(UTC).date(): - # while next_ar_year <= datetime.now(UTC).date(): - enabled = not pending_filings and ar_min_date.year == start_year - tasks.append(create_todo(business, next_ar_year, ar_min_date, ar_max_date, order, enabled)) + return tasks, order - # Include all ar's to todo from last ar filing - next_ar_year += 1 - ar_min_date, ar_max_date = business.get_ar_dates(next_ar_year) - order += 1 - tasks, order = add_tr_tasks(business, tasks, order, pending_tr_type) +def _add_transition_task(business: Business, warnings: list, tasks: list, order: int, pending_filings: list): + """Add the Transition Application todo when the business has a TRANSITION_NOT_FILED warning. - # Transition Application todo task appears below overdue Annual Reports before the restoration date causing the warning - # and it does not affect the 'enabled' status of other todo items. + It appears below overdue Annual Reports before the restoration date causing the warning + and it does not affect the 'enabled' status of other todo items. + """ if ( - any(x["code"] == BusinessWarningCodes.TRANSITION_NOT_FILED.value for x in warnings) and - not Filing.get_incomplete_filings_by_type(business.id, "transition") + not any(x["code"] == BusinessWarningCodes.TRANSITION_NOT_FILED.value for x in warnings) or + Filing.get_incomplete_filings_by_type(business.id, "transition") ): - # Gets all completed restorations of the business (most recent first) - restorations = Filing.get_filings_by_types(business.id, ["restoration", "restorationApplication"]) - if not restorations: - # Should never get here - current_app.logger.error(f"Error - Business id: {business.id}. TRANSITION_NOT_FILED warning and no restoration filing on record.") - return tasks - # Use the first restoration in the list for the most recent completed restoration date - last_restoration_date = restorations[0].effective_date - last_ar_date = business.last_ar_date or business.founding_date - # Get the transition application todo order based on the ar tasks, last restoration date, and ar date information - transition_order, transition_enabled = _find_task_order_for_ta(tasks, order, last_restoration_date, last_ar_date) - # Bump all the task orders by one that are at and above transition_order - tasks = _bump_task_order(tasks, transition_order) - # Append the TA task at with order: transition_order. Disable if there are any incomplete filings - tasks.append(create_transition_todo(business, transition_order, (transition_enabled and not pending_filings))) - order += 1 + return tasks, order + + # Gets all completed restorations of the business (most recent first) + restorations = Filing.get_filings_by_types(business.id, ["restoration", "restorationApplication"]) + if not restorations: + # Should never get here + current_app.logger.error(f"Error - Business id: {business.id}. TRANSITION_NOT_FILED warning and no restoration filing on record.") + return tasks, order + # Use the first restoration in the list for the most recent completed restoration date + last_restoration_date = restorations[0].effective_date + last_ar_date = business.last_ar_date or business.founding_date + # Get the transition application todo order based on the ar tasks, last restoration date, and ar date information + transition_order, transition_enabled = _find_task_order_for_ta(tasks, order, last_restoration_date, last_ar_date) + # Bump all the task orders by one that are at and above transition_order + tasks = _bump_task_order(tasks, transition_order) + # Append the TA task at with order: transition_order. Disable if there are any incomplete filings + tasks.append(create_transition_todo(business, transition_order, (transition_enabled and not pending_filings))) + order += 1 - return tasks + return tasks, order def add_tr_tasks(business: Business, tasks: list, order: int, pending_tr_type: str | None = None): @@ -411,14 +455,14 @@ def create_incorporate_nr_todo(name_request, order, enabled): return todo -def create_conversion_filing_todo(business, order, enabled): - """Return a to-do JSON object.""" +def create_filing_todo(business, filing_type, order, enabled): + """Return a to-do JSON object for a filing that has not been started.""" todo = { "task": { "todo": { "business": business.json(), "header": { - "name": "conversion", + "name": filing_type, "status": "NEW" } } @@ -429,22 +473,19 @@ def create_conversion_filing_todo(business, order, enabled): return todo +def create_conversion_filing_todo(business, order, enabled): + """Return a to-do JSON object for a firm conversion filing.""" + return create_filing_todo(business, "conversion", order, enabled) + + +def create_review_imported_data_todo(business, order, enabled): + """Return a to-do JSON object for a corps review imported data filing.""" + return create_filing_todo(business, "reviewImportedData", order, enabled) + + def create_transition_todo(business, order, enabled): """Return a to-do JSON object for transition application filing.""" - todo = { - "task": { - "todo": { - "business": business.json(), - "header": { - "name": "transition", - "status": "NEW" - } - } - }, - "order": order, - "enabled": enabled - } - return todo + return create_filing_todo(business, "transition", order, enabled) def create_tr_todo(business: Business, order: int, enabled: bool, # noqa: PLR0913 diff --git a/legal-api/src/legal_api/services/filings/validations/common_validations.py b/legal-api/src/legal_api/services/filings/validations/common_validations.py index e9d1567e92..9595cf16fe 100644 --- a/legal-api/src/legal_api/services/filings/validations/common_validations.py +++ b/legal-api/src/legal_api/services/filings/validations/common_validations.py @@ -36,18 +36,9 @@ from legal_api.services import STAFF_ROLE, colin, doc_service, flags, namex from legal_api.services.permissions import ListActionsPermissionsAllowed, PermissionService from legal_api.services.request_context import get_request_context -from legal_api.services.utils import get_date, get_str +from legal_api.services.utils import NO_POSTAL_CODE_COUNTRY_CODES, get_date, get_str from legal_api.utils.auth import jwt -NO_POSTAL_CODE_COUNTRY_CODES = { - "AO", "AG", "AW", "BS", "BZ", "BJ", "BM", "BO", "BQ", "BW", "BF", "BI", - "CM", "CF", "TD", "KM", "CG", "CD", "CK", "CI", "CW", "DJ", "DM", "GQ", - "ER", "FJ", "TF", "GA", "GM", "GH", "GD", "GY", "HM", "HK", - "KI", "KP", "LY", "MO", "MW", "ML", "MR", "NR", - "AN", "NU", "QA", "RW", "KN", "ST", "SC", "SL", "SX", "SB", "SO", "SR", "SY", - "TL", "TG", "TK", "TO", "TT", "TV", "UG", "AE", "VU", "YE", "ZW" -} - # streetAddress, addressCity, and addressCountry no-leading/trailing-whitespace are enforced by # the schema (business-schemas address pattern). postalCode is not patterned (optional/nullable), # so it is still whitespace-validated here. diff --git a/legal-api/src/legal_api/services/utils.py b/legal-api/src/legal_api/services/utils.py index c944bb39bb..c496764273 100644 --- a/legal-api/src/legal_api/services/utils.py +++ b/legal-api/src/legal_api/services/utils.py @@ -15,6 +15,32 @@ from datetime import date import dpath +import pycountry + +# ISO-2 codes of countries that do not use postal codes; mirrors the frontend address rules. +NO_POSTAL_CODE_COUNTRY_CODES = { + "AO", "AG", "AW", "BS", "BZ", "BJ", "BM", "BO", "BQ", "BW", "BF", "BI", + "CM", "CF", "TD", "KM", "CG", "CD", "CK", "CI", "CW", "DJ", "DM", "GQ", + "ER", "FJ", "TF", "GA", "GM", "GH", "GD", "GY", "HM", "HK", + "KI", "KP", "LY", "MO", "MW", "ML", "MR", "NR", + "AN", "NU", "QA", "RW", "KN", "ST", "SC", "SL", "SX", "SB", "SO", "SR", "SY", + "TL", "TG", "TK", "TO", "TT", "TV", "UG", "AE", "VU", "YE", "ZW" +} + + +def is_postal_code_required(country: str) -> bool: + """Return whether an address in ``country`` must carry a postal code. + + Only countries that resolve to a code in NO_POSTAL_CODE_COUNTRY_CODES are exempt; a missing or + unrecognised country is treated as requiring one. + """ + if not country: + return True + try: + country_code = pycountry.countries.search_fuzzy(country)[0].alpha_2 + except LookupError: + return True + return country_code not in NO_POSTAL_CODE_COUNTRY_CODES def get_date(filing: dict, path: str) -> date: diff --git a/legal-api/src/legal_api/services/warnings/business/business_checks/__init__.py b/legal-api/src/legal_api/services/warnings/business/business_checks/__init__.py index d00e55b859..c64191140b 100644 --- a/legal-api/src/legal_api/services/warnings/business/business_checks/__init__.py +++ b/legal-api/src/legal_api/services/warnings/business/business_checks/__init__.py @@ -71,6 +71,47 @@ class BusinessWarningCodes(str, Enum): NO_START_DATE = "NO_START_DATE" + NO_REGISTERED_OFFICE = "NO_REGISTERED_OFFICE" + NO_REGISTERED_OFFICE_MAILING_ADDRESS = "NO_REGISTERED_OFFICE_MAILING_ADDRESS" + NO_REGISTERED_OFFICE_MAILING_ADDRESS_STREET = "NO_REGISTERED_OFFICE_MAILING_ADDRESS_STREET" + NO_REGISTERED_OFFICE_MAILING_ADDRESS_CITY = "NO_REGISTERED_OFFICE_MAILING_ADDRESS_CITY" + NO_REGISTERED_OFFICE_MAILING_ADDRESS_COUNTRY = "NO_REGISTERED_OFFICE_MAILING_ADDRESS_COUNTRY" + NO_REGISTERED_OFFICE_MAILING_ADDRESS_POSTAL_CODE = "NO_REGISTERED_OFFICE_MAILING_ADDRESS_POSTAL_CODE" + NO_REGISTERED_OFFICE_MAILING_ADDRESS_REGION = "NO_REGISTERED_OFFICE_MAILING_ADDRESS_REGION" + NO_REGISTERED_OFFICE_DELIVERY_ADDRESS = "NO_REGISTERED_OFFICE_DELIVERY_ADDRESS" + NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_STREET = "NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_STREET" + NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_CITY = "NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_CITY" + NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_COUNTRY = "NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_COUNTRY" + NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_POSTAL_CODE = "NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_POSTAL_CODE" + NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_REGION = "NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_REGION" + + NO_RECORDS_OFFICE = "NO_RECORDS_OFFICE" + NO_RECORDS_OFFICE_MAILING_ADDRESS = "NO_RECORDS_OFFICE_MAILING_ADDRESS" + NO_RECORDS_OFFICE_MAILING_ADDRESS_STREET = "NO_RECORDS_OFFICE_MAILING_ADDRESS_STREET" + NO_RECORDS_OFFICE_MAILING_ADDRESS_CITY = "NO_RECORDS_OFFICE_MAILING_ADDRESS_CITY" + NO_RECORDS_OFFICE_MAILING_ADDRESS_COUNTRY = "NO_RECORDS_OFFICE_MAILING_ADDRESS_COUNTRY" + NO_RECORDS_OFFICE_MAILING_ADDRESS_POSTAL_CODE = "NO_RECORDS_OFFICE_MAILING_ADDRESS_POSTAL_CODE" + NO_RECORDS_OFFICE_MAILING_ADDRESS_REGION = "NO_RECORDS_OFFICE_MAILING_ADDRESS_REGION" + NO_RECORDS_OFFICE_DELIVERY_ADDRESS = "NO_RECORDS_OFFICE_DELIVERY_ADDRESS" + NO_RECORDS_OFFICE_DELIVERY_ADDRESS_STREET = "NO_RECORDS_OFFICE_DELIVERY_ADDRESS_STREET" + NO_RECORDS_OFFICE_DELIVERY_ADDRESS_CITY = "NO_RECORDS_OFFICE_DELIVERY_ADDRESS_CITY" + NO_RECORDS_OFFICE_DELIVERY_ADDRESS_COUNTRY = "NO_RECORDS_OFFICE_DELIVERY_ADDRESS_COUNTRY" + NO_RECORDS_OFFICE_DELIVERY_ADDRESS_POSTAL_CODE = "NO_RECORDS_OFFICE_DELIVERY_ADDRESS_POSTAL_CODE" + NO_RECORDS_OFFICE_DELIVERY_ADDRESS_REGION = "NO_RECORDS_OFFICE_DELIVERY_ADDRESS_REGION" + + NO_DIRECTOR_MAILING_ADDRESS = "NO_DIRECTOR_MAILING_ADDRESS" + NO_DIRECTOR_MAILING_ADDRESS_STREET = "NO_DIRECTOR_MAILING_ADDRESS_STREET" + NO_DIRECTOR_MAILING_ADDRESS_CITY = "NO_DIRECTOR_MAILING_ADDRESS_CITY" + NO_DIRECTOR_MAILING_ADDRESS_COUNTRY = "NO_DIRECTOR_MAILING_ADDRESS_COUNTRY" + NO_DIRECTOR_MAILING_ADDRESS_POSTAL_CODE = "NO_DIRECTOR_MAILING_ADDRESS_POSTAL_CODE" + NO_DIRECTOR_MAILING_ADDRESS_REGION = "NO_DIRECTOR_MAILING_ADDRESS_REGION" + NO_DIRECTOR_DELIVERY_ADDRESS = "NO_DIRECTOR_DELIVERY_ADDRESS" + NO_DIRECTOR_DELIVERY_ADDRESS_STREET = "NO_DIRECTOR_DELIVERY_ADDRESS_STREET" + NO_DIRECTOR_DELIVERY_ADDRESS_CITY = "NO_DIRECTOR_DELIVERY_ADDRESS_CITY" + NO_DIRECTOR_DELIVERY_ADDRESS_COUNTRY = "NO_DIRECTOR_DELIVERY_ADDRESS_COUNTRY" + NO_DIRECTOR_DELIVERY_ADDRESS_POSTAL_CODE = "NO_DIRECTOR_DELIVERY_ADDRESS_POSTAL_CODE" + NO_DIRECTOR_DELIVERY_ADDRESS_REGION = "NO_DIRECTOR_DELIVERY_ADDRESS_REGION" + MULTIPLE_ANNUAL_REPORTS_NOT_FILED = "MULTIPLE_ANNUAL_REPORTS_NOT_FILED" TRANSITION_NOT_FILED = "TRANSITION_NOT_FILED" TRANSITION_NOT_FILED_AFTER_12_MONTH_RESTORATION = "TRANSITION_NOT_FILED_AFTER_12_MONTH_RESTORATION" @@ -86,6 +127,17 @@ class BusinessWarningReferers(str, Enum): BUSINESS_OFFICE = "BUSINESS_OFFICE" BUSINESS_PARTY = "BUSINESS_PARTY" COMPLETING_PARTY = "COMPLETING_PARTY" + REGISTERED_OFFICE = "REGISTERED_OFFICE" + RECORDS_OFFICE = "RECORDS_OFFICE" + DIRECTOR = "DIRECTOR" + + +# referers whose delivery address must carry a region +OFFICE_REFERERS = ( + BusinessWarningReferers.BUSINESS_OFFICE, + BusinessWarningReferers.REGISTERED_OFFICE, + BusinessWarningReferers.RECORDS_OFFICE +) class BusinessWarnings(str, Enum): @@ -102,6 +154,35 @@ class BusinessWarnings(str, Enum): WARNING_MESSAGE_BASE = {"warningType": WarningType.MISSING_REQUIRED_BUSINESS_INFO} +def _address_warnings(code_prefix: str, label: str) -> dict: + """Build the mailing and delivery address warnings for a referer (e.g. "REGISTERED_OFFICE", "Registered office").""" + addresses = {} + for address_type in ("mailing", "delivery"): + code_base = f"NO_{code_prefix}_{address_type.upper()}_ADDRESS" + addresses[address_type] = { + BusinessWarnings.NO_ADDRESS: { + **WARNING_MESSAGE_BASE, + "code": BusinessWarningCodes(code_base), + "message": f"{label} {address_type} address is required." + }, + **{ + BusinessWarnings(f"NO_ADDRESS_{field}"): { + **WARNING_MESSAGE_BASE, + "code": BusinessWarningCodes(f"{code_base}_{field}"), + "message": f"{field_label} is required for {label.lower()} {address_type} address." + } + for field, field_label in ( + ("STREET", "Street"), + ("CITY", "City"), + ("COUNTRY", "Country"), + ("POSTAL_CODE", "Postal code"), + ("REGION", "Region") + ) + } + } + return {"addresses": addresses} + + REFERER_WARNINGS_MAPPING = { BusinessWarningReferers.BUSINESS_OFFICE: { "addresses": { @@ -242,7 +323,10 @@ class BusinessWarnings(str, Enum): } } } - } + }, + BusinessWarningReferers.REGISTERED_OFFICE: _address_warnings("REGISTERED_OFFICE", "Registered office"), + BusinessWarningReferers.RECORDS_OFFICE: _address_warnings("RECORDS_OFFICE", "Records office"), + BusinessWarningReferers.DIRECTOR: _address_warnings("DIRECTOR", "Director") } diff --git a/legal-api/src/legal_api/services/warnings/business/business_checks/corps.py b/legal-api/src/legal_api/services/warnings/business/business_checks/corps.py index 857861028a..959f36b589 100644 --- a/legal-api/src/legal_api/services/warnings/business/business_checks/corps.py +++ b/legal-api/src/legal_api/services/warnings/business/business_checks/corps.py @@ -14,20 +14,91 @@ """Business checks for corps.""" -from business_model.models import Business -from legal_api.services.warnings.business.business_checks import BusinessWarningCodes, WarningType +from business_model.models import Address, Business, OfficeType, PartyRole +from legal_api.services.warnings.business.business_checks import ( + WARNING_MESSAGE_BASE, + BusinessWarningCodes, + BusinessWarningReferers, + WarningType, +) +from legal_api.services.warnings.business.business_checks.firms import check_address + +# office type -> (warning referer, missing office code, missing office message) +OFFICE_CHECKS = { + OfficeType.REGISTERED: (BusinessWarningReferers.REGISTERED_OFFICE, + BusinessWarningCodes.NO_REGISTERED_OFFICE, + "A registered office is required."), + OfficeType.RECORDS: (BusinessWarningReferers.RECORDS_OFFICE, + BusinessWarningCodes.NO_RECORDS_OFFICE, + "A records office is required.") +} def check_business(business: Business, is_staff: bool = False) -> list: """Check business data.""" result = [] + result.extend(check_offices(business)) + result.extend(check_directors(business)) result.extend(check_amalgamating_business(business, is_staff)) result.extend(check_transition_application(business)) return result +def _with_data(warnings: list, data: dict) -> list: + """Return copies of the warnings carrying ``data`` so the UI can identify the office/party to fix.""" + return [{**warning, "data": data} for warning in warnings] + + +def check_offices(business: Business) -> list: + """Check for missing registered office and records office address data.""" + result = [] + + for office_type, (referer, no_office_code, no_office_message) in OFFICE_CHECKS.items(): + data = {"officeType": office_type} + office = business.offices \ + .filter_by(office_type=office_type, deactivated_date=None) \ + .first() + + if not office: + result.append({ + **WARNING_MESSAGE_BASE, + "code": no_office_code, + "message": no_office_message, + "data": data + }) + continue + + addresses = office.addresses.all() + mailing_address = next((x for x in addresses if x.address_type == Address.MAILING), None) + result.extend(_with_data(check_address(mailing_address, Address.MAILING, referer), data)) + + delivery_address = next((x for x in addresses if x.address_type == Address.DELIVERY), None) + result.extend(_with_data(check_address(delivery_address, Address.DELIVERY, referer), data)) + + return result + + +def check_directors(business: Business) -> list: + """Check for missing address data on active directors.""" + result = [] + + director_roles = business.party_roles \ + .filter(PartyRole.role == PartyRole.RoleTypes.DIRECTOR.value) \ + .filter(PartyRole.cessation_date.is_(None)) + + for party_role in director_roles: + party = party_role.party + data = {"partyId": party.id} + result.extend(_with_data( + check_address(party.mailing_address, Address.MAILING, BusinessWarningReferers.DIRECTOR), data)) + result.extend(_with_data( + check_address(party.delivery_address, Address.DELIVERY, BusinessWarningReferers.DIRECTOR), data)) + + return result + + def check_transition_application(business: Business) -> list: """Check if a business is currently pending a post restoration transition application.""" result = [] @@ -55,7 +126,7 @@ def check_amalgamating_business(business: Business, is_staff: bool = False) -> l name_request = amalgamation_application.get("nameRequest", {}) # legalName is only present when the resulting business will be named resulting_business_name = name_request.get("legalName") - + warning = { "code": "AMALGAMATING_BUSINESS", "message": "This business is part of a future effective amalgamation.", diff --git a/legal-api/src/legal_api/services/warnings/business/business_checks/firms.py b/legal-api/src/legal_api/services/warnings/business/business_checks/firms.py index f11b6b6eca..73964550b0 100644 --- a/legal-api/src/legal_api/services/warnings/business/business_checks/firms.py +++ b/legal-api/src/legal_api/services/warnings/business/business_checks/firms.py @@ -16,8 +16,10 @@ from typing import Final from business_model.models import Address, Business, Filing, Office, Party, PartyRole +from legal_api.services.utils import is_postal_code_required from . import ( + OFFICE_REFERERS, WARNING_MESSAGE_BASE, BusinessWarningCodes, BusinessWarningReferers, @@ -254,12 +256,12 @@ def check_address(address: Address, result.append(get_address_business_warning(referer, address_type, BusinessWarnings.NO_ADDRESS_COUNTRY)) - if not address.postal_code: + if not address.postal_code and is_postal_code_required(address.country): result.append(get_address_business_warning(referer, address_type, BusinessWarnings.NO_ADDRESS_POSTAL_CODE)) - if (referer == BusinessWarningReferers.BUSINESS_OFFICE + if (referer in OFFICE_REFERERS and address_type == Address.DELIVERY and not address.region): result.append(get_address_business_warning(referer, diff --git a/legal-api/tests/unit/resources/v2/test_business.py b/legal-api/tests/unit/resources/v2/test_business.py index 7364cc5da9..051c18a8fa 100644 --- a/legal-api/tests/unit/resources/v2/test_business.py +++ b/legal-api/tests/unit/resources/v2/test_business.py @@ -579,12 +579,12 @@ def test_get_business_with_allowed_filings(session, client, jwt): ('NO_WARNINGS_EXIST_NO_MISSING_DATA', 'SP', 'FM0000003', False, False), ('NO_WARNINGS_EXIST_NO_MISSING_DATA', 'GP', 'FM0000004', False, False), ('NO_WARNINGS_NON_FIRM', 'CP', 'CP7654321', True, False), - ('NO_WARNINGS_NON_FIRM', 'BEN', 'CP7654321', True, False), - ('NO_WARNINGS_NON_FIRM', 'BC', 'BC7654321', True, False), + ('WARNINGS_EXIST_MISSING_DATA_CORPS', 'BEN', 'BC7654321', True, True), + ('WARNINGS_EXIST_MISSING_DATA_CORPS', 'BC', 'BC7654321', True, True), ]) def test_get_business_with_incomplete_info(session, client, jwt, test_name, legal_type, identifier, has_missing_business_info, missing_business_info_warning_expected): - """Assert that SP/GPs with missing business info is populating warnings list.""" + """Assert that SP/GPs and corps with missing business info are populating warnings list.""" if has_missing_business_info: business = factory_business(entity_type=legal_type, identifier=identifier) diff --git a/legal-api/tests/unit/resources/v2/test_business_tasks.py b/legal-api/tests/unit/resources/v2/test_business_tasks.py index b4e75c4f75..159701c4d6 100644 --- a/legal-api/tests/unit/resources/v2/test_business_tasks.py +++ b/legal-api/tests/unit/resources/v2/test_business_tasks.py @@ -29,7 +29,7 @@ from legal_api.services.authz import STAFF_ROLE from registry_schemas.example_data import ANNUAL_REPORT, FILING_HEADER, RESTORATION, TRANSITION_FILING_TEMPLATE from tests import integration_payment -from tests.unit.models import factory_business, factory_business_mailing_address, factory_completed_filing, factory_filing, factory_pending_filing +from tests.unit.models import factory_business, factory_business_mailing_address, factory_business_office, factory_completed_filing, factory_filing, factory_pending_filing from tests.unit.services.utils import create_header from tests.unit.services.warnings import create_business @@ -100,12 +100,14 @@ def test_get_tasks_no_filings(session, client, jwt): def test_get_tasks_sentinel_founding_date_skips_ar(session, client, jwt): """Assert COLIN year-1 sentinel founding date does not generate thousands of AR todos.""" identifier = 'BC1234567' - factory_business( + business = factory_business( identifier, founding_date=datetime(1, 1, 1, 8, 0, 0, tzinfo=UTC), entity_type=Business.LegalTypes.COMP.value, state=Business.State.ACTIVE, ) + factory_business_office(business, 'registeredOffice') + factory_business_office(business, 'recordsOffice') rv = client.get(f'/api/v2/businesses/{identifier}/tasks', headers=create_header(jwt, [STAFF_ROLE], identifier)) assert rv.status_code == HTTPStatus.OK @@ -164,7 +166,9 @@ def test_get_tasks_next_year(session, client, jwt): def test_bcorps_get_tasks_no_filings(session, client, jwt): """Assert that to-do for the current year is returned when there are no filings.""" identifier = 'CP7654321' - factory_business(identifier, datetime.now(), None, Business.LegalTypes.BCOMP.value) + business = factory_business(identifier, datetime.now(), None, Business.LegalTypes.BCOMP.value) + factory_business_office(business, 'registeredOffice') + factory_business_office(business, 'recordsOffice') rv = client.get(f'/api/v2/businesses/{identifier}/tasks', headers=create_header(jwt, [STAFF_ROLE], identifier)) @@ -412,6 +416,8 @@ def test_construct_task_list_ta(app, session, client, jwt, test_name, identifier with patch('legal_api.services.warnings.business.business_checks.business.involuntary_dissolution_check', return_value=[]): business = factory_business(identifier, founding_date, last_ar_date, legal_type) + factory_business_office(business, 'registeredOffice') + factory_business_office(business, 'recordsOffice') if restored_date: filing = {'filing': {'header': {'name': 'restoration', 'date': restored_date.isoformat(), 'certifiedBy': 'test'}, 'restoration': {'type': 'fullRestoration'}}} @@ -585,6 +591,57 @@ def test_conversion_filing_task(session, client, jwt, test_name, legal_type, ide for x in rv_json['tasks']) assert not conversion_to_do +@pytest.mark.parametrize( + 'test_name, legal_type, identifier, has_missing_business_info, has_draft, has_other_draft, admin_freeze, ' + 'todo_expected, enabled_expected', [ + ('TODO_MISSING_DATA', 'BC', 'BC7654321', True, False, False, False, True, True), + ('TODO_MISSING_DATA', 'BEN', 'BC7654322', True, False, False, False, True, True), + ('TODO_MISSING_DATA', 'ULC', 'BC7654323', True, False, False, False, True, True), + ('TODO_MISSING_DATA', 'CC', 'BC7654324', True, False, False, False, True, True), + ('TODO_MISSING_DATA', 'C', 'C7654325', True, False, False, False, True, True), + ('TODO_DISABLED_OTHER_DRAFT', 'BC', 'BC7654328', True, False, True, False, True, False), + ('TODO_DISABLED_ADMIN_FREEZE', 'BC', 'BC7654329', True, False, False, True, True, False), + ('NO_TODO_DRAFT_EXISTS', 'BC', 'BC7654326', True, True, False, False, False, None), + ('NO_TODO_NO_MISSING_DATA', 'BC', 'BC7654327', False, False, False, False, False, None), + ('NO_TODO_NON_CORP', 'CP', 'CP7654321', True, False, False, False, False, None), + ('NO_TODO_NON_CORP', 'SP', 'FM0000001', True, False, False, False, False, None), + ]) +def test_review_imported_data_task(session, client, jwt, test_name, legal_type, identifier, has_missing_business_info, + has_draft, has_other_draft, admin_freeze, todo_expected, enabled_expected): + """Assert the reviewImportedData todo shows up only for corps with missing business info and no draft. + + The todo is disabled when the business has a default blocker (pending filing, admin freeze) since the API + would reject the filing. + """ + business = factory_business(entity_type=legal_type, identifier=identifier, admin_freeze=admin_freeze) + if not has_missing_business_info: + factory_business_office(business, 'registeredOffice') + factory_business_office(business, 'recordsOffice') + if has_draft: + factory_filing(business, {'filing': {'header': {'name': 'reviewImportedData'}}}, + filing_type='reviewImportedData') + if has_other_draft: + factory_filing(business, {'filing': {'header': {'name': 'changeOfAddress'}}}, filing_type='changeOfAddress') + + rv = client.get(f'/api/v2/businesses/{identifier}/tasks', headers=create_header(jwt, [STAFF_ROLE], identifier)) + + assert rv.status_code == HTTPStatus.OK + todos = [x for x in rv.json['tasks'] + if x['task'].get('todo', {}).get('header', {}).get('name') == 'reviewImportedData'] + + if todo_expected: + assert len(todos) == 1 + assert todos[0]['task']['todo']['header']['status'] == 'NEW' + assert todos[0]['enabled'] is enabled_expected + assert todos[0]['order'] == 1 + assert any(x['warningType'] == 'MISSING_REQUIRED_BUSINESS_INFO' + for x in todos[0]['task']['todo']['business']['warnings']) + else: + assert not todos + assert not any(x['task'].get('todo', {}).get('header', {}).get('name') == 'conversion' + for x in rv.json['tasks'] if legal_type != 'SP') + + @pytest.mark.parametrize('test_name, business_restored, transition_completed, transition_task_expected', [ ('Transition application needed but not completed', True, False, True), ('Transition application needed and completed', True, True, False), diff --git a/legal-api/tests/unit/services/test_authorization.py b/legal-api/tests/unit/services/test_authorization.py index 84dd01688b..9a8a74b20a 100644 --- a/legal-api/tests/unit/services/test_authorization.py +++ b/legal-api/tests/unit/services/test_authorization.py @@ -2424,6 +2424,23 @@ def test_allowed_filings_warnings(monkeypatch, app, session, jwt, test_name, sta assert filing_types == expected +@pytest.mark.parametrize('username, roles', [('staff', [STAFF_ROLE]), ('general', [BASIC_USER])]) +def test_allowed_filings_corps_missing_info_warnings(monkeypatch, app, session, jwt, username, roles): + """Assert corps with MISSING_REQUIRED_BUSINESS_INFO warnings can still file reviewImportedData but not correction.""" + with jwt_request_context(app, jwt, roles=roles, username=username): + monkeypatch.setattr('legal_api.services.flags.value', + lambda flag, _user, _account_id: '' if flag == 'enabled-specific-filings' else {}) + monkeypatch.setattr('business_model.models.User.get_or_create_user_by_jwt', lambda _: None) + for legal_type in ['BC', 'BEN', 'CC', 'ULC', 'C', 'CBEN', 'CCC', 'CUL']: + business = create_business(legal_type, Business.State.ACTIVE) + business.warnings = [{**MISSING_BUSINESS_INFO_WARNINGS[0], 'code': 'NO_REGISTERED_OFFICE', + 'data': {'officeType': 'registeredOffice'}}] + filing_names = [x['name'] for x in get_allowed_filings(business, Business.State.ACTIVE, legal_type, jwt)] + assert 'reviewImportedData' in filing_names + assert 'correction' not in filing_names + assert 'dissolution' not in filing_names + + @pytest.mark.parametrize( 'test_name,state,legal_types,username,roles,state_filing_types,state_filing_sub_types,expected', [ diff --git a/legal-api/tests/unit/services/warnings/business/business_checks/test_corps.py b/legal-api/tests/unit/services/warnings/business/business_checks/test_corps.py index 136d750853..15dd0299db 100644 --- a/legal-api/tests/unit/services/warnings/business/business_checks/test_corps.py +++ b/legal-api/tests/unit/services/warnings/business/business_checks/test_corps.py @@ -15,9 +15,42 @@ import pytest from unittest.mock import Mock, patch from datetime import datetime, timedelta -from tests.unit.models import factory_business -from legal_api.services.warnings.business.business_checks import WarningType -from legal_api.services.warnings.business.business_checks.corps import check_business +from business_model.models import Address, Office, OfficeType, PartyRole +from tests.unit.models import factory_business, factory_business_office, factory_party_role +from tests.unit.services.warnings import factory_address +from legal_api.services.warnings.business.business_checks import BusinessWarningReferers, WarningType +from legal_api.services.warnings.business.business_checks.corps import check_business, check_directors, check_offices +from legal_api.services.warnings.business.business_checks.firms import check_address + + +def create_complete_business(identifier="BC1234567"): + """Create a corp with a complete registered office and records office.""" + business = factory_business(identifier=identifier, entity_type="BC") + factory_business_office(business, OfficeType.REGISTERED) + factory_business_office(business, OfficeType.RECORDS) + return business + + +def add_office(business, office_type, mailing_address=None, delivery_address=None): + """Attach an office with the given addresses (None = no address row) to the business.""" + office = Office(office_type=office_type) + for address in (mailing_address, delivery_address): + if address: + office.addresses.append(address) + business.offices.append(office) + business.save() + return office + + +def add_director(business, mailing_address, delivery_address, cessation_date=None): + """Attach a director with the given addresses to the business.""" + officer = {'firstName': 'Jane', 'lastName': 'Doe', 'middleInitial': '', 'partyType': 'person', + 'organizationName': ''} + party_role = factory_party_role(delivery_address, mailing_address, officer, datetime(2020, 1, 1), cessation_date, + PartyRole.RoleTypes.DIRECTOR) + business.party_roles.append(party_role) + business.save() + return party_role @pytest.mark.parametrize("has_amalgamation, expected_warning", [ (True, True), # Future effective amalgamation @@ -25,7 +58,7 @@ ]) def test_check_business(session, has_amalgamation, expected_warning): """Test the check_business function.""" - business = factory_business(identifier="BC1234567") + business = create_complete_business() mock_filing = Mock() mock_filing.id = "filing-id" @@ -82,7 +115,7 @@ def test_check_business(session, has_amalgamation, expected_warning): ]) def test_check_business_amalgamation_resulting_name(session, name_request, expected_resulting_name): """Test that resultingBusinessName reflects whether the amalgamation result is named or numbered.""" - business = factory_business(identifier="BC1234567") + business = create_complete_business() mock_filing = Mock() mock_filing.id = "filing-id" @@ -114,7 +147,7 @@ def test_check_business_amalgamation_resulting_name(session, name_request, expec def test_check_business_excludes_filing_id_for_non_staff(session): """Test that filing ids and the resulting business identifier are only included for staff users.""" - business = factory_business(identifier="BC1234567") + business = create_complete_business() mock_filing = Mock( id="filing-id", effective_date=datetime.now() + timedelta(days=1), @@ -146,3 +179,191 @@ def test_check_business_excludes_filing_id_for_non_staff(session): } assert 'filingId' not in warning['data'] assert 'resultingBusinessIdentifier' not in warning['data'] + + +def test_check_business_complete_corp_has_no_missing_info_warnings(session): + """Assert a corp with complete offices and directors yields no MISSING_REQUIRED_BUSINESS_INFO warnings.""" + business = create_complete_business() + add_director(business, factory_address('mailing'), factory_address('delivery')) + + with patch('business_model.models.business.Business.is_pending_amalgamating_business', return_value=None): + result = check_business(business) + + assert result == [] + + +def test_check_offices_missing_offices(session): + """Assert a corp without registered/records offices gets one warning per office, identified by officeType.""" + business = factory_business(identifier="BC1234567", entity_type="BC") + + result = check_offices(business) + + assert [(x['code'], x['message'], x['data']) for x in result] == [ + ('NO_REGISTERED_OFFICE', 'A registered office is required.', {'officeType': 'registeredOffice'}), + ('NO_RECORDS_OFFICE', 'A records office is required.', {'officeType': 'recordsOffice'}), + ] + assert all(x['warningType'] == WarningType.MISSING_REQUIRED_BUSINESS_INFO for x in result) + + +def test_check_offices_ignores_deactivated_office(session): + """Assert a deactivated office does not satisfy the office check.""" + business = factory_business(identifier="BC1234567", entity_type="BC") + office = add_office(business, OfficeType.REGISTERED, factory_address('mailing'), factory_address('delivery')) + office.deactivated_date = datetime.now() + factory_business_office(business, OfficeType.RECORDS) + + result = check_offices(business) + + assert [x['code'] for x in result] == ['NO_REGISTERED_OFFICE'] + + +@pytest.mark.parametrize( + 'office_type, address_type, null_field, expected_code, expected_msg', + [ + (OfficeType.REGISTERED, 'mailing', 'NO_ADDRESS', 'NO_REGISTERED_OFFICE_MAILING_ADDRESS', + 'Registered office mailing address is required.'), + (OfficeType.REGISTERED, 'mailing', 'street', 'NO_REGISTERED_OFFICE_MAILING_ADDRESS_STREET', + 'Street is required for registered office mailing address.'), + (OfficeType.REGISTERED, 'mailing', 'city', 'NO_REGISTERED_OFFICE_MAILING_ADDRESS_CITY', + 'City is required for registered office mailing address.'), + (OfficeType.REGISTERED, 'mailing', 'country', 'NO_REGISTERED_OFFICE_MAILING_ADDRESS_COUNTRY', + 'Country is required for registered office mailing address.'), + (OfficeType.REGISTERED, 'mailing', 'postal_code', 'NO_REGISTERED_OFFICE_MAILING_ADDRESS_POSTAL_CODE', + 'Postal code is required for registered office mailing address.'), + (OfficeType.REGISTERED, 'mailing', 'region', None, None), + (OfficeType.REGISTERED, 'delivery', 'NO_ADDRESS', 'NO_REGISTERED_OFFICE_DELIVERY_ADDRESS', + 'Registered office delivery address is required.'), + (OfficeType.REGISTERED, 'delivery', 'street', 'NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_STREET', + 'Street is required for registered office delivery address.'), + (OfficeType.REGISTERED, 'delivery', 'region', 'NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_REGION', + 'Region is required for registered office delivery address.'), + (OfficeType.RECORDS, 'mailing', 'NO_ADDRESS', 'NO_RECORDS_OFFICE_MAILING_ADDRESS', + 'Records office mailing address is required.'), + (OfficeType.RECORDS, 'mailing', 'postal_code', 'NO_RECORDS_OFFICE_MAILING_ADDRESS_POSTAL_CODE', + 'Postal code is required for records office mailing address.'), + (OfficeType.RECORDS, 'delivery', 'NO_ADDRESS', 'NO_RECORDS_OFFICE_DELIVERY_ADDRESS', + 'Records office delivery address is required.'), + (OfficeType.RECORDS, 'delivery', 'city', 'NO_RECORDS_OFFICE_DELIVERY_ADDRESS_CITY', + 'City is required for records office delivery address.'), + (OfficeType.RECORDS, 'delivery', 'country', 'NO_RECORDS_OFFICE_DELIVERY_ADDRESS_COUNTRY', + 'Country is required for records office delivery address.'), + (OfficeType.RECORDS, 'delivery', 'region', 'NO_RECORDS_OFFICE_DELIVERY_ADDRESS_REGION', + 'Region is required for records office delivery address.'), + ]) +def test_check_offices_address_fields(session, office_type, address_type, null_field, expected_code, expected_msg): + """Assert each missing office address field raises the matching warning carrying the officeType.""" + business = factory_business(identifier="BC1234567", entity_type="BC") + other_type = OfficeType.RECORDS if office_type == OfficeType.REGISTERED else OfficeType.REGISTERED + factory_business_office(business, other_type) + + addresses = {'mailing': factory_address('mailing'), 'delivery': factory_address('delivery')} + if null_field == 'NO_ADDRESS': + addresses[address_type] = None + else: + addresses[address_type] = factory_address(address_type, make_null_field_name=null_field) + add_office(business, office_type, addresses['mailing'], addresses['delivery']) + + result = check_offices(business) + + if expected_code: + assert len(result) == 1 + warning = result[0] + assert warning['code'] == expected_code + assert warning['message'] == expected_msg + assert warning['warningType'] == WarningType.MISSING_REQUIRED_BUSINESS_INFO + assert warning['data'] == {'officeType': office_type} + else: + assert result == [] + + +@pytest.mark.parametrize( + 'address_type, null_field, expected_code, expected_msg', + [ + ('mailing', None, None, None), + ('mailing', 'region', None, None), + ('mailing', 'NO_ADDRESS', 'NO_DIRECTOR_MAILING_ADDRESS', 'Director mailing address is required.'), + ('mailing', 'street', 'NO_DIRECTOR_MAILING_ADDRESS_STREET', + 'Street is required for director mailing address.'), + ('mailing', 'city', 'NO_DIRECTOR_MAILING_ADDRESS_CITY', 'City is required for director mailing address.'), + ('mailing', 'country', 'NO_DIRECTOR_MAILING_ADDRESS_COUNTRY', + 'Country is required for director mailing address.'), + ('mailing', 'postal_code', 'NO_DIRECTOR_MAILING_ADDRESS_POSTAL_CODE', + 'Postal code is required for director mailing address.'), + ('delivery', 'region', None, None), + ('delivery', 'NO_ADDRESS', 'NO_DIRECTOR_DELIVERY_ADDRESS', 'Director delivery address is required.'), + ('delivery', 'street', 'NO_DIRECTOR_DELIVERY_ADDRESS_STREET', + 'Street is required for director delivery address.'), + ('delivery', 'city', 'NO_DIRECTOR_DELIVERY_ADDRESS_CITY', 'City is required for director delivery address.'), + ('delivery', 'country', 'NO_DIRECTOR_DELIVERY_ADDRESS_COUNTRY', + 'Country is required for director delivery address.'), + ('delivery', 'postal_code', 'NO_DIRECTOR_DELIVERY_ADDRESS_POSTAL_CODE', + 'Postal code is required for director delivery address.'), + ]) +def test_check_directors_address_fields(session, address_type, null_field, expected_code, expected_msg): + """Assert each missing director address field raises the matching warning carrying the partyId.""" + business = factory_business(identifier="BC1234567", entity_type="BC") + addresses = {'mailing': factory_address('mailing'), 'delivery': factory_address('delivery')} + if null_field == 'NO_ADDRESS': + addresses[address_type] = None + elif null_field: + addresses[address_type] = factory_address(address_type, make_null_field_name=null_field) + party_role = add_director(business, addresses['mailing'], addresses['delivery']) + + result = check_directors(business) + + if expected_code: + assert len(result) == 1 + warning = result[0] + assert warning['code'] == expected_code + assert warning['message'] == expected_msg + assert warning['warningType'] == WarningType.MISSING_REQUIRED_BUSINESS_INFO + assert warning['data'] == {'partyId': party_role.party.id} + else: + assert result == [] + + +@pytest.mark.parametrize('country, expected_codes', [ + ('CA', ['NO_DIRECTOR_MAILING_ADDRESS_POSTAL_CODE', 'NO_DIRECTOR_DELIVERY_ADDRESS_POSTAL_CODE']), + ('HK', []), +]) +def test_check_directors_postal_code_required_by_country(session, country, expected_codes): + """Assert a director without a postal code is only flagged when their country uses postal codes.""" + business = factory_business(identifier="BC1234567", entity_type="BC") + add_director(business, + factory_address('mailing', country=country, postal_code=None), + factory_address('delivery', country=country, postal_code=None)) + + result = check_directors(business) + + assert [x['code'] for x in result] == expected_codes + + +def test_check_directors_skips_ceased_directors_and_flags_each_active_one(session): + """Assert ceased directors are ignored and each active director is reported with its own partyId.""" + business = factory_business(identifier="BC1234567", entity_type="BC") + add_director(business, None, factory_address('delivery'), cessation_date=datetime(2021, 1, 1)) + active_one = add_director(business, None, factory_address('delivery')) + active_two = add_director(business, factory_address('mailing', make_null_field_name='city'), + factory_address('delivery')) + + result = check_directors(business) + + assert [(x['code'], x['data']) for x in result] == [ + ('NO_DIRECTOR_MAILING_ADDRESS', {'partyId': active_one.party.id}), + ('NO_DIRECTOR_MAILING_ADDRESS_CITY', {'partyId': active_two.party.id}), + ] + + +def test_warnings_carry_their_own_data(session): + """Assert adding data does not mutate the shared warning mapping (two offices, same code, different data).""" + business = factory_business(identifier="BC1234567", entity_type="BC") + add_office(business, OfficeType.REGISTERED, factory_address('mailing', make_null_field_name='street'), + factory_address('delivery')) + add_office(business, OfficeType.RECORDS, factory_address('mailing', make_null_field_name='street'), + factory_address('delivery')) + + result = check_offices(business) + + assert [x['data'] for x in result] == [{'officeType': 'registeredOffice'}, {'officeType': 'recordsOffice'}] + assert 'data' not in check_address(factory_address('mailing', make_null_field_name='street'), Address.MAILING, + BusinessWarningReferers.REGISTERED_OFFICE)[0] diff --git a/legal-api/tests/unit/services/warnings/business/business_checks/test_firms.py b/legal-api/tests/unit/services/warnings/business/business_checks/test_firms.py index c4ea91366b..25f79b7cdb 100644 --- a/legal-api/tests/unit/services/warnings/business/business_checks/test_firms.py +++ b/legal-api/tests/unit/services/warnings/business/business_checks/test_firms.py @@ -82,6 +82,18 @@ 'NO_COMPLETING_PARTY_MAILING_ADDRESS_POSTAL_CODE', 'Postal code is required for completing party mailing address.'), ('SUCCESS', 'mailing', 'region', BusinessWarningReferers.COMPLETING_PARTY, None, None), + + # corps referers: region required only on office delivery addresses + ('FAIL_NO_REGION', 'delivery', 'region', BusinessWarningReferers.REGISTERED_OFFICE, + 'NO_REGISTERED_OFFICE_DELIVERY_ADDRESS_REGION', 'Region is required for registered office delivery address.'), + ('SUCCESS', 'mailing', 'region', BusinessWarningReferers.REGISTERED_OFFICE, None, None), + ('FAIL_NO_REGION', 'delivery', 'region', BusinessWarningReferers.RECORDS_OFFICE, + 'NO_RECORDS_OFFICE_DELIVERY_ADDRESS_REGION', 'Region is required for records office delivery address.'), + ('SUCCESS', 'mailing', 'region', BusinessWarningReferers.RECORDS_OFFICE, None, None), + ('SUCCESS', 'delivery', 'region', BusinessWarningReferers.DIRECTOR, None, None), + ('SUCCESS', 'mailing', 'region', BusinessWarningReferers.DIRECTOR, None, None), + ('FAIL_NO_STREET', 'delivery', 'street', BusinessWarningReferers.DIRECTOR, + 'NO_DIRECTOR_DELIVERY_ADDRESS_STREET', 'Street is required for director delivery address.'), ]) def test_check_address(session, test_name, address_type, null_addr_field_name, referer, expected_code, expected_msg): """Assert that business address checks functions properly.""" @@ -104,6 +116,27 @@ def test_check_address(session, test_name, address_type, null_addr_field_name, r assert len(result) == 0 +@pytest.mark.parametrize( + 'country, expected_codes', + [ + ('CA', ['NO_BUSINESS_PARTY_MAILING_ADDRESS_POSTAL_CODE']), + ('Canada', ['NO_BUSINESS_PARTY_MAILING_ADDRESS_POSTAL_CODE']), + ('US', ['NO_BUSINESS_PARTY_MAILING_ADDRESS_POSTAL_CODE']), + ('HK', []), + ('Hong Kong', []), + ('AE', []), + ('Not A Country', ['NO_BUSINESS_PARTY_MAILING_ADDRESS_POSTAL_CODE']), + (None, ['NO_BUSINESS_PARTY_MAILING_ADDRESS_COUNTRY', 'NO_BUSINESS_PARTY_MAILING_ADDRESS_POSTAL_CODE']), + ]) +def test_check_address_postal_code_required_by_country(session, country, expected_codes): + """Assert a missing postal code is only flagged for countries that use postal codes (mirrors filing validation).""" + address = factory_address(address_type='mailing', country=country, postal_code=None) + + result = check_address(address, 'mailing', BusinessWarningReferers.BUSINESS_PARTY) + + assert [x['code'] for x in result] == expected_codes + + @pytest.mark.parametrize( 'test_name, legal_type, role, party_type, expected_code, expected_msg', [ diff --git a/legal-api/tests/unit/services/warnings/business/business_checks/test_in_liquidation.py b/legal-api/tests/unit/services/warnings/business/business_checks/test_in_liquidation.py index 25cf3c4c9d..c4f6e4929a 100644 --- a/legal-api/tests/unit/services/warnings/business/business_checks/test_in_liquidation.py +++ b/legal-api/tests/unit/services/warnings/business/business_checks/test_in_liquidation.py @@ -21,7 +21,7 @@ from business_model.models import Business from legal_api.services.warnings.business import check_business from legal_api.services.warnings.business.business_checks import WarningType, BusinessWarningCodes -from tests.unit.models import factory_business +from tests.unit.models import factory_business, factory_business_office FOUNDING_DATE = datetime(2023, 3, 3) IN_LIQUIDATION_DATE = datetime(2024, 6, 10, 0, 0, tzinfo=FixedOffsetTimezone(0)) @@ -56,6 +56,8 @@ def test_check_business(session, test_name, founding_date, in_liquidation_date, founding_date=founding_date, in_liquidation_date=in_liquidation_date, last_lr_year=last_lr_year) + factory_business_office(business, 'registeredOffice') + factory_business_office(business, 'recordsOffice') if test_name != 'NOT_IN_LIQUIDATION': business.in_liquidation = True