Skip to content

Commit

Permalink
21777 add exclude future effective filing flag (#2799)
Browse files Browse the repository at this point in the history
* 21777 add exclude future effective filing flag

* refactor: encapsulate the exclude_in_dissolution and exclude_future_effective_filing parameters

* refactor: encapsulate the exclude_in_dissolution and exclude_future_effective_filing parameters

* update naming convention

* add the tests

* update dataclass name

* fix misc
  • Loading branch information
kzdev420 authored Jul 3, 2024
1 parent 07b450f commit 713ac26
Show file tree
Hide file tree
Showing 3 changed files with 39 additions and 15 deletions.
20 changes: 14 additions & 6 deletions legal-api/src/legal_api/services/involuntary_dissolution.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,18 +45,24 @@ class EligibilityDetails:
ar_overdue: bool
transition_overdue: bool

@dataclass
class EligibilityFilters:
"""Details about the exclude of a business for dissolution."""

exclude_in_dissolution: bool = True
exclude_future_effective_filing: bool = False

@classmethod
def check_business_eligibility(
cls, identifier: str, exclude_in_dissolution=True
cls, identifier: str, eligibility_filters: EligibilityFilters = EligibilityFilters()
) -> Tuple[bool, EligibilityDetails]:
"""Return true if the business with provided identifier is eligible for dissolution.
Returns:
eligible (bool): True if the business is eligible for dissolution.
eligibility_details (EligibilityDetails): Details regarding eligibility.
"""
query = cls._get_businesses_eligible_query(exclude_in_dissolution).\
filter(Business.identifier == identifier)
query = cls._get_businesses_eligible_query(eligibility_filters).filter(Business.identifier == identifier)
result = query.one_or_none()

if result is None:
Expand Down Expand Up @@ -94,7 +100,7 @@ def get_in_dissolution_batch_processing(business_id: int):
one_or_none()

@staticmethod
def _get_businesses_eligible_query(exclude_in_dissolution=True):
def _get_businesses_eligible_query(eligibility_filters: EligibilityFilters = EligibilityFilters()):
"""Return SQLAlchemy clause for fetching businesses eligible for involuntary dissolution.
Args:
Expand Down Expand Up @@ -125,7 +131,9 @@ def _get_businesses_eligible_query(exclude_in_dissolution=True):
filter(Business.legal_type.in_(InvoluntaryDissolutionService.ELIGIBLE_TYPES)).\
filter(Business.no_dissolution.is_(False))

if exclude_in_dissolution:
future_effective_filing = False if eligibility_filters.exclude_future_effective_filing \
else _has_future_effective_filing()
if eligibility_filters.exclude_in_dissolution:
query = query.filter(not_(in_dissolution))

query = query.filter(
Expand All @@ -136,7 +144,7 @@ def _get_businesses_eligible_query(exclude_in_dissolution=True):
).\
filter(
~or_(
_has_future_effective_filing(),
future_effective_filing,
_has_delay_of_dissolution_filing(),
_is_limited_restored(),
_is_xpro_from_nwpta()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,24 +25,29 @@ def check_business(business: Business) -> list:

ar_overdue_warning = {
'code': BusinessWarningCodes.MULTIPLE_ANNUAL_REPORTS_NOT_FILED,
'message': 'Multiple annual reports not filed. Eligible for involuntary dissolution.',
'message': 'Multiple annual reports not filed. Eligible for involuntary dissolution.',
'warningType': WarningType.NOT_IN_GOOD_STANDING
}
transition_warning = {
'code': BusinessWarningCodes.TRANSITION_NOT_FILED,
'message': 'Transition filing not filed. Eligible for involuntary dissolution.',
'message': 'Transition filing not filed. Eligible for involuntary dissolution.',
'warningType': WarningType.NOT_IN_GOOD_STANDING
}

eligibility, details = InvoluntaryDissolutionService.check_business_eligibility(business.identifier, True)
eligibility, details = InvoluntaryDissolutionService.check_business_eligibility(
business.identifier, InvoluntaryDissolutionService.EligibilityFilters(exclude_future_effective_filing=True))
if eligibility:
if details.transition_overdue:
result.append(transition_warning)
elif details.ar_overdue:
result.append(ar_overdue_warning)
elif batch_datas := InvoluntaryDissolutionService.get_in_dissolution_batch_processing(business.id):
batch_processing, _ = batch_datas
_, dis_details = InvoluntaryDissolutionService.check_business_eligibility(business.identifier, False)
_, dis_details = InvoluntaryDissolutionService.check_business_eligibility(
business.identifier, InvoluntaryDissolutionService.EligibilityFilters(
exclude_in_dissolution=False, exclude_future_effective_filing=True
)
)
if dis_details.transition_overdue:
result.append(transition_warning)
elif dis_details.ar_overdue:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,24 +20,30 @@
factory_batch,
factory_batch_processing,
factory_business,
factory_completed_filing
factory_completed_filing,
factory_pending_filing
)
from legal_api.services.warnings.business.business_checks import WarningType
from legal_api.services.warnings.business.business_checks.involuntary_dissolution import check_business
from legal_api.utils.datetime import datetime

from datedelta import datedelta
from registry_schemas.example_data import FILING_HEADER, RESTORATION, TRANSITION_FILING_TEMPLATE
from registry_schemas.example_data import CHANGE_OF_ADDRESS, FILING_HEADER, RESTORATION

RESTORATION_FILING = copy.deepcopy(FILING_HEADER)
RESTORATION_FILING['filing']['restoration'] = RESTORATION

CHANGE_OF_ADDRESS_FILING = copy.deepcopy(FILING_HEADER)
CHANGE_OF_ADDRESS_FILING['filing']['changeOfAddress'] = CHANGE_OF_ADDRESS

@pytest.mark.parametrize('test_name, no_dissolution, batch_status, batch_processing_status', [
('NOT_ELIGIBLE', True, None, None),
('ELIGIBLE_AR_OVERDUE', False, 'PROCESSING', 'COMPLETED'),
('ELIGIBLE_TRANSITION_OVERDUE', False, 'PROCESSING', 'COMPLETED'),
('ELIGIBLE_AR_OVERDUE_FUTURE_EFFECTIVE_FILING_HAS_WARNINGS', False, 'PROCESSING', 'COMPLETED'),
('IN_DISSOLUTION_AR_OVERDUE', False, 'PROCESSING', 'PROCESSING'),
('IN_DISSOLUTION_TRANSITION_OVERDUE', False, 'PROCESSING', 'PROCESSING')
('IN_DISSOLUTION_TRANSITION_OVERDUE', False, 'PROCESSING', 'PROCESSING'),
('IN_DISSOLUTION_AR_OVERDUE_FUTURE_EFFECTIVE_FILING_HAS_WARNINGS', False, 'PROCESSING', 'PROCESSING'),
])
def test_check_business(session, test_name, no_dissolution, batch_status, batch_processing_status):
"""Test the check_business function."""
Expand Down Expand Up @@ -68,6 +74,9 @@ def test_check_business(session, test_name, no_dissolution, batch_status, batch_
batch_processing.meta_data = json.dumps(meta_data)
batch_processing.save()

if 'FUTURE_EFFECTIVE_FILING' in test_name:
factory_pending_filing(business, CHANGE_OF_ADDRESS_FILING)

result = check_business(business)

if test_name == 'NOT_ELIGIBLE':
Expand All @@ -78,8 +87,10 @@ def test_check_business(session, test_name, no_dissolution, batch_status, batch_
assert result[1]['code'] == 'DISSOLUTION_IN_PROGRESS'
assert result[1]['message'] == 'Business is in the process of involuntary dissolution.'
assert result[1]['warningType'] == WarningType.INVOLUNTARY_DISSOLUTION

res_meta_data = json.loads(result[1]['data'])
assert res_meta_data == meta_data

if 'TRANSITION_OVERDUE' in test_name:
assert res_meta_data['overdueTransition'] == True
else:
Expand All @@ -90,9 +101,9 @@ def test_check_business(session, test_name, no_dissolution, batch_status, batch_
warning = result[0]
if 'TRANSITION_OVERDUE' in test_name:
assert warning['code'] == 'TRANSITION_NOT_FILED'
assert warning['message'] == 'Transition filing not filed. Eligible for involuntary dissolution.'
assert warning['message'] == 'Transition filing not filed. Eligible for involuntary dissolution.'
assert warning['warningType'] == WarningType.NOT_IN_GOOD_STANDING
else:
assert warning['code'] == 'MULTIPLE_ANNUAL_REPORTS_NOT_FILED'
assert warning['message'] == 'Multiple annual reports not filed. Eligible for involuntary dissolution.'
assert warning['message'] == 'Multiple annual reports not filed. Eligible for involuntary dissolution.'
assert warning['warningType'] == WarningType.NOT_IN_GOOD_STANDING

0 comments on commit 713ac26

Please sign in to comment.