Catching up with main

backend/audit/intakelib/checks/check_gsa_migration_keyword.py

@@ -33,6 +33,7 @@ def check_for_gsa_migration_keyword(ir):
         "contains_chart_or_table",
         "is_minimis_rate_used",
         "compliance_requirement",
+        "secondary_auditor_address_zipcode",
         "secondary_auditor_name",
         "secondary_auditor_address_state",
     ]

backend/audit/test_schemas.py

@@ -257,7 +257,7 @@ def test_invalid_zip(self):
 
                     with self.assertRaisesRegex(
                         exceptions.ValidationError,
-                        "does not match",
+                        "is not valid under any of the given schemas",
                         msg=f"ValidationError not raised with zip = {bad_zip}",
                     ):
                         validate(instance, schema)
@@ -290,7 +290,7 @@ def test_invalid_zip_plus_4(self):
 
                     with self.assertRaisesRegex(
                         exceptions.ValidationError,
-                        "does not match",
+                        "is not valid under any of the given schemas",
                         msg=f"ValidationError not raised with zip = {bad_zip}",
                     ):
                         validate(instance, schema)

backend/audit/test_validators.py

@@ -884,7 +884,7 @@ def setUp(self):
     def test_no_errors_when_audit_information_is_valid(self):
         """No errors should be raised when audit information is valid"""
         for case in self.SIMPLE_CASES:
-            validate_audit_information_json(case)
+            validate_audit_information_json(case, False)
 
     def test_error_raised_for_missing_required_fields_with_not_gaap(self):
         """Test that missing certain fields raises a validation error when 'gaap_results' contains 'not_gaap'."""
@@ -895,14 +895,24 @@ def test_error_raised_for_missing_required_fields_with_not_gaap(self):
         ]:
             case = jsoncopy(self.SIMPLE_CASES[1])
             del case[required_field]
-            self.assertRaises(ValidationError, validate_audit_information_json, case)
+            self.assertRaises(
+                ValidationError, validate_audit_information_json, case, False
+            )
 
     def test_error_raised_for_missing_required_fields(self):
         """Test that missing required fields raises a validation error."""
         for key in self.SIMPLE_CASES[0].keys():
             case = jsoncopy(self.SIMPLE_CASES[0])
             del case[key]
-            self.assertRaises(ValidationError, validate_audit_information_json, case)
+            self.assertRaises(
+                ValidationError, validate_audit_information_json, case, False
+            )
+
+    def test_gsa_migration(self):
+        case = jsoncopy(self.SIMPLE_CASES[1])
+        case["is_sp_framework_required"] = ""
+        case["is_low_risk_auditee"] = ""
+        self.assertRaises(ValidationError, validate_audit_information_json, case, True)
 
 
 class TribalAccessTests(SimpleTestCase):

backend/audit/validators.py

