diff --git a/legal-api/src/legal_api/resources/v2/__init__.py b/legal-api/src/legal_api/resources/v2/__init__.py index 09e0624260..3f9a6f1387 100644 --- a/legal-api/src/legal_api/resources/v2/__init__.py +++ b/legal-api/src/legal_api/resources/v2/__init__.py @@ -20,6 +20,7 @@ from .business import bp as businesses_bp from .business.business_digital_credentials import bp_dc as digital_credentials_bp from .configuration import bp as configuration_bp +from .dissolution import bp as dissolution_bp from .document_signature import bp as document_signature_bp from .internal_services import bp as internal_bp from .meta import bp as meta_bp @@ -46,6 +47,7 @@ def init_app(self, app): self.app.register_blueprint(administrative_bn_bp) self.app.register_blueprint(businesses_bp) self.app.register_blueprint(digital_credentials_bp) + self.app.register_blueprint(dissolution_bp) self.app.register_blueprint(document_signature_bp) self.app.register_blueprint(namerequest_bp) self.app.register_blueprint(naics_bp) diff --git a/legal-api/src/legal_api/resources/v2/dissolution.py b/legal-api/src/legal_api/resources/v2/dissolution.py new file mode 100644 index 0000000000..2612c5cd0f --- /dev/null +++ b/legal-api/src/legal_api/resources/v2/dissolution.py @@ -0,0 +1,40 @@ +# Copyright © 2024 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. +"""API endpoints for managing Involuntary Dissolution resources.""" +from http import HTTPStatus + +from flask import Blueprint, jsonify +from flask_cors import cross_origin + +from legal_api.models import UserRoles +from legal_api.services import InvoluntaryDissolutionService +from legal_api.utils.auth import jwt + + +bp = Blueprint('INVOLUNTARY_DISSOLUTION', __name__, url_prefix='/api/v2/admin/dissolutions') + + +@bp.route('/statistics', methods=['GET']) +@cross_origin(origin='*') +@jwt.has_one_of_roles([UserRoles.staff]) +def get_statistics(): + """Return a JSON object with statistic information.""" + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + data = { + 'eligibleCount': count + } + + return jsonify({ + 'data': data + }), HTTPStatus.OK diff --git a/legal-api/src/legal_api/services/__init__.py b/legal-api/src/legal_api/services/__init__.py index 215003351b..70909e282e 100644 --- a/legal-api/src/legal_api/services/__init__.py +++ b/legal-api/src/legal_api/services/__init__.py @@ -34,6 +34,7 @@ from .digital_credentials import DigitalCredentialsService from .document_meta import DocumentMetaService from .flags import Flags +from .involuntary_dissolution import InvoluntaryDissolutionService from .minio import MinioService from .naics import NaicsService from .namex import NameXService diff --git a/legal-api/src/legal_api/services/involuntary_dissolution.py b/legal-api/src/legal_api/services/involuntary_dissolution.py new file mode 100644 index 0000000000..c3b129e138 --- /dev/null +++ b/legal-api/src/legal_api/services/involuntary_dissolution.py @@ -0,0 +1,193 @@ +# Copyright © 2024 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. + +"""This provides the service for involuntary dissolution.""" +from sqlalchemy import and_, exists, func, not_, or_, text +from sqlalchemy.orm import aliased + +from legal_api.models import Batch, BatchProcessing, Business, Filing, db + + +DEFAULT_MIN_DATE = func.date('1800-01-01 00:00:00+00:00') + + +class InvoluntaryDissolutionService(): + """Provides services to get information for involuntary dissolution.""" + + @staticmethod + def get_businesses_eligible_count(): + """Return the number of businesses eligible for involuntary dissolution.""" + eligible_types = [ + Business.LegalTypes.COMP.value, + Business.LegalTypes.BC_ULC_COMPANY.value, + Business.LegalTypes.BC_CCC.value, + Business.LegalTypes.BCOMP.value, + Business.LegalTypes.CONTINUE_IN.value, + Business.LegalTypes.ULC_CONTINUE_IN.value, + Business.LegalTypes.CCC_CONTINUE_IN.value, + Business.LegalTypes.BCOMP_CONTINUE_IN.value, + Business.LegalTypes.EXTRA_PRO_A.value, + Business.LegalTypes.LIMITED_CO.value + ] + + subquery = exists().where(BatchProcessing.business_id == Business.id, + BatchProcessing.status.notin_( + [BatchProcessing.BatchProcessingStatus.WITHDRAWN.value, + BatchProcessing.BatchProcessingStatus.COMPLETED.value]), + BatchProcessing.batch_id == Batch.id, + Batch.status != Batch.BatchStatus.COMPLETED.value, + Batch.batch_type == Batch.BatchType.INVOLUNTARY_DISSOLUTION.value) + + query = db.session.query(Business).\ + filter(Business.state == Business.State.ACTIVE).\ + filter(Business.legal_type.in_(eligible_types)).\ + filter(Business.no_dissolution.is_(False)).\ + filter(not_(subquery)).\ + filter( + or_( + _has_specific_filing_overdue(), + _has_no_transition_filed_after_restoration() + ) + ).\ + filter( + ~or_( + _has_future_effective_filing(), + _has_change_of_address_filing(), + _has_delay_of_dissolution_filing(), + _is_limited_restored(), + _is_xpro_from_nwpta() + ) + ) + + return query.count() + + +def _has_specific_filing_overdue(): + """Return SQLAlchemy clause for specific filing overdue check. + + Check if the date of filed recognition(IA)/restoration/annual report + of the business is over 26 months, whichever is latest. + """ + from legal_api.core.filing import Filing as CoreFiling # pylint: disable=import-outside-toplevel + + latest_date = func.greatest( + Business.founding_date, + db.session.query(func.max(Filing.effective_date)).filter( + Filing.business_id == Business.id, + Filing._filing_type.in_([ # pylint: disable=protected-access + CoreFiling.FilingTypes.RESTORATION.value, + CoreFiling.FilingTypes.RESTORATIONAPPLICATION.value + ]), + Filing._status == Filing.Status.COMPLETED.value # pylint: disable=protected-access + ).scalar_subquery(), + Business.last_ar_date + ) + + latest_date_cutoff = latest_date + text("""INTERVAL '26 MONTHS'""") + + return latest_date_cutoff < func.timezone('UTC', func.now()) + + +def _has_no_transition_filed_after_restoration(): + """Return SQLAlchemy clause for no transition filed after restoration check. + + Check if the business needs to file Transition but does not file it within 12 months after restoration. + """ + from legal_api.core.filing import Filing as CoreFiling # pylint: disable=import-outside-toplevel + + new_act_date = func.date('2004-03-29 00:00:00+00:00') + + restoration_filing = aliased(Filing) + transition_filing = aliased(Filing) + + return exists().where( + and_( + Business.legal_type != Business.LegalTypes.EXTRA_PRO_A.value, + Business.founding_date < new_act_date, + restoration_filing.business_id == Business.id, + restoration_filing._filing_type.in_([ # pylint: disable=protected-access + CoreFiling.FilingTypes.RESTORATION.value, + CoreFiling.FilingTypes.RESTORATIONAPPLICATION.value + ]), + restoration_filing._status == Filing.Status.COMPLETED.value, # pylint: disable=protected-access + not_( + exists().where( + and_( + transition_filing.business_id == Business.id, + transition_filing._filing_type \ + == CoreFiling.FilingTypes.TRANSITION.value, # pylint: disable=protected-access + transition_filing._status == \ + Filing.Status.COMPLETED.value, # pylint: disable=protected-access + transition_filing.effective_date.between( + restoration_filing.effective_date, + restoration_filing.effective_date + text("""INTERVAL '1 YEAR'""") + ) + ) + ) + ) + ) + ) + + +def _has_future_effective_filing(): + """Return SQLAlchemy clause for future effective filing check. + + Check if the business has future effective filings. + """ + return db.session.query(Filing). \ + filter(Filing.business_id == Business.id). \ + filter(Filing._status.in_([Filing.Status.PENDING, Filing.Status.PAID.value])). \ + exists() # pylint: disable=protected-access + + +def _has_change_of_address_filing(): + """Return SQLAlchemy clause for Change of Address filing check. + + Check if the business has Change of Address filings within last 32 days. + """ + coa_date_cutoff = func.coalesce(Business.last_coa_date, DEFAULT_MIN_DATE) + text("""INTERVAL '32 DAYS'""") + return coa_date_cutoff >= func.timezone('UTC', func.now()) + + +def _has_delay_of_dissolution_filing(): + """Return SQLAlchemy clause for Delay of Dissolution filing check. + + Check if the business has Delay of Dissolution filing. + """ + # TODO to implement in the future + return False + + +def _is_limited_restored(): + """Return SQLAlchemy clause for Limited Restoration check. + + Check if the business is in limited restoration status. + """ + return and_( + Business.restoration_expiry_date.isnot(None), + Business.restoration_expiry_date >= func.timezone('UTC', func.now()) + ) + + +def _is_xpro_from_nwpta(): + """Return SQLAlchemy clause for Expro from NWPTA jurisdictions check. + + Check if the business is extraprovincial and from NWPTA jurisdictions. + """ + return and_( + Business.legal_type == Business.LegalTypes.EXTRA_PRO_A.value, + Business.jurisdiction == 'CA', + Business.foreign_jurisdiction_region.isnot(None), + Business.foreign_jurisdiction_region.in_(['AB', 'SK', 'MB']) + ) diff --git a/legal-api/tests/unit/models/__init__.py b/legal-api/tests/unit/models/__init__.py index 385c1700c9..3f758ef6f8 100644 --- a/legal-api/tests/unit/models/__init__.py +++ b/legal-api/tests/unit/models/__init__.py @@ -28,6 +28,7 @@ amalgamation, amalgamating_business, Batch, + BatchProcessing, Business, Comment, Filing, @@ -402,3 +403,21 @@ def factory_batch(batch_type=Batch.BatchType.INVOLUNTARY_DISSOLUTION, ) batch.save() return batch + + +def factory_batch_processing(batch_id, + business_id, + identifier, + step = BatchProcessing.BatchProcessingStep.WARNING_LEVEL_1.value, + status = BatchProcessing.BatchProcessingStatus.PROCESSING.value, + notes = ''): + batch_processing = BatchProcessing( + batch_id = batch_id, + business_id = business_id, + business_identifier = identifier, + step = step, + status = status, + notes = notes + ) + batch_processing.save() + return batch_processing diff --git a/legal-api/tests/unit/resources/v2/test_dissolution.py b/legal-api/tests/unit/resources/v2/test_dissolution.py new file mode 100644 index 0000000000..3fb614af23 --- /dev/null +++ b/legal-api/tests/unit/resources/v2/test_dissolution.py @@ -0,0 +1,39 @@ +# Copyright © 2024 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 dissolution end-point. + +Test-Suite to ensure that admin/dissolutions endpoints are working as expected. +""" +from http import HTTPStatus + +from legal_api.models import UserRoles +from tests.unit.services.utils import create_header + + +def test_get_dissolutions_statistics(session, client, jwt): + """Assert that the endpoint returns statistic information.""" + rv = client.get('/api/v2/admin/dissolutions/statistics', + headers=create_header(jwt, [UserRoles.staff])) + + assert rv.status_code == HTTPStatus.OK + assert 'data' in rv.json + assert 'eligibleCount' in rv.json['data'] + + +def test_get_dissolutions_statistics_invalid_role(session, client, jwt): + """Assert that the endpoint validates invalid user role.""" + rv = client.get('/api/v2/admin/dissolutions/statistics', + headers=create_header(jwt, [UserRoles.basic])) + assert rv.status_code == HTTPStatus.UNAUTHORIZED diff --git a/legal-api/tests/unit/services/test_involuntary_dissolution.py b/legal-api/tests/unit/services/test_involuntary_dissolution.py new file mode 100644 index 0000000000..ded2e13fd7 --- /dev/null +++ b/legal-api/tests/unit/services/test_involuntary_dissolution.py @@ -0,0 +1,279 @@ +# Copyright © 2024 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 Involuntary Dissolution Service. + +Test suite to ensure that the Involuntary Dissolution Service is working as expected. +""" +import copy + +import pytest +from datedelta import datedelta +from registry_schemas.example_data import FILING_HEADER, RESTORATION, TRANSITION_FILING_TEMPLATE + +from legal_api.models import Batch, BatchProcessing, Business +from legal_api.services import InvoluntaryDissolutionService +from legal_api.utils.datetime import datetime +from tests.unit.models import ( + factory_batch, + factory_batch_processing, + factory_business, + factory_completed_filing, + factory_pending_filing, +) + + +RESTORATION_FILING = copy.deepcopy(FILING_HEADER) +RESTORATION_FILING['filing']['restoration'] = RESTORATION + + +def test_get_businesses_eligible_count(session): + """Assert service returns the number of businesses eligible for involuntary dissolution.""" + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + assert count == 0 + + +@pytest.mark.parametrize( + 'test_name, state, exclude', [ + ('TEST_ACTIVE', 'ACTIVE', False), + ('TEST_HISTORICAL', 'HISTORICAL', True), + ('TEST_LIQUIDATION', 'LIQUIDATION', True) + ] +) +def test_get_businesses_eligible_count_active_business(session, test_name, state, exclude): + """Assert service returns eligible count for active businesses.""" + factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value, state=state) + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, legal_type, exclude', [ + ('TEST_BC', 'BC', False), + ('TEST_ULC', 'ULC', False), + ('TEST_CCC', 'CC', False), + ('TEST_BEN', 'BEN', False), + ('TEST_CONTINUE_IN', 'C', False), + ('TEST_ULC_CONTINUE_IN', 'CUL', False), + ('TEST_CCC_CONTINUE_IN', 'CCC', False), + ('TEST_BEN_CONTINUE_IN', 'CBEN', False), + ('TEST_XPRO', 'A', False), + ('TEST_LLC', 'LLC', False), + ('TEST_COOP', 'CP', True), + ('TEST_SP', 'SP', True), + ('TEST_GP', 'GP', True) + ] +) +def test_get_businesses_eligible_count_eligible_type(session, test_name, legal_type, exclude): + """Assert service returns eligible count for businesses with eligible types.""" + factory_business(identifier='BC1234567', entity_type=legal_type) + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, no_dissolution, exclude', [ + ('TEST_NO_DISSOLUTION', True, True), + ('TEST_DISSOLUTION', False, False), + ] +) +def test_get_businesses_eligible_count_no_dissolution(session, test_name, no_dissolution, exclude): + """Assert service returns eligible count for businesses with no_dissolution flag off.""" + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + business.no_dissolution = no_dissolution + business.save() + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, batch_status, batch_processing_status, exclude', [ + ('IN_DISSOLUTION', 'PROCESSING', 'PROCESSING', True), + ('IN_DISSOLUTION_BATCH_COMPLETE', 'COMPLETED', 'WITHDRAWN', False), + ('IN_DISSOLUTION_COMPLETED', 'PROCESSING', 'COMPLETED', False), + ('IN_DISSOLUTION_WITHDRAWN', 'PROCESSING', 'WITHDRAWN', False), + ('NOT_IN_DISSOLUTION', None, None, False), + ] +) +def test_get_businesses_eligible_count_in_dissolution(session, test_name, batch_status, batch_processing_status, exclude): + """Assert service returns eligible count for businesses not already in dissolution.""" + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + if test_name.startswith('IN_DISSOLUTION'): + batch = factory_batch( + batch_type = Batch.BatchType.INVOLUNTARY_DISSOLUTION.value, + status = batch_status, + ) + factory_batch_processing( + batch_id = batch.id, + business_id = business.id, + identifier = business.identifier, + status = batch_processing_status + ) + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, exclude', [ + ('RECOGNITION_OVERDUE', False), + ('RESTORATION_OVERDUE', False), + ('AR_OVERDUE', False), + ('NO_OVERDUE', True) + ] +) +def test_get_businesses_eligible_count_specific_filing_overdue(session, test_name, exclude): + """Assert service returns eligible count including business which has specific filing overdue.""" + if test_name == 'RECOGNITION_OVERDUE': + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + elif test_name == 'RESTORATION_OVERDUE': + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + effective_date = datetime.utcnow() - datedelta(years=3) + factory_completed_filing(business, RESTORATION_FILING, filing_type='restoration', filing_date=effective_date) + elif test_name == 'AR_OVERDUE': + last_ar_date = datetime.utcnow() - datedelta(years=3) + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value, last_ar_date=last_ar_date) + else: + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value, founding_date=datetime.utcnow()) + + business.save() + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, exclude', [ + ('TRANSITION', True), + ('NO_NEED_TRANSITION', True), + ('MISSING_TRANSITION', False) + ] +) +def test_get_businesses_eligible_count_no_transition_filed(session, test_name, exclude): + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value, last_ar_date=datetime.utcnow()) + factory_completed_filing(business, RESTORATION_FILING, filing_type='restoration') + if test_name == 'TRANSITION': + factory_completed_filing(business, TRANSITION_FILING_TEMPLATE, filing_type='transition') + elif test_name == 'NO_NEED_TRANSITION': + business.founding_date = datetime.utcnow() + business.save() + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, exclude', [ + ('FED', True), + ('NO_FED', False) + ] +) +def test_get_businesses_eligible_count_fed_filing(session, test_name, exclude): + """Assert service returns eligible count excluding business which has future effective filings.""" + bussiness = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + if test_name == 'FED': + factory_pending_filing(bussiness, RESTORATION_FILING) + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, exclude', [ + ('COA', True), + ('NO_COA', False) + ] +) +def test_get_businesses_eligible_count_coa_filing(session, test_name, exclude): + """Assert service returns eligible count excluding business which has change of address within last 32 days.""" + bussiness = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + if test_name == 'COA': + bussiness.last_coa_date = datetime.utcnow() + bussiness.save() + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, exclude', [ + ('LIMITED_RESTORATION', True), + ('LIMITED_RESTORATION_EXPIRED', False), + ('NON_LIMITED_RESTORATION', False) + ] +) +def test_get_businesses_eligible_count_limited_restored(session, test_name, exclude): + """Assert service returns eligible count excluding business which is in limited restoration status.""" + bussiness = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + if test_name == 'LIMITED_RESTORATION': + bussiness.restoration_expiry_date = datetime.utcnow() + datedelta(years=1) + elif test_name == 'LIMITED_RESTORATION_EXPIRED': + bussiness.restoration_expiry_date = datetime.utcnow() + datedelta(years=-1) + bussiness.save() + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1 + + +@pytest.mark.parametrize( + 'test_name, jurisdiction, region, exclude', [ + ('XPRO_NWPTA', 'CA', 'AB', True), + ('XPRO_NON_NWPTA', 'CA', 'ON', False), + ('NON_XPRO', None, None, False) + ] +) +def test_get_businesses_eligible_count_xpro_from_nwpta(session, test_name, jurisdiction, region, exclude): + """Assert service returns eligible count excluding expro from NWPTA jurisdictions.""" + if test_name == 'NON_XPRO': + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.COMP.value) + else: + business = factory_business(identifier='BC1234567', entity_type=Business.LegalTypes.EXTRA_PRO_A.value) + business.jurisdiction = jurisdiction + business.foreign_jurisdiction_region = region + business.save() + + count = InvoluntaryDissolutionService.get_businesses_eligible_count() + if exclude: + assert count == 0 + else: + assert count == 1