From 34e1b4a6792fe2e43acdf3164e7e81d776d8601f Mon Sep 17 00:00:00 2001 From: Kial Jinnah Date: Wed, 19 Aug 2026 14:27:41 -0400 Subject: [PATCH 1/4] 34088 Colin API - business snapshot endpoint Signed-off-by: Kial Jinnah --- colin-api/src/colin_api/models/__init__.py | 2 + .../src/colin_api/models/business_snapshot.py | 199 ++++++++++++++++++ colin-api/src/colin_api/models/shares.py | 3 + colin-api/src/colin_api/resources/business.py | 27 ++- colin-api/src/colin_api/version.py | 2 +- colin-api/tests/unit/__init__.py | 120 +++++++++++ .../tests/unit/api/test_business_auth_info.py | 77 ++----- .../tests/unit/api/test_business_snapshot.py | 184 ++++++++++++++++ colin-api/tests/unit/conftest.py | 46 ++++ colin-api/tests/unit/models/__init__.py | 1 + colin-api/tests/unit/models/test_shares.py | 40 ++++ 11 files changed, 634 insertions(+), 67 deletions(-) create mode 100644 colin-api/src/colin_api/models/business_snapshot.py create mode 100644 colin-api/tests/unit/api/test_business_snapshot.py create mode 100644 colin-api/tests/unit/models/__init__.py create mode 100644 colin-api/tests/unit/models/test_shares.py diff --git a/colin-api/src/colin_api/models/__init__.py b/colin-api/src/colin_api/models/__init__.py index e2ec54740f..07fcac795b 100644 --- a/colin-api/src/colin_api/models/__init__.py +++ b/colin-api/src/colin_api/models/__init__.py @@ -1,6 +1,8 @@ """Model imports.""" + from .address import Address from .business import Business +from .business_snapshot import BusinessSnapshot from .cont_out import ContOut from .corp_involved import CorpInvolved from .corp_name import CorpName diff --git a/colin-api/src/colin_api/models/business_snapshot.py b/colin-api/src/colin_api/models/business_snapshot.py new file mode 100644 index 0000000000..42f3d4d08d --- /dev/null +++ b/colin-api/src/colin_api/models/business_snapshot.py @@ -0,0 +1,199 @@ +# Copyright © 2026 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. +"""Snapshot of a COLIN business, normalized to LEAR structure.""" +from __future__ import annotations + +from datetime import datetime +from typing import Dict, Optional + +import pycountry +from flask import current_app + +from colin_api.exceptions import PartiesNotFoundException +from colin_api.models.business import Business +from colin_api.models.corp_party import Party +from colin_api.models.office import Office +from colin_api.models.shares import ShareObject +from colin_api.resources.db import DB + + +class BusinessSnapshot: # pylint: disable=too-few-public-methods + """Builds the LEAR-structured snapshot dict for a COLIN business.""" + + # snapshot offices are limited to the two the amalgamation flow prepopulates + OFFICE_TYPES = ('registeredOffice', 'recordsOffice') + + @classmethod + def get_snapshot(cls, orig_identifier: str) -> Dict: + """Return the business/parties/offices/shareClasses/resolutions snapshot.""" + identifier = orig_identifier + if identifier.startswith('BC'): + identifier = identifier[2:] + + con = DB.connection + business = Business.find_by_identifier(identifier, con=con) + cursor = con.cursor() + + try: + parties = Party.get_current(cursor, identifier) + except PartiesNotFoundException: + # no current directors on file is bad data but not an error state here + parties = [] + + offices = Office.convert_obj_list(Office.get_current(cursor, identifier)) or {} + + # mirror the /sharestructure resource: current structure is the one with no end event + share_structs = ShareObject.get_all(cursor, identifier) or [] + share_struct = next((x for x in share_structs if not x.end_event_id), None) + share_classes = share_struct.to_dict()['shareClasses'] if share_struct else [] + + resolutions = Business.get_resolutions(cursor, identifier) + + return { + 'business': cls._business_dict(business, orig_identifier, cursor), + 'parties': [cls._normalize_party(party) for party in parties], + 'offices': { + office_type: cls._normalize_office(office) + for office_type, office in offices.items() if office_type in cls.OFFICE_TYPES + }, + 'shareClasses': [cls._normalize_share_class(share_class) for share_class in share_classes], + 'resolutions': [{'date': date} for date in resolutions], + } + + @classmethod + def _business_dict(cls, business: Business, orig_identifier: str, cursor) -> Dict: + """Return the business section in the shape of LEAR's slim business json.""" + return { + 'identifier': orig_identifier, + 'legalName': business.corp_name, + 'legalType': business.corp_type, + 'state': business.lear_state, + # tri-state: None when COLIN can't compute good standing for the corp + 'goodStanding': business.good_standing, + # find_by_identifier returns the COLIN 'True'/'False' string + 'adminFreeze': business.admin_freeze == 'True', + 'foundingDate': cls._to_iso_datetime(business.founding_date), + 'taxId': business.business_number, + 'hasFutureEffectiveFiling': cls._has_future_effective_filing(cursor, business.corp_num), + } + + @staticmethod + def _has_future_effective_filing(cursor, corp_num: str) -> bool: + """Return whether any filing has an effective date still in the future.""" + current_date = datetime.utcnow().strftime('%Y-%m-%d') + cursor.execute( + """ + select count(*) + from event join filing on event.event_id = filing.event_id + where event.corp_num=:corp_num + and filing.effective_dt > TO_DATE(:current_date, 'YYYY-mm-dd') + """, + corp_num=corp_num, + current_date=current_date + ) + return cursor.fetchone()[0] > 0 + + @classmethod + def _normalize_party(cls, party: Party) -> Dict: + """Return a party in the structure of LEAR's /parties items.""" + raw = party.as_dict() + return { + 'officer': {**raw['officer'], 'id': raw['id'], 'email': None}, + 'deliveryAddress': cls._normalize_address(raw['deliveryAddress']), + 'mailingAddress': cls._normalize_address(raw['mailingAddress']), + 'roles': raw['roles'] or [], + } + + @classmethod + def _normalize_office(cls, office: Dict) -> Dict: + """Return an office's addresses in LEAR structure.""" + return { + 'deliveryAddress': cls._normalize_address(office.get('deliveryAddress')), + 'mailingAddress': cls._normalize_address(office.get('mailingAddress')), + } + + @classmethod + def _normalize_address(cls, address: Optional[Dict]) -> Optional[Dict]: + """Return an Address.as_dict in the structure of LEAR's address json.""" + if not address: + return None + return { + 'id': address.get('addressId'), + 'streetAddress': address.get('streetAddress'), + 'streetAddressAdditional': address.get('streetAddressAdditional'), + 'addressCity': address.get('addressCity'), + 'addressRegion': address.get('addressRegion'), + 'addressCountry': cls._country_to_alpha2(address.get('addressCountry')), + 'postalCode': address.get('postalCode'), + 'deliveryInstructions': address.get('deliveryInstructions'), + } + + @classmethod + def _normalize_share_class(cls, share_class: Dict) -> Dict: + """Return a share class in the structure of LEAR's /share-classes items.""" + return { + 'id': share_class['id'], + 'name': share_class['name'], + # COLIN never stores a priority - the class id preserves creation order + 'priority': share_class['displayOrder'], + 'hasMaximumShares': share_class['hasMaximumShares'], + 'maxNumberOfShares': cls._to_int(share_class['maxNumberOfShares']), + 'hasParValue': share_class['hasParValue'], + 'parValue': float(share_class['parValue']) if share_class['parValue'] is not None else None, + 'currency': share_class['currency'], + 'currencyAdditional': share_class['currencyAdditional'], + 'hasRightsOrRestrictions': share_class['hasRightsOrRestrictions'], + 'series': [cls._normalize_share_series(series) for series in share_class['series']], + } + + @classmethod + def _normalize_share_series(cls, series: Dict) -> Dict: + """Return a share series in the structure of LEAR's series json.""" + return { + 'id': series['id'], + 'name': series['name'], + 'priority': series['displayOrder'], + 'hasMaximumShares': series['hasMaximumShares'], + 'maxNumberOfShares': cls._to_int(series['maxNumberOfShares']), + 'hasRightsOrRestrictions': series['hasRightsOrRestrictions'], + } + + _country_cache: Dict[str, str] = {} + + @classmethod + def _country_to_alpha2(cls, country: Optional[str]) -> Optional[str]: + """Map COLIN's country description to the alpha-2 code LEAR stores.""" + if not country: + return country + if len(country) == 2: # already a code + return country + if country not in cls._country_cache: + try: + cls._country_cache[country] = pycountry.countries.search_fuzzy(country)[0].alpha_2 + except LookupError: + current_app.logger.error('Could not map COLIN country %s to an alpha-2 code', country) + cls._country_cache[country] = country + return cls._country_cache[country] + + @staticmethod + def _to_int(value) -> Optional[int]: + """Coerce an Oracle NUMBER to int, preserving None.""" + return int(value) if value is not None else None + + @staticmethod + def _to_iso_datetime(value: Optional[str]) -> Optional[str]: + """Rewrite convert_to_json_datetime's '-00:00' suffix to the ISO '+00:00' LEAR uses.""" + if value and value.endswith('-00:00'): + return value[:-6] + '+00:00' + return value diff --git a/colin-api/src/colin_api/models/shares.py b/colin-api/src/colin_api/models/shares.py index 2124ea7894..3e9d832578 100644 --- a/colin-api/src/colin_api/models/shares.py +++ b/colin-api/src/colin_api/models/shares.py @@ -58,6 +58,7 @@ class ShareClass(Share): # pylint: disable=too-many-instance-attributes; # pylint: disable=too-few-public-methods currency_type = None + other_currency = None has_par_value = None par_value_amt = None series = None @@ -77,6 +78,7 @@ def to_dict(self): 'maxNumberOfShares': self.max_number_shares, 'parValue': self.par_value_amt, 'currency': self.currency_type, + 'currencyAdditional': self.other_currency, 'hasMaximumShares': self.has_max_shares == 'N' or False, 'hasParValue': self.has_par_value == 'Y' or False, 'hasRightsOrRestrictions': self.has_special_rights == 'Y' or False, @@ -140,6 +142,7 @@ def _get_share_classes(cls, cursor, event_id, corp_num): row = dict(zip([x[0].lower() for x in description], row)) share_class = ShareClass() share_class.currency_type = row['currency_typ_cd'] + share_class.other_currency = row['other_currency'] share_class.has_max_shares = row['max_share_ind'] share_class.has_special_rights = row['spec_rights_ind'] share_class.has_par_value = row['par_value_ind'] diff --git a/colin-api/src/colin_api/resources/business.py b/colin-api/src/colin_api/resources/business.py index 9a52b8c41b..ff1f859511 100644 --- a/colin-api/src/colin_api/resources/business.py +++ b/colin-api/src/colin_api/resources/business.py @@ -21,7 +21,7 @@ from flask_restx import Namespace, Resource, cors from colin_api.exceptions import GenericException -from colin_api.models import Business, CorpName +from colin_api.models import Business, BusinessSnapshot, CorpName from colin_api.resources.db import DB from colin_api.utils.auth import COLIN_SVC_ROLE, jwt from colin_api.utils.util import cors_preflight @@ -91,6 +91,31 @@ def get(identifier: str): ), HTTPStatus.INTERNAL_SERVER_ERROR +@cors_preflight('GET') +@API.route('//snapshot', methods=['GET', 'OPTIONS']) +class BusinessSnapshotInfo(Resource): + """LEAR-shaped snapshot of a COLIN business.""" + + @staticmethod + @cors.crossdomain(origin='*') + @jwt.requires_roles([COLIN_SVC_ROLE]) + def get(identifier: str): + """Return the business/parties/offices/shareClasses/resolutions snapshot.""" + try: + snapshot = BusinessSnapshot.get_snapshot(identifier) + return jsonify(snapshot), HTTPStatus.OK + + except GenericException as err: # pylint: disable=duplicate-code + return jsonify({'message': err.error}), err.status_code + + except Exception as err: # pylint: disable=broad-except; want to catch all errors + # general catch-all exception + current_app.logger.error(err.with_traceback(None)) + return jsonify( + {'message': 'Error when trying to retrieve business record from COLIN'} + ), HTTPStatus.INTERNAL_SERVER_ERROR + + @cors_preflight('GET, POST') @API.route('//', methods=['GET']) @API.route('/', methods=['POST']) diff --git a/colin-api/src/colin_api/version.py b/colin-api/src/colin_api/version.py index 49ceb9a9f8..74e4fcb3db 100644 --- a/colin-api/src/colin_api/version.py +++ b/colin-api/src/colin_api/version.py @@ -22,4 +22,4 @@ Development release segment: .devN """ -__version__ = '2.171.9' # pylint: disable=invalid-name +__version__ = '2.172.0' # pylint: disable=invalid-name diff --git a/colin-api/tests/unit/__init__.py b/colin-api/tests/unit/__init__.py index d755f6a1d8..2201725833 100644 --- a/colin-api/tests/unit/__init__.py +++ b/colin-api/tests/unit/__init__.py @@ -15,4 +15,124 @@ """The Unit Test for the API. For our purposes this server and its Postgres Database are part of the Unit Test Suite. + +Also holds the shared builders for COLIN model objects used by the mocked-Oracle test +suites (auth-info, snapshot) - the fixtures composing them live in conftest.py. """ +from colin_api.models import Business, Office, Party, ShareObject +from colin_api.models.shares import Share, ShareClass +from colin_api.utils.auth import jwt as _jwt + + +# what Address.as_dict emits +RAW_ADDRESS = { + 'streetAddress': '123 FAKE ST', + 'streetAddressAdditional': '', + 'addressCity': 'VICTORIA', + 'addressRegion': 'BC', + 'addressCountry': 'CANADA', + 'postalCode': 'V8V 8V8', + 'deliveryInstructions': '', + 'addressId': 4444, + 'actions': [] +} + +# the LEAR shape of the same address +LEAR_ADDRESS = { + 'id': 4444, + 'streetAddress': '123 FAKE ST', + 'streetAddressAdditional': '', + 'addressCity': 'VICTORIA', + 'addressRegion': 'BC', + 'addressCountry': 'CA', + 'postalCode': 'V8V 8V8', + 'deliveryInstructions': '' +} + + +def bypass_auth(mocker, roles_valid=True): + """Stub out token validation on the jwt manager. + + The attribute names differ between flask-jwt-oidc releases (and validate_roles has taken + different arities), so patch whichever the installed version exposes. MagicMock accepts + any signature, so this holds across versions. + """ + for attr in ('_require_auth_validation', '_validate_token', 'validate_token'): + if hasattr(_jwt, attr): + mocker.patch.object(_jwt, attr, return_value=None) + mocker.patch.object(_jwt, 'validate_roles', return_value=roles_valid) + + +def build_business(**overrides): + """Return a Business object as find_by_identifier would build it.""" + business = Business() + business.corp_num = '0870226' + business.corp_name = 'COLIN TEST COMPANY LTD.' + business.corp_type = 'BC' + # CORP_OP_STATE.OP_STATE_TYP_CD - only ever ACT/HIS; drives the response's LEAR-style state + business.corp_state_class = 'ACT' + business.good_standing = True + business.business_number = '791861078BC0001' + # COLIN returns admin_freeze as a 'True'/'False' string + business.admin_freeze = 'False' + # convert_to_json_datetime emits a '-00:00' suffix + business.founding_date = '2000-01-01T08:00:00-00:00' + business.email = 'registered.office@test.com' + for key, value in overrides.items(): + setattr(business, key, value) + return business + + +def build_director(): + """Return a Party object as Party.get_current would build it.""" + party = Party() + party.officer = { + 'firstName': 'JANE', 'lastName': 'DOE', 'middleInitial': '', + 'organizationName': '', 'partyType': 'person' + } + party.delivery_address = dict(RAW_ADDRESS) + party.mailing_address = dict(RAW_ADDRESS) + party.title = '' + party.appointment_date = '2010-05-05' + party.cessation_date = None + party.start_event_id = 111 + party.end_event_id = '' + party.corp_party_id = 999 + party.roles = [{'roleType': 'Director', 'appointmentDate': '2010-05-05', 'cessationDate': None}] + return party + + +def build_office(office_type): + """Return an Office object as Office.get_current would build it.""" + office = Office() + office.office_type = office_type + office.delivery_address = dict(RAW_ADDRESS) + office.mailing_address = dict(RAW_ADDRESS) + return office + + +def build_share_structure(): + """Return the current ShareObject as ShareObject.get_all would build it.""" + series = Share() + series.share_id = 1 + series.share_name = 'SERIES 1' + series.has_max_shares = 'Y' # COLIN semantics: 'N' means has a maximum + series.has_special_rights = 'N' + series.max_number_shares = None + + share_class = ShareClass() + share_class.share_id = 0 + share_class.share_name = 'CLASS A' + share_class.currency_type = 'OTH' + share_class.other_currency = 'BITCOIN' + share_class.has_max_shares = 'N' + share_class.has_par_value = 'Y' + share_class.has_special_rights = 'Y' + share_class.par_value_amt = 1.5 + share_class.max_number_shares = 10000 + share_class.series = [series] + + share_struct = ShareObject() + share_struct.end_event_id = None + share_struct.share_classes = [share_class] + return share_struct diff --git a/colin-api/tests/unit/api/test_business_auth_info.py b/colin-api/tests/unit/api/test_business_auth_info.py index bbb6dfd1c9..423892c1d9 100644 --- a/colin-api/tests/unit/api/test_business_auth_info.py +++ b/colin-api/tests/unit/api/test_business_auth_info.py @@ -22,73 +22,20 @@ own behaviour: the corp password query, the corp type restriction, response shape and error mapping. """ -from types import SimpleNamespace -from unittest.mock import MagicMock - -import pytest - from colin_api.exceptions import BusinessNotFoundException from colin_api.models import Business -from colin_api.utils.auth import jwt as _jwt +from tests.unit import build_business, bypass_auth AUTH_INFO_URL = '/api/v1/businesses/BC0870226/auth-info' PASS_CODE = '111111111' -def _business(**overrides): - """Return a Business object as find_by_identifier would build it.""" - business = Business() - business.corp_num = '0870226' - business.corp_name = 'COLIN TEST COMPANY LTD.' - business.corp_type = 'BC' - # CORP_OP_STATE.OP_STATE_TYP_CD - only ever ACT/HIS; drives the response's LEAR-style status - business.corp_state_class = 'ACT' - business.good_standing = True - business.business_number = '791861078BC0001' - # COLIN returns admin_freeze as a 'True'/'False' string - business.admin_freeze = 'False' - business.email = 'registered.office@test.com' - for key, value in overrides.items(): - setattr(business, key, value) - return business - - -def _bypass_auth(mocker, roles_valid=True): - """Stub out token validation on the jwt manager. - - The attribute names differ between flask-jwt-oidc releases (and validate_roles has taken - different arities), so patch whichever the installed version exposes. MagicMock accepts - any signature, so this holds across versions. - """ - for attr in ('_require_auth_validation', '_validate_token', 'validate_token'): - if hasattr(_jwt, attr): - mocker.patch.object(_jwt, attr, return_value=None) - mocker.patch.object(_jwt, 'validate_roles', return_value=roles_valid) - - -@pytest.fixture -def authorized(mocker): - """Bypass the colin service role check - the gate itself is asserted separately.""" - _bypass_auth(mocker) - - -@pytest.fixture -def mock_db(mocker): - """Mock the Oracle connection, exposing the connection and cursor for assertions.""" - cursor = MagicMock() - cursor.fetchone.return_value = (PASS_CODE,) - connection = MagicMock() - connection.cursor.return_value = cursor - db = MagicMock() - db.connection = connection - mocker.patch('colin_api.models.business.DB', db) - return SimpleNamespace(connection=connection, cursor=cursor) - - def test_get_auth_info(client, mocker, authorized, mock_db): # pylint: disable=unused-argument """Assert the auth info needed by auth is returned for a COLIN corp.""" - mocker.patch.object(Business, 'find_by_identifier', return_value=_business()) + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) + # the corp password row - conftest's mock_db defaults fetchone to a count-style (0,) + mock_db.cursor.fetchone.return_value = (PASS_CODE,) rv = client.get(AUTH_INFO_URL) @@ -108,7 +55,7 @@ def test_get_auth_info(client, mocker, authorized, mock_db): # pylint: disable= def test_get_auth_info_maps_historical_state(client, mocker, authorized, mock_db): # pylint: disable=unused-argument """Assert a non-active corp (eg. amalgamated or dissolved) reports the LEAR-style HISTORICAL.""" - mocker.patch.object(Business, 'find_by_identifier', return_value=_business(corp_state_class='HIS')) + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business(corp_state_class='HIS')) rv = client.get(AUTH_INFO_URL) @@ -118,7 +65,7 @@ def test_get_auth_info_maps_historical_state(client, mocker, authorized, mock_db def test_get_auth_info_strips_bc_prefix(client, mocker, authorized, mock_db): # pylint: disable=unused-argument """Assert the BC prefix is stripped, since COLIN stores the bare corp number.""" - find = mocker.patch.object(Business, 'find_by_identifier', return_value=_business()) + find = mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) assert client.get(AUTH_INFO_URL).status_code == 200 @@ -134,7 +81,7 @@ def test_get_auth_info_restricted_to_in_scope_corp_types(client, mocker, authori The response carries a credential, so the surface is limited to the corp types that can be affiliated from COLIN while not loaded in LEAR. """ - find = mocker.patch.object(Business, 'find_by_identifier', return_value=_business()) + find = mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) client.get(AUTH_INFO_URL) @@ -143,7 +90,7 @@ def test_get_auth_info_restricted_to_in_scope_corp_types(client, mocker, authori def test_get_auth_info_queries_corp_password(client, mocker, authorized, mock_db): # pylint: disable=unused-argument """Assert the passcode is read from the corporation corp_password column.""" - mocker.patch.object(Business, 'find_by_identifier', return_value=_business()) + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) client.get(AUTH_INFO_URL) @@ -157,7 +104,7 @@ def test_get_auth_info_reuses_single_connection(client, mocker, authorized, DB.connection acquires a session from a pool capped at 10, so the business lookup and the passcode lookup must share one. """ - find = mocker.patch.object(Business, 'find_by_identifier', return_value=_business()) + find = mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) client.get(AUTH_INFO_URL) @@ -169,7 +116,7 @@ def test_get_auth_info_reuses_single_connection(client, mocker, authorized, def test_get_auth_info_normalizes_admin_freeze(client, mocker, authorized, mock_db): # pylint: disable=unused-argument """Assert COLIN's 'True'/'False' string is returned as a real boolean.""" - mocker.patch.object(Business, 'find_by_identifier', return_value=_business(admin_freeze='True')) + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business(admin_freeze='True')) rv = client.get(AUTH_INFO_URL) @@ -179,7 +126,7 @@ def test_get_auth_info_normalizes_admin_freeze(client, mocker, authorized, def test_get_auth_info_without_passcode(client, mocker, authorized, mock_db): # pylint: disable=unused-argument """Assert a business with no corp password returns a null passcode rather than failing.""" - mocker.patch.object(Business, 'find_by_identifier', return_value=_business()) + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) mock_db.cursor.fetchone.return_value = None rv = client.get(AUTH_INFO_URL) @@ -212,7 +159,7 @@ def test_get_auth_info_handles_unexpected_error(client, mocker, authorized, def test_get_auth_info_requires_colin_service_role(client, mocker): """Assert the endpoint is gated on the colin service role.""" - _bypass_auth(mocker, roles_valid=False) + bypass_auth(mocker, roles_valid=False) rv = client.get(AUTH_INFO_URL) diff --git a/colin-api/tests/unit/api/test_business_snapshot.py b/colin-api/tests/unit/api/test_business_snapshot.py new file mode 100644 index 0000000000..a9d38a1f1b --- /dev/null +++ b/colin-api/tests/unit/api/test_business_snapshot.py @@ -0,0 +1,184 @@ +# Copyright © 2026 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 business snapshot end-point.""" +from colin_api.exceptions import BusinessNotFoundException, PartiesNotFoundException +from colin_api.models import Business, Party, ShareObject +from tests.unit import LEAR_ADDRESS, build_business, bypass_auth + + +SNAPSHOT_URL = '/api/v1/businesses/BC0870226/snapshot' + + +def test_get_snapshot(client, mocker, authorized, mock_db, mock_lookups): # pylint: disable=unused-argument + """Assert the full LEAR-normalized snapshot is returned.""" + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 200 + assert rv.json == { + 'business': { + 'identifier': 'BC0870226', + 'legalName': 'COLIN TEST COMPANY LTD.', + 'legalType': 'BC', + 'state': 'ACTIVE', + 'goodStanding': True, + 'adminFreeze': False, + 'foundingDate': '2000-01-01T08:00:00+00:00', + 'taxId': '791861078BC0001', + 'hasFutureEffectiveFiling': False + }, + 'parties': [{ + 'officer': { + 'id': 999, + 'firstName': 'JANE', + 'lastName': 'DOE', + 'middleInitial': '', + 'organizationName': '', + 'partyType': 'person', + 'email': None + }, + 'deliveryAddress': LEAR_ADDRESS, + 'mailingAddress': LEAR_ADDRESS, + 'roles': [{'roleType': 'Director', 'appointmentDate': '2010-05-05', 'cessationDate': None}] + }], + 'offices': { + 'registeredOffice': {'deliveryAddress': LEAR_ADDRESS, 'mailingAddress': LEAR_ADDRESS}, + 'recordsOffice': {'deliveryAddress': LEAR_ADDRESS, 'mailingAddress': LEAR_ADDRESS} + }, + 'shareClasses': [{ + 'id': 0, + 'name': 'CLASS A', + 'priority': 0, + 'hasMaximumShares': True, + 'maxNumberOfShares': 10000, + 'hasParValue': True, + 'parValue': 1.5, + 'currency': 'OTH', + 'currencyAdditional': 'BITCOIN', + 'hasRightsOrRestrictions': True, + 'series': [{ + 'id': 1, + 'name': 'SERIES 1', + 'priority': 1, + 'hasMaximumShares': False, + 'maxNumberOfShares': None, + 'hasRightsOrRestrictions': False + }] + }], + 'resolutions': [{'date': '2020-01-01'}, {'date': '2019-06-15'}] + } + + +def test_get_snapshot_maps_historical_state(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert a non-active corp (eg. amalgamated or dissolved) reports the LEAR-style HISTORICAL.""" + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business(corp_state_class='HIS')) + + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 200 + assert rv.json['business']['state'] == 'HISTORICAL' + + +def test_get_snapshot_reports_future_effective_filing(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert an outstanding future effective filing is reported.""" + mock_db.cursor.fetchone.return_value = (1,) + + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 200 + assert rv.json['business']['hasFutureEffectiveFiling'] is True + # the count query runs against the bare corp num + assert mock_db.cursor.execute.call_args.kwargs['corp_num'] == '0870226' + assert 'effective_dt' in mock_db.cursor.execute.call_args.args[0] + + +def test_get_snapshot_without_parties(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert a corp with no current directors on file still returns a snapshot.""" + mocker.patch.object(Party, 'get_current', side_effect=PartiesNotFoundException(identifier='0870226')) + + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 200 + assert rv.json['parties'] == [] + + +def test_get_snapshot_without_share_structure(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert a corp with no share structure returns an empty list rather than failing.""" + mocker.patch.object(ShareObject, 'get_all', return_value=None) + + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 200 + assert rv.json['shareClasses'] == [] + + +def test_get_snapshot_null_good_standing(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert COLIN's tri-state good standing passes through as null when unknown.""" + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business(good_standing=None)) + + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 200 + assert rv.json['business']['goodStanding'] is None + + +def test_get_snapshot_single_connection_and_scope(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert one pooled session is used and the lookup is restricted to in-scope corp types.""" + find = mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) + + assert client.get(SNAPSHOT_URL).status_code == 200 + + # BC prefix stripped, same restriction as auth-info, and the pooled session is shared + assert find.call_args.args[0] == '0870226' + assert find.call_args.kwargs['corp_types'] == ['BC', 'ULC', 'CC'] + assert find.call_args.kwargs['con'] is mock_db.connection + assert mock_db.connection.cursor.call_count == 1 + + +def test_get_snapshot_no_results(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert a business that does not exist in COLIN returns a 404.""" + mocker.patch.object(Business, 'find_by_identifier', + side_effect=BusinessNotFoundException(identifier='BC0000000')) + + rv = client.get('/api/v1/businesses/BC0000000/snapshot') + + assert rv.status_code == 404 + assert None is not rv.json['message'] + + +def test_get_snapshot_handles_unexpected_error(client, mocker, authorized, mock_db, + mock_lookups): # pylint: disable=unused-argument + """Assert an unexpected failure returns a 500 without leaking internals.""" + mocker.patch.object(Business, 'find_by_identifier', side_effect=Exception('oracle exploded')) + + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 500 + assert 'oracle exploded' not in str(rv.json) + + +def test_get_snapshot_requires_colin_service_role(client, mocker): + """Assert the endpoint is gated on the colin service role.""" + bypass_auth(mocker, roles_valid=False) + + rv = client.get(SNAPSHOT_URL) + + assert rv.status_code == 401 diff --git a/colin-api/tests/unit/conftest.py b/colin-api/tests/unit/conftest.py index 0d196d4763..6eecd547ae 100644 --- a/colin-api/tests/unit/conftest.py +++ b/colin-api/tests/unit/conftest.py @@ -12,11 +12,17 @@ # See the License for the specific language governing permissions and # limitations under the License. """Common setup and fixtures for the pytest suite used by this service.""" +from types import SimpleNamespace +from unittest.mock import MagicMock + import pytest from sqlalchemy import event, text from colin_api import create_app from colin_api import jwt as _jwt +from colin_api.models import Business, Office, Party, ShareObject + +from . import build_business, build_director, build_office, build_share_structure, bypass_auth @pytest.fixture(scope='session') @@ -54,6 +60,46 @@ def client_ctx(app): # pylint: disable=redefined-outer-name yield _client +@pytest.fixture +def authorized(mocker): + """Bypass the colin service role check - the gate itself is asserted separately.""" + bypass_auth(mocker) + + +@pytest.fixture +def mock_db(mocker): + """Mock the Oracle connection for the modules that acquire it directly. + + cursor.fetchone defaults to (0,) so count-style lookups (eg. the snapshot's + future-effective filing count) read zero; tests needing a specific row (eg. the + corp password) override the return value. + """ + cursor = MagicMock() + cursor.fetchone.return_value = (0,) + connection = MagicMock() + connection.cursor.return_value = cursor + db = MagicMock() # pylint: disable=invalid-name; mirrors the patched module attribute + db.connection = connection + mocker.patch('colin_api.models.business.DB', db) + mocker.patch('colin_api.models.business_snapshot.DB', db) + return SimpleNamespace(connection=connection, cursor=cursor) + + +@pytest.fixture +def mock_lookups(mocker): + """Stub the model lookups the snapshot composes, with realistic model objects.""" + mocker.patch.object(Business, 'find_by_identifier', return_value=build_business()) + mocker.patch.object(Party, 'get_current', return_value=[build_director()]) + mocker.patch.object( + Office, 'get_current', + # the liquidation office proves out-of-scope office types are dropped + return_value=[build_office('registeredOffice'), build_office('recordsOffice'), + build_office('liquidationOffice')] + ) + mocker.patch.object(ShareObject, 'get_all', return_value=[build_share_structure()]) + mocker.patch.object(Business, 'get_resolutions', return_value=['2020-01-01', '2019-06-15']) + + @pytest.fixture(scope='function') def session(app, db): # pylint: disable=redefined-outer-name, invalid-name """Return a function-scoped session.""" diff --git a/colin-api/tests/unit/models/__init__.py b/colin-api/tests/unit/models/__init__.py new file mode 100644 index 0000000000..2ae0b1ff88 --- /dev/null +++ b/colin-api/tests/unit/models/__init__.py @@ -0,0 +1 @@ +"""Unit tests for colin-api models.""" diff --git a/colin-api/tests/unit/models/test_shares.py b/colin-api/tests/unit/models/test_shares.py new file mode 100644 index 0000000000..bd85533fcf --- /dev/null +++ b/colin-api/tests/unit/models/test_shares.py @@ -0,0 +1,40 @@ +# Copyright © 2026 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 for the ShareObject model.""" +from unittest.mock import MagicMock + +from colin_api.models import ShareObject + + +def test_get_share_classes_carries_other_currency(): + """Assert the OTH free-text currency is read and serialized as currencyAdditional.""" + cursor = MagicMock() + cursor.description = [ + ('SHARE_CLASS_ID',), ('CURRENCY_TYP_CD',), ('MAX_SHARE_IND',), ('SHARE_QUANTITY',), + ('SPEC_RIGHTS_IND',), ('PAR_VALUE_IND',), ('PAR_VALUE_AMT',), ('CLASS_NME',), ('OTHER_CURRENCY',) + ] + # one class row, then no series rows for it + cursor.fetchall.side_effect = [ + [(0, 'OTH', 'N', 5000, 'Y', 'Y', 2.0, 'CLASS A', 'BITCOIN')], + [] + ] + + # pylint: disable-next=protected-access + share_classes = ShareObject._get_share_classes(cursor, event_id=1, corp_num='0870226') + + assert len(share_classes) == 1 + assert share_classes[0].other_currency == 'BITCOIN' + assert share_classes[0].to_dict()['currencyAdditional'] == 'BITCOIN' + assert share_classes[0].to_dict()['currency'] == 'OTH' From 09fce5c71cfac86072c03a2f36dcfa8d15901e40 Mon Sep 17 00:00:00 2001 From: Kial Jinnah Date: Wed, 19 Aug 2026 14:35:36 -0400 Subject: [PATCH 2/4] chore: test fix Signed-off-by: Kial Jinnah --- colin-api/tests/unit/api/test_business_snapshot.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/colin-api/tests/unit/api/test_business_snapshot.py b/colin-api/tests/unit/api/test_business_snapshot.py index a9d38a1f1b..23c3ea1eea 100644 --- a/colin-api/tests/unit/api/test_business_snapshot.py +++ b/colin-api/tests/unit/api/test_business_snapshot.py @@ -147,8 +147,8 @@ def test_get_snapshot_single_connection_and_scope(client, mocker, authorized, mo # BC prefix stripped, same restriction as auth-info, and the pooled session is shared assert find.call_args.args[0] == '0870226' - assert find.call_args.kwargs['corp_types'] == ['BC', 'ULC', 'CC'] assert find.call_args.kwargs['con'] is mock_db.connection + assert 'corp_types' not in find.call_args.kwargs assert mock_db.connection.cursor.call_count == 1 From 1a4bdf977c5e25ec6b64218b7b874aaf27ca250a Mon Sep 17 00:00:00 2001 From: Kial Jinnah Date: Wed, 19 Aug 2026 15:08:56 -0400 Subject: [PATCH 3/4] chore: sonarqube Signed-off-by: Kial Jinnah --- colin-api/src/colin_api/models/filing.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/colin-api/src/colin_api/models/filing.py b/colin-api/src/colin_api/models/filing.py index d419929e2b..f55d51176a 100644 --- a/colin-api/src/colin_api/models/filing.py +++ b/colin-api/src/colin_api/models/filing.py @@ -1298,7 +1298,7 @@ def get_future_effective_filings(cls, business: Business) -> List: """Get the list of all future effective filings for a business.""" try: future_effective_filings = [] - current_date = datetime.datetime.utcnow().strftime('%Y-%m-%d') + current_date = datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d') cursor = DB.connection.cursor() cursor.execute( """ From 4d93bd1a9ec86fafd3915a810b2751a554b2bbf0 Mon Sep 17 00:00:00 2001 From: Kial Jinnah Date: Wed, 19 Aug 2026 15:13:29 -0400 Subject: [PATCH 4/4] chore: sonarqube Signed-off-by: Kial Jinnah --- colin-api/src/colin_api/models/business_snapshot.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/colin-api/src/colin_api/models/business_snapshot.py b/colin-api/src/colin_api/models/business_snapshot.py index 42f3d4d08d..662de119f4 100644 --- a/colin-api/src/colin_api/models/business_snapshot.py +++ b/colin-api/src/colin_api/models/business_snapshot.py @@ -14,7 +14,7 @@ """Snapshot of a COLIN business, normalized to LEAR structure.""" from __future__ import annotations -from datetime import datetime +from datetime import datetime, timezone from typing import Dict, Optional import pycountry @@ -91,7 +91,7 @@ def _business_dict(cls, business: Business, orig_identifier: str, cursor) -> Dic @staticmethod def _has_future_effective_filing(cursor, corp_num: str) -> bool: """Return whether any filing has an effective date still in the future.""" - current_date = datetime.utcnow().strftime('%Y-%m-%d') + current_date = datetime.now(timezone.utc).strftime('%Y-%m-%d') cursor.execute( """ select count(*)