21331 Dissolutions Job - Implement stage 2 (bcgov#2736)

* 21331 - Dissolutions Job - Implement stage 2 Signed-off-by: Hongjing Chen <[email protected]> * update to use STAGE_1_DELAY & STAGE_2_DELAY Signed-off-by: Hongjing Chen <[email protected]> * update check_run_schedule & add can_run_today Signed-off-by: Hongjing Chen <[email protected]> * match only day of week & fix unit test Signed-off-by: Hongjing Chen <[email protected]> * misc updates after rebasing Signed-off-by: Hongjing Chen <[email protected]> * add nats dependencies into job Signed-off-by: Hongjing Chen <[email protected]> * update batch_processing.notes & add debug logging Signed-off-by: Hongjing Chen <[email protected]> * update unit test - test that check_run_schedule only matches the day of week Signed-off-by: Hongjing Chen <[email protected]> --------- Signed-off-by: Hongjing Chen <[email protected]>
PaulGarewal · Jun 7, 2024 · 7b365da · 7b365da
1 parent ed59c98
commit 7b365da
Show file tree

Hide file tree

Showing 6 changed files with 221 additions and 12 deletions.
diff --git a/jobs/involuntary-dissolutions/involuntary_dissolutions.py b/jobs/involuntary-dissolutions/involuntary_dissolutions.py
@@ -25,7 +25,7 @@
 from legal_api.services.flags import Flags
 from legal_api.services.involuntary_dissolution import InvoluntaryDissolutionService
 from sentry_sdk.integrations.logging import LoggingIntegration
-from sqlalchemy import Date, cast
+from sqlalchemy import Date, cast, func, text
 
 import config  # pylint: disable=import-error
 from utils.logging import setup_logging  # pylint: disable=import-error
@@ -138,6 +138,72 @@ def initiate_dissolution_process(app: Flask):  # pylint: disable=redefined-outer
         app.logger.error(err)
 
 
+def stage_2_process(app: Flask):
+    """Run dissolution stage 2 process for businesses meet moving criteria."""
+    if not (stage_1_delay := app.config.get('STAGE_1_DELAY', None)):
+        app.logger.debug('Skipping stage 2 run since config STAGE_1_DELAY is missing.')
+        return
+
+    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_1
+        )
+        .filter(
+            BatchProcessing.created_date + text(f"""INTERVAL '{stage_1_delay} DAYS'""")
+            <= 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:
+            batch_processing.step = BatchProcessing.BatchProcessingStep.WARNING_LEVEL_2
+        else:
+            batch_processing.status = BatchProcessing.BatchProcessingStatus.WITHDRAWN
+            batch_processing.notes = 'Moved back into good standing'
+        batch_processing.last_modified = datetime.utcnow()
+        batch_processing.save()
+
+
+def can_run_today(cron_value: str):
+    """Check if cron string is valid for today."""
+    tz = pytz.timezone('US/Pacific')
+    today = tz.localize(datetime.today())
+    result = croniter.match(cron_value, datetime(today.year, today.month, today.day))
+    return result
+
+
+def check_run_schedule():
+    """Check if any of the dissolution stage is valid for this run."""
+    stage_1_schedule_config = Configuration.find_by_name(
+        config_name=Configuration.Names.DISSOLUTIONS_STAGE_1_SCHEDULE.value
+    )
+    stage_2_schedule_config = Configuration.find_by_name(
+        config_name=Configuration.Names.DISSOLUTIONS_STAGE_2_SCHEDULE.value
+    )
+    stage_3_schedule_config = Configuration.find_by_name(
+        config_name=Configuration.Names.DISSOLUTIONS_STAGE_3_SCHEDULE.value
+    )
+
+    cron_valid_1 = can_run_today(stage_1_schedule_config.val)
+    cron_valid_2 = can_run_today(stage_2_schedule_config.val)
+    cron_valid_3 = can_run_today(stage_3_schedule_config.val)
+
+    return cron_valid_1, cron_valid_2, cron_valid_3
+
+
 async def run(loop, application: Flask = None):  # pylint: disable=redefined-outer-name
     """Run the stage 1-3 methods for dissolving businesses."""
     if application is None:
@@ -148,13 +214,23 @@ async def run(loop, application: Flask = None):  # pylint: disable=redefined-out
         application.logger.debug(f'enable-involuntary-dissolution flag on: {flag_on}')
         if flag_on:
             # check if batch can be run today
-            new_dissolutions_schedule_config = Configuration.find_by_name(config_name='DISSOLUTIONS_STAGE_1_SCHEDULE')
-            tz = pytz.timezone('US/Pacific')
-            cron_valid = croniter.match(new_dissolutions_schedule_config.val, tz.localize(datetime.today()))
-            if cron_valid:
+            stage_1_valid, stage_2_valid, stage_3_valid = check_run_schedule()
+            application.logger.debug(
+                f'Run schedule check: stage_1: {stage_1_valid}, stage_2: {stage_2_valid}, stage_3: {stage_3_valid}'
+            )
+            if not any([stage_1_valid, stage_2_valid, stage_3_valid]):
+                application.logger.debug(
+                    'Skipping job run since current day of the week does not match any cron schedule.'
+                )
+                return
+
+            if stage_1_valid:
                 initiate_dissolution_process(application)
-            else:
-                application.logger.debug('Skipping job run since current day of the week does not match the cron schedule.')  # noqa: E501
+            if stage_2_valid:
+                stage_2_process(application)
+            if stage_3_valid:
+                pass
+
 
 if __name__ == '__main__':
     application = create_app()

diff --git a/jobs/involuntary-dissolutions/requirements.txt b/jobs/involuntary-dissolutions/requirements.txt
@@ -5,6 +5,8 @@ Jinja2==2.11.3
 MarkupSafe==1.1.1
 Werkzeug==1.0.1
 aniso8601==9.0.1
+asyncio-nats-client==0.11.4
+asyncio-nats-streaming==0.4.0
 attrs==23.1.0
 blinker==1.4
 certifi==2020.12.5
@@ -15,6 +17,8 @@ flask-jwt-oidc==0.3.0
 gunicorn==20.1.0
 itsdangerous==1.1.0
 jsonschema==4.19.0
+nest_asyncio
+protobuf==3.15.8
 pyasn1==0.4.8
 pyrsistent==0.17.3
 python-dotenv==0.17.1
@@ -26,5 +30,5 @@ six==1.15.0
 urllib3==1.26.11
 requests==2.25.1
 cachelib==0.1.1
-legal_api[nats] @ git+https://github.com/bcgov/lear.git#egg=legal_api&subdirectory=legal-api
+git+https://github.com/bcgov/lear.git#egg=legal_api&subdirectory=legal-api
 git+https://github.com/bcgov/[email protected]#egg=registry_schemas
diff --git a/jobs/involuntary-dissolutions/requirements/prod.txt b/jobs/involuntary-dissolutions/requirements/prod.txt
@@ -3,7 +3,10 @@ Flask
 Flask-Script
 Flask-Moment
 Flask-RESTplus
+asyncio-nats-client
+asyncio-nats-streaming
 flask-jwt-oidc>=0.1.5
+nest_asyncio
 python-dotenv
 sentry-sdk[flask]
 werkzeug
diff --git a/jobs/involuntary-dissolutions/tests/unit/test_involuntary_dissolutions.py b/jobs/involuntary-dissolutions/tests/unit/test_involuntary_dissolutions.py
@@ -15,22 +15,42 @@
 """Tests for the Involuntary Dissolutions Job.
 Test suite to ensure that the Involuntary Dissolutions Job is working as expected.
 """
-from datetime import datetime, timedelta 
+from datetime import datetime
 from unittest.mock import MagicMock, patch
 
 import pytest
 import pytz
 from datedelta import datedelta
-from legal_api.models import Batch, BatchProcessing, Business, Configuration, db
+from legal_api.models import Batch, BatchProcessing, Configuration
 
-from involuntary_dissolutions import initiate_dissolution_process
+from involuntary_dissolutions import check_run_schedule, initiate_dissolution_process, stage_2_process
 
 from . import factory_batch, factory_batch_processing, factory_business
 
 
 CREATED_DATE = (datetime.utcnow() + datedelta(days=-60)).replace(tzinfo=pytz.UTC)
 
 
+def test_check_run_schedule():
+    """Assert that schedule check validates the day of run based on cron string in config."""
+    with patch.object(Configuration, 'find_by_name') as mock_find_by_name:
+        mock_stage_1_config = MagicMock()
+        mock_stage_2_config = MagicMock()
+        mock_stage_3_config = MagicMock()
+        mock_stage_1_config.val = '0 0 * * 1-2'
+        mock_stage_2_config.val = '0 0 * * 2'
+        mock_stage_3_config.val = '0 0 * * 3'
+        mock_find_by_name.side_effect = [mock_stage_1_config, mock_stage_2_config, mock_stage_3_config]
+
+        with patch('involuntary_dissolutions.datetime', wraps=datetime) as mock_datetime:
+            mock_datetime.today.return_value = datetime(2024, 6, 4, 1, 2, 3, 4)
+            cron_valid_1, cron_valid_2, cron_valid_3 = check_run_schedule()
+
+            assert cron_valid_1 is True
+            assert cron_valid_2 is True
+            assert cron_valid_3 is False
+
+
 def test_initiate_dissolution_process_job_already_ran(app, session):
     """Assert that the job is skipped correctly if it already ran today."""
     factory_business(identifier='BC1234567')
@@ -84,3 +104,107 @@ def test_initiate_dissolution_process(app, session):
         assert batch_processing.status == BatchProcessing.BatchProcessingStatus.PROCESSING
         assert batch_processing.created_date.date() == datetime.now().date()
         assert batch_processing.meta_data
+
+
+@pytest.mark.parametrize(
+    'test_name, batch_status, status, step, created_date, found', [
+        (
+            'FOUND',
+            Batch.BatchStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStep.WARNING_LEVEL_1,
+            CREATED_DATE,
+            True
+        ),
+        (
+            'NOT_FOUND_BATCH_STATUS',
+            Batch.BatchStatus.HOLD,
+            BatchProcessing.BatchProcessingStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStep.WARNING_LEVEL_1,
+            CREATED_DATE,
+            False
+        ),
+        (
+            'NOT_FOUND_BATCH_PROCESSING_STATUS',
+            Batch.BatchStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStatus.HOLD,
+            BatchProcessing.BatchProcessingStep.WARNING_LEVEL_1,
+            CREATED_DATE,
+            False
+        ),
+        (
+            'NOT_FOUND_STEP',
+            Batch.BatchStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStep.WARNING_LEVEL_2,
+            CREATED_DATE,
+            False
+        ),
+        (
+            'NOT_FOUND_CREATED_DATE',
+            Batch.BatchStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStep.WARNING_LEVEL_1,
+            datetime.utcnow(),
+            False
+        )
+    ]
+
+)
+def test_stage_2_process_find_entry(app, session, test_name, batch_status, status, step, created_date, found):
+    """Assert that only businesses that meet conditions can be processed for stage 2."""
+    business = factory_business(identifier='BC1234567')
+
+    batch = factory_batch(status=batch_status)
+    last_modified = CREATED_DATE
+    batch_processing = factory_batch_processing(
+        batch_id=batch.id,
+        business_id=business.id,
+        identifier=business.identifier,
+        status=status,
+        step=step,
+        created_date=created_date,
+        last_modified=last_modified
+    )
+
+    stage_2_process(app)
+
+    if found:
+        assert batch_processing.last_modified != last_modified
+    else:
+        assert batch_processing.last_modified == last_modified
+
+
+@pytest.mark.parametrize(
+    'test_name, status, step', [
+        (
+            'MOVE_2_STAGE_2',
+            BatchProcessing.BatchProcessingStatus.PROCESSING,
+            BatchProcessing.BatchProcessingStep.WARNING_LEVEL_2
+        ),
+        (
+            'MOVE_BACK_2_GOOD_STANDING',
+            BatchProcessing.BatchProcessingStatus.WITHDRAWN,
+            BatchProcessing.BatchProcessingStep.WARNING_LEVEL_1
+        ),
+    ]
+)
+def test_stage_2_process_update_business(app, session, test_name, status, step):
+    """Assert that businesses are processed correctly."""
+    business = factory_business(identifier='BC1234567')
+    batch = factory_batch(status=Batch.BatchStatus.PROCESSING)
+    batch_processing = factory_batch_processing(
+        batch_id=batch.id,
+        business_id=business.id,
+        identifier=business.identifier,
+        created_date=CREATED_DATE
+    )
+
+    if test_name == 'MOVE_BACK_2_GOOD_STANDING':
+        business.last_ar_date = datetime.utcnow()
+        business.save()
+
+    stage_2_process(app)
+
+    assert batch_processing.status == status
+    assert batch_processing.step == step
diff --git a/jobs/involuntary-dissolutions/tests/unit/test_job.py b/jobs/involuntary-dissolutions/tests/unit/test_job.py
@@ -1,6 +1,8 @@
 import os
+
 import psycopg2
 
+
 def test_connection_failed():
     status = False
     try:

diff --git a/legal-api/src/legal_api/services/involuntary_dissolution.py b/legal-api/src/legal_api/services/involuntary_dissolution.py
@@ -98,7 +98,7 @@ def _get_businesses_eligible_query(exclude_in_dissolution=True):
         """Return SQLAlchemy clause for fetching businesses eligible for involuntary dissolution.
 
         Args:
-            include_in_dissolution (bool): Whether to include the in_dissolution check in the query.
+            exclude_in_dissolution (bool): If True, exclude businesses already in dissolution.
         """
         in_dissolution = (
             exists().where(