31921 - Dissolution Completing Party

legal-api/src/legal_api/services/filings/validations/common_validations.py

@@ -20,14 +20,15 @@
 
 import pycountry
 import PyPDF2
-from flask import current_app, g
+from flask import current_app, g, request
 from flask_babel import _
 
 from legal_api.errors import Error
 from legal_api.models import Address, Business, PartyRole
 from legal_api.services import MinioService, colin, flags, namex
 from legal_api.services.bootstrap import AccountService
 from legal_api.services.permissions import ListActionsPermissionsAllowed, PermissionService
+from legal_api.services.request_context import get_request_context
 from legal_api.services.utils import get_str
 from legal_api.utils.datetime import datetime as dt
 
@@ -241,7 +242,7 @@ def check_good_standing_permission(business: Business) -> Optional[Error]:
     if business.good_standing:
         return None
 
-    if not flags.is_on("enable-list-actions-permissions"):
+    if not flags.is_on("enabled-deeper-permission-action"):
         return None
 
     required_permission = ListActionsPermissionsAllowed.OVERRIDE_NIGS.value
@@ -834,17 +835,15 @@ def is_address_changed(addr1: dict, addr2: dict) -> bool:
    ]
    return all(is_same_str(addr1.get(key), addr2.get(key)) for key in keys)
 
-def check_completing_party_permission(msg: list, filing_type:str) -> None:
+def check_completing_party_permission(msg: list, filing_type:str) -> Optional[Error]:
     """Check completing party permission and append error message if not allowed."""
     permission_error = PermissionService.check_user_permission(
         ListActionsPermissionsAllowed.EDITABLE_COMPLETING_PARTY.value,
         message="Permission Denied: You do not have permission to edit the completing party."
     )
     if permission_error:
-        msg.append({
-            "error": permission_error.msg[0].get("message"),
-            "path": f"/filing/{filing_type}/parties"
-        })
+       return permission_error
+    return None
 
 
 def validate_staff_payment(filing_json: dict) -> bool:
@@ -1083,4 +1082,60 @@ def validate_document_delivery_email_changed(email: str, org_id: int) -> dict:
     email_changed = not is_same_str(existing_email, email)
     result["email_changed"] = email_changed
 
-    return result
+    return result
+
+def validate_permission_and_completing_party(business: Optional[Business], filing_json: dict, filing_type: str, msg: list, check_options: Optional[dict] = None
+) -> Optional[Error]:
+    """Validate completing party permission and changes."""
+    if check_options is None:
+        check_options = {}
+    check_name = check_options.get("check_name", True)
+    check_email = check_options.get("check_email", True)
+    check_address = check_options.get("check_address", True)
+    check_document_email = check_options.get("check_document_email", True)
+    # check if completing party is entered
+    completing_party_exists = has_completing_party(filing_json, filing_type)
+    completing_party_result = None
+    account_id = None
+    if get_request_context() and hasattr(request, "headers"):
+        account_id = request.headers.get("account-id",
+                                            request.headers.get("accountId", None))
+    if account_id and completing_party_exists and filing_json.get("filing", {}).get(filing_type, {}).get("parties"):
+        completing_party_result = validate_completing_party(filing_json, filing_type, account_id)
+        if completing_party_result.get("error"):
+            msg.extend(completing_party_result["error"])
+
+        # Check if any relevant fields changed
+        should_check_permission = (
+            (check_email and completing_party_result.get("email_changed")) or
+            (check_name and completing_party_result.get("name_changed")) or
+            (check_address and completing_party_result.get("address_changed"))
+            )
+        if should_check_permission:
+            error = check_completing_party_permission(msg, filing_type)
+            if error:
+                return error
+
+    if check_document_email:
+        return check_document_email_changes(filing_json, filing_type, account_id, msg)
+
+    return None
+
+def check_document_email_changes(
+        filing_json: dict,
+        filing_type: str,
+        account_id: int,
+        msg: list
+) -> Optional[Error]:
+    """Check if document delivery email has changed."""
+    document_optional_email = filing_json.get("filing", {}).get("header", {}).get("documentOptionalEmail")
+    if not (document_optional_email and account_id):
+        return None
+
+    email_validation_result = validate_document_delivery_email_changed(document_optional_email, int(account_id))
+    if email_validation_result.get("error"):
+        msg.extend(email_validation_result["error"])
+    if email_validation_result.get("email_changed"):
+        return check_completing_party_permission(msg, filing_type)
+
+    return None

legal-api/src/legal_api/services/filings/validations/dissolution.py

@@ -28,6 +28,7 @@
     validate_effective_date,
     validate_parties_addresses,
     validate_pdf,