@@ -250,6 +250,8 @@ def validate_use_of_gsa_migration_keyword(general_information, is_data_migration
         general_information.get("auditee_uei", ""),
         general_information.get("ein", ""),
         general_information.get("auditor_ein", ""),
+        general_information.get("auditee_zip", ""),
+        general_information.get("auditor_zip", ""),
     ]:
         raise ValidationError(
             _(f"{settings.GSA_MIGRATION} not permitted outside of migrations"),
@@ -258,6 +260,22 @@ def validate_use_of_gsa_migration_keyword(general_information, is_data_migration
     return general_information
 
 
+def validate_use_of_gsa_migration_keyword_in_audit_info(
+    audit_information, is_data_migration
+):
+    """Check if GSA_MIGRATION keyword is used and is allowed to be used in audit information"""
+
+    if not is_data_migration and settings.GSA_MIGRATION in [
+        audit_information.get("is_sp_framework_required", ""),
+        audit_information.get("is_low_risk_auditee", ""),
+    ]:
+        raise ValidationError(
+            _(f"{settings.GSA_MIGRATION} not permitted outside of migrations"),
+        )
+
+    return audit_information
+
+
 def validate_general_information_schema_rules(general_information):
     """Check general information schema rules"""
 
@@ -340,7 +358,20 @@ def validate_general_information_complete_json(value, is_data_migration=True):
     return value
 
 
-def validate_audit_information_json(value):
+def validate_audit_information_json(value, is_data_migration=True):
+    """
+    Apply JSON Schema and Python checks to audit information record.
+
+    Keyword arguments:
+    is_data_migration -- True if ignoring GSA_MIGRATION emails. (default True)
+    """
+
+    validate_use_of_gsa_migration_keyword_in_audit_info(value, is_data_migration)
+    validate_audit_information_schema(value)
+    return value
+
+
+def validate_audit_information_schema(value):
     """
     Apply JSON Schema for audit information and report errors.
     """

backend/audit/views/audit_info_form_view.py

@@ -93,7 +93,7 @@ def post(self, request, *args, **kwargs):
                     for key in keys_to_delete:
                         form.cleaned_data.pop(key, None)
 
-                validated = validate_audit_information_json(form.cleaned_data)
+                validated = validate_audit_information_json(form.cleaned_data, False)
 
                 sac.audit_information = validated
                 sac.save(

backend/census_historical_migration/sac_general_lib/general_information.py

@@ -441,6 +441,32 @@ def xform_replace_empty_or_invalid_auditee_ein_with_gsa_migration(general_inform
     return general_information
 
 
+def xform_replace_empty_zips(general_information):
+    """Replaces empty auditor and auditee zipcodes with GSA Migration keyword"""
+    # Transformation recorded.
+    if not general_information.get("auditor_zip"):
+        general_information["auditor_zip"] = settings.GSA_MIGRATION
+        track_transformations(
+            "CPAZIPCODE",
+            "",
+            "auditor_zip",
+            general_information["auditor_zip"],
+            "xform_replace_empty_zips",
+        )
+
+    if not general_information.get("auditee_zip"):
+        general_information["auditee_zip"] = settings.GSA_MIGRATION
+        track_transformations(
+            "ZIPCODE",
+            "",
+            "auditee_zip",
+            general_information["auditee_zip"],
+            "xform_replace_empty_zips",
+        )
+
+    return general_information
+
+
 def general_information(audit_header):
     """Generates general information JSON."""
     xform_update_multiple_eins_flag(audit_header)
@@ -458,6 +484,7 @@ def general_information(audit_header):
         xform_replace_empty_auditee_email,
         xform_replace_empty_or_invalid_auditor_ein_with_gsa_migration,
         xform_replace_empty_or_invalid_auditee_ein_with_gsa_migration,
+        xform_replace_empty_zips,
     ]
 
     for transform in transformations:

backend/census_historical_migration/test_federal_awards_xforms.py

@@ -25,6 +25,7 @@
     xform_missing_major_program,
     is_valid_extension,
     xform_cluster_names,
+    xform_sanitize_additional_award_identification,
 )
 
 
@@ -690,3 +691,23 @@ def test_replace_invalid_direct_award_flag(self):
         self.assertEqual(results[1], "N")
         # Expect third audit DIRECT flag to remain unchanged
         self.assertEqual(results[2], "Y")
+
+
+class TestXformSanitizeAdditionalAwardIdentification(SimpleTestCase):
+    class MockAudit:
+        def __init__(self, value):
+            self.AWARDIDENTIFICATION = value
+
+    def test_sanitize_valid(self):
+        audits = [self.MockAudit('=""12345"'), self.MockAudit("text")]
+        identifications = ['=""12345"', "text"]
+        expected = ["12345", "text"]
+        result = xform_sanitize_additional_award_identification(audits, identifications)
+        self.assertEqual(result, expected)
+
+    def test_empty_and_none_identifications(self):
+        audits = [self.MockAudit(""), self.MockAudit(None), self.MockAudit('=""12345"')]
+        identifications = ["", None, '=""12345"']
+        expected = ["", None, "12345"]
+        result = xform_sanitize_additional_award_identification(audits, identifications)
+        self.assertEqual(result, expected)

backend/census_historical_migration/test_general_information_xforms.py

@@ -19,6 +19,7 @@
     xform_replace_empty_or_invalid_auditee_uei_with_gsa_migration,
     xform_replace_empty_or_invalid_auditor_ein_with_gsa_migration,
     xform_replace_empty_or_invalid_auditee_ein_with_gsa_migration,
+    xform_replace_empty_zips,
 )
 from .exception_utils import (
     DataMigrationError,
@@ -431,3 +432,25 @@ def test_auditee_ein_empty_replaced(self):
             result = xform_replace_empty_or_invalid_auditee_ein_with_gsa_migration(info)
             mock_track.assert_called_once()
             self.assertEqual(result["ein"], settings.GSA_MIGRATION)
+
+
+class TestXformReplaceEmptyAuditorZip(SimpleTestCase):
+    def test_empty_auditor_zip(self):
+        """Test that an empty auditor_zip and auditee_zip are replaced with 'GSA_MIGRATION'"""
+        input_data = {
+            "auditee_zip": "",
+            "auditor_zip": "",
+        }
+        expected_output = {
+            "auditee_zip": settings.GSA_MIGRATION,
+            "auditor_zip": settings.GSA_MIGRATION,
+        }
+        self.assertEqual(xform_replace_empty_zips(input_data), expected_output)
+
+    def test_non_empty_auditor_zip(self):
+        """Test that a non-empty auditor_zip and auditee_zip remain unchanged"""
+        input_data = {
+            "auditee_zip": "10108",
+            "auditor_zip": "12345",
+        }
+        self.assertEqual(xform_replace_empty_zips(input_data), input_data)

backend/census_historical_migration/test_secondary_auditors_xforms.py

@@ -3,11 +3,12 @@
 
 from .workbooklib.secondary_auditors import (
     xform_address_state,
+    xform_address_zipcode,
     xform_cpafirmname,
 )
 
 
-class TestXformSecondaryAuditors(SimpleTestCase):
+class TestXformSecondaryAuditorsState(SimpleTestCase):
     class SecondaryAuditor:
         def __init__(self, state):
             self.CPASTATE = state
@@ -33,6 +34,32 @@ def test_missing_address_state(self):
         self.assertEqual(secondary_auditors[0].CPASTATE, settings.GSA_MIGRATION)
 
 
+class TestXformSecondaryAuditorsZipcode(SimpleTestCase):
+    class SecondaryAuditor:
+        def __init__(self, zipcode):
+            self.CPAZIPCODE = zipcode
+
+    def test_normal_address_zipcode(self):
+        # Setup specific test data
+        secondary_auditors = [
+            self.SecondaryAuditor("10108"),
+        ]
+
+        xform_address_zipcode(secondary_auditors)
+
+        self.assertEqual(secondary_auditors[0].CPAZIPCODE, "10108")
+
+    def test_missing_address_zipcode(self):
+        # Setup specific test data
+        secondary_auditors = [
+            self.SecondaryAuditor(""),
+        ]
+
+        xform_address_zipcode(secondary_auditors)
+
+        self.assertEqual(secondary_auditors[0].CPAZIPCODE, settings.GSA_MIGRATION)
+
+
 class TestXformCpaFirmName(SimpleTestCase):
     class MockSecondaryAuditorHeader:
         def __init__(

backend/census_historical_migration/transforms/xform_remove_hyphen_and_pad_zip.py

@@ -1,16 +1,22 @@
 from ..exception_utils import DataMigrationError
 from .xform_string_to_string import string_to_string
+from django.conf import settings
 
 
 def xform_remove_hyphen_and_pad_zip(zip):
     """
     Transform a ZIP code string by removing hyphen if present and adding a leading zero when necessary.
+    - If the ZIP is settings.GSA_MIGRATION, return it
     - If the ZIP code has hyphen, remove that character.
     - If the ZIP code has 4 or 8 digits, pads with a leading zero.
     - Returns the ZIP code if it has 5 digits or 9 digitis (after padding if needed).
     - Raises an error for other cases.
     """
     # Transformation to be documented.
+
+    if zip == settings.GSA_MIGRATION:
+        return zip
+
     strzip = string_to_string(zip)
     if "-" in strzip:
         parts = strzip.split("-")