21091 Legal API - create staff only endpoint to retrieve businesses e…

…ligible count for involuntary dissolution (bcgov#2683)

* 21091

Signed-off-by: Hongjing Chen <[email protected]>

* add CBEN into eligible types

Signed-off-by: Hongjing Chen <[email protected]>

* great changes to use db query instead

Signed-off-by: Hongjing Chen <[email protected]>

* update unit test

Signed-off-by: Hongjing Chen <[email protected]>

* update criteria for xpro check

Signed-off-by: Hongjing Chen <[email protected]>

* add more tests

Signed-off-by: Hongjing Chen <[email protected]>

* update tests to use factory methods

Signed-off-by: Hongjing Chen <[email protected]>

* add more conditions for in_dissolution check & update tests

Signed-off-by: Hongjing Chen <[email protected]>

* add conditions to ensure filings are completed

Signed-off-by: Hongjing Chen <[email protected]>

* add check for limited restoration & update tests

Signed-off-by: Hongjing Chen <[email protected]>

* updates & use new endpoint

Signed-off-by: Hongjing Chen <[email protected]>

---------

Signed-off-by: Hongjing Chen <[email protected]>

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)

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

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

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'])
+    )

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

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