+    validate_permission_and_completing_party,
 )
 from legal_api.services.permissions import (
     ListFilingsPermissionsAllowed,
@@ -74,16 +75,12 @@ def validate(business: Business, dissolution: dict) -> Optional[Error]:
     filing_type = "dissolution"
     dissolution_type = get_str(dissolution, "/filing/dissolution/dissolutionType")
     msg = []
-    # Check good standing permission
-    err = check_good_standing_permission(business)
-    if err:
-        return err
-
-    if flags.is_on("enabled-deeper-permission-action"):
-        err = _validate_dissolution_permission(business, dissolution_type, filing_type)
+
+    if flags.is_on("enabled-deeper-permission-action"):        
+        err = _validate_dissolution_permission(business, dissolution, dissolution_type, filing_type, msg)
         if err:
             return err
-        
+
     err = validate_dissolution_type(dissolution, business.legal_type)
     if err:
         msg.extend(err)
@@ -416,25 +413,28 @@ def _check_dissolution_permission(required_permission: str, dissolution_type: st
     message = "Permission Denied - You do not have permissions file {dissolution_type} {filing_type} filing."
     return PermissionService.check_user_permission(required_permission, message=message)
 
-def _validate_dissolution_permission(business: Business, dissolution_type: str, filing_type: str) -> Optional[Error]:
+def _validate_dissolution_permission(business: Business, dissolution: dict, dissolution_type: str, filing_type: str, msg: list) -> Optional[Error]:
     """Validate dissolution permission based on business and dissolution type."""
 
-    if dissolution_type == DissolutionTypes.ADMINISTRATIVE.value:
+    err = check_good_standing_permission(business)
+    if err:
+        return err
+
+    permission_mapping = {
+        DissolutionTypes.ADMINISTRATIVE.value: ListFilingsPermissionsAllowed.DISSOLUTION_ADMIN_FILING.value,
+        DissolutionTypes.INVOLUNTARY.value: ListFilingsPermissionsAllowed.DISSOLUTION_INVOLUNTARY_FILING.value,
+        DissolutionTypes.VOLUNTARY.value: ListFilingsPermissionsAllowed.DISSOLUTION_VOLUNTARY_FILING.value,
+        DissolutionTypes.DELAY.value: ListFilingsPermissionsAllowed.DISSOLUTION_DELAY_FILING.value
+    }
+    if dissolution_type in permission_mapping:
             error = _check_dissolution_permission(
-                ListFilingsPermissionsAllowed.DISSOLUTION_ADMIN_FILING.value,
+                permission_mapping[dissolution_type],
                 dissolution_type,
                 filing_type
             )
             if error:
                 return error
-    if dissolution_type == DissolutionTypes.INVOLUNTARY.value:
-        error = _check_dissolution_permission(
-            ListFilingsPermissionsAllowed.DISSOLUTION_INVOLUNTARY_FILING.value,
-            dissolution_type,
-            filing_type
-        )
-        if error:
-            return error
+
     if business.legal_type in (Business.LegalTypes.SOLE_PROP.value, Business.LegalTypes.PARTNERSHIP.value):
         error = _check_dissolution_permission(
             ListFilingsPermissionsAllowed.DISSOLUTION_FIRM_FILING.value,
@@ -443,20 +443,14 @@ def _validate_dissolution_permission(business: Business, dissolution_type: str,
         )
         if error:
             return error
-    if dissolution_type == DissolutionTypes.VOLUNTARY.value:
-        error = _check_dissolution_permission(
-            ListFilingsPermissionsAllowed.DISSOLUTION_VOLUNTARY_FILING.value,
-            dissolution_type,
-            filing_type
-        )
-        if error:
-            return error
-
-    if dissolution_type == DissolutionTypes.DELAY.value:
-        error = _check_dissolution_permission(
-            ListFilingsPermissionsAllowed.DISSOLUTION_DELAY_FILING.value,
-            dissolution_type,
-            filing_type
-        )
-        if error:
-            return error
+
+    return validate_permission_and_completing_party(
+            None,
+            dissolution,
+            filing_type,
+            msg,
+            {"check_name":True,
+            "check_email":True,
+            "check_address":True,
+            "check_document_email":True}
+                            )

legal-api/tests/unit/services/filings/validations/test_dissolution.py

@@ -14,6 +14,7 @@
 """Test suite to ensure Voluntary Dissolution is validated correctly."""
 import copy
 from datetime import datetime, timedelta
+from email.policy import default
 from http import HTTPStatus
 from unittest.mock import MagicMock, patch
 from datetime import date
@@ -611,7 +612,9 @@ def test_dissolution_good_standing_permission(session, test_name, good_standing,
         patch.object(PermissionService, 'check_user_permission', return_value=permission_error),
         patch.object(dissolution, 'validate_dissolution_parties_roles', return_value=None),
         patch.object(dissolution, 'validate_affidavit', return_value=None),
-        patch.object(dissolution, '_validate_dissolution_permission', return_value=None)
+        patch.object(dissolution, 'check_good_standing_permission', return_value=permission_error if not good_standing  and not has_permission and flag_enabled else None),
+        patch.object(dissolution, 'validate_permission_and_completing_party', return_value=None),
+        patch.object(dissolution, '_check_dissolution_permission', return_value=None),
     ):
         err = validate(business, filing)
 
@@ -622,4 +625,56 @@ def test_dissolution_good_standing_permission(session, test_name, good_standing,
             assert lists_are_equal(err.msg, [{'message': expected_msg}])
     else:
         assert err is None
-
+
+@pytest.mark.parametrize(
+    'test_name, dissolution_permission_error, completing_party_exists, account_id, has_parties, completing_party_result , legal_type, document_optional_email, email_validation_result, completing_party_permission_error, expected_error',
+    [
+        ('SUCCESS_NO_ERRORS', None, False, None, False, None, 'BC', None, None, None, None),
+        ('FAIL_DISSOLUTION_PERMISSION_ERROR', Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied'}]), False, None, False, None, 'BC', None, None, None, Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied'}])),
+        ('SUCCESS_COMPLETING_PARTY_NO_CHANGES', None, True, '123', True, {'error': [], 'email_changed': False, 'name_changed': False, 'address_changed': False}, 'SP', None, None, None, None),
+        ('FAIL_COMPLETING_PARTY_EMAIL_CHANGED_NO_PERMISSION', None, True, '123', True, {'error': [], 'email_changed': True, 'name_changed': False, 'address_changed': False}, 'SP', None, None, Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}]), Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}])),
+        ('FAIL_COMPLETING_PARTY_NAME_CHANGED_NO_PERMISSION', None, True, '123', True, {'error': [], 'email_changed': False, 'name_changed': True, 'address_changed': False}, 'SP', None, None, Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}]), Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}])),
+        ('FAIL_COMPLETING_PARTY_ADDRESS_CHANGED_NO_PERMISSION', None, True, '123', True, {'error': [], 'email_changed': False, 'name_changed': False, 'address_changed': True}, 'SP', None, None, Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}]), Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}])),
+        ('FAIL_DOCUMENT_EMAIL_PERMISSION_ERROR', None, False, '123', False, None, 'BC', 'test@example.com', {'errors': [], 'email_changed': True}, Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}]), Error(HTTPStatus.FORBIDDEN, [{'message': 'Permission Denied: You do not have permission to edit the completing party.'}])),
+    ]
+)
+def test_dissolution_validate_dissolution_permission_and_completing_party(
+    session, test_name, dissolution_permission_error, completing_party_exists, account_id, has_parties, completing_party_result, legal_type,
+    document_optional_email, email_validation_result, completing_party_permission_error, expected_error):
+    """Test _validate_dissolution_permission and validate_completing_party_permission methods."""
+    business = Business(identifier='BC1234567', legal_type=legal_type)
+    filing = copy.deepcopy(FILING_HEADER)
+    filing['filing']['dissolution'] = copy.deepcopy(DISSOLUTION)
+    msg = []
+    if has_parties:
+        filing['filing']['dissolution']['parties'] = [{'roles': [{'roleType': 'completing_party'}]}]
+    if document_optional_email:
+        filing['filing']['header']['documentOptionalEmail'] = document_optional_email
+    mock_request = MagicMock()
+    if account_id:
+        def mock_get(key, default=None):
+            if key in ('account-id', 'accountId'):
+                return account_id
+            return None
+        mock_request.headers = MagicMock()
+        mock_request.headers.get = MagicMock(side_effect=mock_get)
+    else:
+        mock_request.headers.get = MagicMock(return_value=None)
+    with (
+        patch.object(dissolution, 'check_good_standing_permission', return_value=None),
+        patch.object(dissolution, '_check_dissolution_permission', return_value=dissolution_permission_error),
+        patch('legal_api.services.filings.validations.common_validations.has_completing_party', return_value=completing_party_exists),
+        patch('legal_api.services.filings.validations.common_validations.validate_completing_party', return_value=completing_party_result),
+        patch('legal_api.services.filings.validations.common_validations.validate_document_delivery_email_changed', return_value=email_validation_result if email_validation_result else {'errors': [], 'email_changed': False}),
+        patch('legal_api.services.filings.validations.common_validations.check_completing_party_permission', return_value=completing_party_permission_error),
+        patch('legal_api.services.filings.validations.common_validations.get_request_context', return_value=MagicMock() if account_id else None),
+        patch('legal_api.services.filings.validations.common_validations.request', mock_request),
+        patch.object(flags, 'is_on', return_value=True),
+        ):
+        err = dissolution._validate_dissolution_permission(
+            business,filing,'voluntary','dissolution',msg)
+        if expected_error:
+            assert err is not None
+            assert err.code == expected_error.code
+        else:
+            assert err is None