21332 Dissolutions Job - Implement stage 3 (bcgov#2775)

* Add stage 3 process logic

* Add unit tests

* Fix unit tests

* Add filing_json to involuntary dissolution filing

* Fix linting

* Update unit tests

* Assert queue method called from unit test

jobs/involuntary-dissolutions/config.py

@@ -18,6 +18,7 @@
 
 from dotenv import find_dotenv, load_dotenv
 
+
 load_dotenv(find_dotenv())
 
 CONFIGURATION = {

jobs/involuntary-dissolutions/involuntary_dissolutions.py

@@ -21,11 +21,16 @@
 import sentry_sdk  # noqa: I001, E501; pylint: disable=ungrouped-imports; conflicts with Flake8
 from croniter import croniter
 from flask import Flask
-from legal_api.models import Batch, BatchProcessing, Configuration, db  # noqa: I001
+from legal_api.core.filing import Filing as CoreFiling
+from legal_api.models import Batch, BatchProcessing, Business, Configuration, Filing, db  # noqa: I001
+from legal_api.services.filings.validations.dissolution import DissolutionTypes
 from legal_api.services.flags import Flags
 from legal_api.services.involuntary_dissolution import InvoluntaryDissolutionService
+from legal_api.services.queue import QueueService
+from sentry_sdk import capture_message
 from sentry_sdk.integrations.logging import LoggingIntegration
 from sqlalchemy import Date, cast, func
+from sqlalchemy.orm import aliased
 
 import config  # pylint: disable=import-error
 from utils.logging import setup_logging  # pylint: disable=import-error
@@ -73,7 +78,77 @@ def shell_context():
     app.shell_context_processor(shell_context)
 
 
-def initiate_dissolution_process(app: Flask):  # pylint: disable=redefined-outer-name,too-many-locals
+def create_invountary_dissolution_filing(business_id: int):
+    """Create a filing entry to represent an involuntary dissolution filing."""
+    business = Business.find_by_internal_id(business_id)
+
+    filing = Filing()
+    filing.business_id = business.id
+    filing._filing_type = CoreFiling.FilingTypes.DISSOLUTION  # pylint: disable=protected-access
+    filing._filing_sub_type = DissolutionTypes.INVOLUNTARY  # pylint: disable=protected-access
+    filing.filing_json = {
+        'filing': {
+            'header': {
+                'date': datetime.utcnow().date().isoformat(),
+                'name': 'dissolution',
+                'certifiedBy': ''
+            },
+            'business': {
+                'legalName': business.legal_name,
+                'legalType': business.legal_type,
+                'identifier': business.identifier,
+                'foundingDate': business.founding_date.isoformat()
+            },
+            'dissolution': {
+                'dissolutionDate': datetime.utcnow().date().isoformat(),
+                'dissolutionType': 'involuntary'
+            }
+        }
+    }
+
+    filing.save()
+
+    return filing
+
+
+async def put_filing_on_queue(filing_id: int, app: Flask, qsm: QueueService):
+    """Send queue message to filer to dissolve business."""
+    try:
+        subject = app.config['NATS_FILER_SUBJECT']
+        payload = {'filing': {'id': filing_id}}
+        await qsm.publish_json_to_subject(payload, subject)
+    except Exception as err:  # pylint: disable=broad-except # noqa F841;
+        # mark any failure for human review
+        capture_message(
+            f'Queue Error: Failed to place filing {filing_id} on Queue with error:{err}',
+            level='error'
+        )
+
+
+def mark_eligible_batches_completed():
+    """Mark batches completed if all of their associated batch_processings are compeleted."""
+    AliasBatchProcessing = aliased(BatchProcessing)  # pylint: disable=invalid-name # noqa N806
+    batches = (
+        db.session.query(Batch)
+        .join(BatchProcessing, Batch.id == BatchProcessing.batch_id)
+        .filter(Batch.batch_type == 'INVOLUNTARY_DISSOLUTION')
+        .filter(Batch.status != 'COMPLETED')
+        .filter(
+            ~db.session.query(AliasBatchProcessing)
+            .filter(Batch.id == AliasBatchProcessing.batch_id)
+            .filter(AliasBatchProcessing.status.notin_(['COMPLETED', 'WITHDRAWN']))
+            .exists()
+        )
+        .all()
+    )
+
+    for batch in batches:
+        batch.status = Batch.BatchStatus.COMPLETED
+        batch.end_date = datetime.utcnow()
+        batch.save()
+
+
+def stage_1_process(app: Flask):  # pylint: disable=redefined-outer-name,too-many-locals
     """Initiate dissolution process for new businesses that meet dissolution criteria."""
     try:
         # check if batch has already run today
@@ -133,7 +208,7 @@ def initiate_dissolution_process(app: Flask):  # pylint: disable=redefined-outer
 
 
 def stage_2_process(app: Flask):
-    """Run dissolution stage 2 process for businesses meet moving criteria."""
+    """Update batch processing data for previously created in_progress batches."""
     batch_processings = (
         db.session.query(BatchProcessing)
         .filter(BatchProcessing.batch_id == Batch.id)
@@ -169,6 +244,46 @@ def stage_2_process(app: Flask):
         batch_processing.save()
 
 
+async def stage_3_process(app: Flask, qsm: QueueService):
+    """Process actual dissolution of businesses."""
+    batch_processings = (
+        db.session.query(BatchProcessing)
+        .filter(BatchProcessing.batch_id == Batch.id)
+        .filter(Batch.batch_type == Batch.BatchType.INVOLUNTARY_DISSOLUTION)
+        .filter(Batch.status == Batch.BatchStatus.PROCESSING)
+        .filter(
+            BatchProcessing.status == BatchProcessing.BatchProcessingStatus.PROCESSING
+        )
+        .filter(
+            BatchProcessing.step == BatchProcessing.BatchProcessingStep.WARNING_LEVEL_2
+        )
+        .filter(
+            BatchProcessing.trigger_date <= func.timezone('UTC', func.now())
+        )
+        .all()
+    )
+
+    # TODO: add check if warnings have been sent out & set batch_processing.status to error if not
+
+    for batch_processing in batch_processings:
+        eligible, _ = InvoluntaryDissolutionService.check_business_eligibility(
+            batch_processing.business_identifier, exclude_in_dissolution=False
+        )
+        if eligible:
+            filing = create_invountary_dissolution_filing(batch_processing.business_id)
+            await put_filing_on_queue(filing.id, app, qsm)
+
+            batch_processing.step = BatchProcessing.BatchProcessingStep.DISSOLUTION
+            batch_processing.status = BatchProcessing.BatchProcessingStatus.COMPLETED
+        else:
+            batch_processing.status = BatchProcessing.BatchProcessingStatus.WITHDRAWN
+            batch_processing.notes = 'Moved back into good standing'
+        batch_processing.last_modified = datetime.utcnow()
+        batch_processing.save()
+
+    mark_eligible_batches_completed()
+
+
 def can_run_today(cron_value: str):
     """Check if cron string is valid for today."""
     tz = pytz.timezone('US/Pacific')
@@ -178,7 +293,7 @@ def can_run_today(cron_value: str):
 
 
 def check_run_schedule():
-    """Check if any of the dissolution stage is valid for this run."""
+    """Check if any of the dissolution stages are valid for this run."""
     stage_1_schedule_config = Configuration.find_by_name(
         config_name=Configuration.Names.DISSOLUTIONS_STAGE_1_SCHEDULE.value
     )
@@ -196,7 +311,7 @@ def check_run_schedule():
     return cron_valid_1, cron_valid_2, cron_valid_3
 
 
-async def run(loop, application: Flask = None):  # pylint: disable=redefined-outer-name
+async def run(application: Flask, qsm: QueueService):  # pylint: disable=redefined-outer-name
     """Run the stage 1-3 methods for dissolving businesses."""
     if application is None:
         application = create_app()
@@ -217,18 +332,19 @@ async def run(loop, application: Flask = None):  # pylint: disable=redefined-out
                 return
 
             if stage_1_valid:
-                initiate_dissolution_process(application)
+                stage_1_process(application)
             if stage_2_valid:
                 stage_2_process(application)
             if stage_3_valid:
-                pass
+                await stage_3_process(application, qsm)
 
 
 if __name__ == '__main__':
     application = create_app()
     try:
         event_loop = asyncio.get_event_loop()
-        event_loop.run_until_complete(run(event_loop, application))
+        queue_service = QueueService(app=application, loop=event_loop)
+        event_loop.run_until_complete(run(application, queue_service))
     except Exception as err:  # pylint: disable=broad-except; Catching all errors from the frameworks
         application.logger.error(err)  # pylint: disable=no-member
         raise err

jobs/involuntary-dissolutions/requirements/dev.txt

@@ -3,6 +3,7 @@
 # Testing
 pytest
 pytest-mock
+pytest-asyncio
 requests
 pyhamcrest
 

jobs/involuntary-dissolutions/setup.py

@@ -15,6 +15,7 @@
 
 from setuptools import find_packages, setup
 
+
 setup(
     name='involuntary-dissolutions',
     packages=find_packages()

jobs/involuntary-dissolutions/tests/unit/__init__.py

@@ -14,8 +14,8 @@
 """The Test-Suite used to ensure that the Involuntary Dissolutions Job is working correctly."""
 
 import datetime
-from datedelta import datedelta
 
+from datedelta import datedelta
 from legal_api.models import Batch, BatchProcessing, Business