Merge pull request #4011 from GSA-TTS/main

.github/workflows/trivy.yml

@@ -39,7 +39,7 @@ jobs:
         run: docker build -t ${{ env.DOCKER_NAME }}:${{ steps.date.outputs.date }} .
 
       - name: Run Trivy vulnerability scanner
-        uses: aquasecurity/trivy-action@0.22.0
+        uses: aquasecurity/trivy-action@0.23.0
         with:
           image-ref: '${{ env.DOCKER_NAME }}:${{ steps.date.outputs.date }}'
           scan-type: 'image'
@@ -74,7 +74,7 @@ jobs:
         run: docker pull ${{ matrix.image.name }}
 
       - name: Run Trivy vulnerability scanner on Third Party Images
-        uses: aquasecurity/trivy-action@0.22.0
+        uses: aquasecurity/trivy-action@0.23.0
         with:
           image-ref: '${{ matrix.image.name }}'
           scan-type: 'image'

backend/audit/validators.py

@@ -268,6 +268,13 @@ def validate_use_of_gsa_migration_keyword_in_audit_info(
     if not is_data_migration and settings.GSA_MIGRATION in [
         audit_information.get("is_sp_framework_required", ""),
         audit_information.get("is_low_risk_auditee", ""),
+        audit_information.get("is_going_concern_included", ""),
+        audit_information.get("is_internal_control_deficiency_disclosed", ""),
+        audit_information.get("is_internal_control_material_weakness_disclosed", ""),
+        audit_information.get("is_material_noncompliance_disclosed", ""),
+        audit_information.get("is_aicpa_audit_guide_included", ""),
+        ",".join(audit_information.get("agencies", [])),
+        ",".join(audit_information.get("gaap_results", [])),
     ]:
         raise ValidationError(
             _(f"{settings.GSA_MIGRATION} not permitted outside of migrations"),

backend/census_historical_migration/end_to_end_core.py

@@ -35,6 +35,7 @@
 from census_historical_migration.migration_result import MigrationResult
 from .change_record import InspectionRecord
 from .invalid_record import InvalidRecord
+from .report_type_flag import AceFlag
 
 from django.core.exceptions import ValidationError
 from django.utils import timezone as django_timezone
@@ -114,6 +115,7 @@ def run_end_to_end(user, audit_header):
     """Attempts to migrate the given audit"""
     InspectionRecord.reset()
     InvalidRecord.reset()
+    AceFlag.reset()
     try:
         sac = setup_sac(user, audit_header)
         builder_loader = workbook_builder_loader(user, sac, audit_header)
@@ -236,6 +238,7 @@ def track_invalid_audit_records(audit_year, dbkey, report_id):
     # Save the invalid audit record and reset InvalidRecord
     invalid_audit_record.save()
     InvalidRecord.reset()
+    AceFlag.reset()
 
 
 def record_migration_status(audit_year, dbkey):

backend/census_historical_migration/invalid_migration_tags.py

@@ -11,3 +11,5 @@ class INVALID_MIGRATION_TAGS:
         "extra_finding_reference_numbers_in_findingstext"
     )
     INCORRECT_FINDINGS_COUNT = "incorrect_findings_count"
+
+    ACE_AUDIT_REPORT = "ace_audit_report"

backend/census_historical_migration/report_type_flag.py

@@ -0,0 +1,23 @@
+import copy
+from typing import Any
+
+
+class AceFlag:
+    """Hold ACE report flag for the ongoing report migration"""
+
+    DEFAULT: dict[str, Any] = {
+        "is_ace_report": False,
+    }
+    fields = copy.deepcopy(DEFAULT)
+
+    @staticmethod
+    def reset():
+        AceFlag.fields = copy.deepcopy(AceFlag.DEFAULT)
+
+    @staticmethod
+    def set_ace_report_flag(flag):
+        AceFlag.fields["is_ace_report"] = flag
+
+    @staticmethod
+    def get_ace_report_flag():
+        return AceFlag.fields["is_ace_report"]

backend/census_historical_migration/sac_general_lib/audit_information.py

@@ -1,11 +1,15 @@
 import re
 
 from django.conf import settings
+
+from census_historical_migration.invalid_migration_tags import INVALID_MIGRATION_TAGS
+from census_historical_migration.invalid_record import InvalidRecord
+from census_historical_migration.report_type_flag import AceFlag
 from ..transforms.xform_string_to_int import string_to_int
 from ..transforms.xform_string_to_bool import string_to_bool
 from ..transforms.xform_string_to_string import string_to_string
 from ..exception_utils import DataMigrationError
-from ..workbooklib.excel_creation_utils import get_audits
+from ..workbooklib.excel_creation_utils import get_audits, track_invalid_records
 from ..base_field_maps import FormFieldMap, FormFieldInDissem
 from ..sac_general_lib.utils import (
     create_json_from_db_object,
@@ -328,17 +332,63 @@ def xform_lowrisk(audit_header):
 
 def audit_information(audit_header):
     """Generates audit information JSON."""
-    xform_sp_framework_required(audit_header)
-    xform_lowrisk(audit_header)
-    results = xform_build_sp_framework_gaap_results(audit_header)
+    if AceFlag.get_ace_report_flag():
+        audit_info = ace_audit_information(audit_header)
+    else:
+        xform_sp_framework_required(audit_header)
+        xform_lowrisk(audit_header)
+        results = xform_build_sp_framework_gaap_results(audit_header)
+        audit_info = create_json_from_db_object(audit_header, mappings)
+        audit_info = {
+            key: results.get(key, audit_info.get(key))
+            for key in set(audit_info) | set(results)
+        }
+
     agencies_prefixes = _get_agency_prefixes(audit_header.DBKEY, audit_header.AUDITYEAR)
-    audit_info = create_json_from_db_object(audit_header, mappings)
-    audit_info = {
-        key: results.get(key, audit_info.get(key))
-        for key in set(audit_info) | set(results)
-    }
     audit_info["agencies"] = list(agencies_prefixes)
 
     audit.validators.validate_audit_information_json(audit_info)
 
     return audit_info
+
+
+def ace_audit_information(audit_header):
+    """Constructs the audit information JSON object for an ACE report."""
+    actual = create_json_from_db_object(audit_header, mappings)
+    default = {
+        "dollar_threshold": settings.GSA_MIGRATION_INT,
+        "gaap_results": [settings.GSA_MIGRATION],
+        "is_going_concern_included": settings.GSA_MIGRATION,
+        "is_internal_control_deficiency_disclosed": settings.GSA_MIGRATION,
+        "is_internal_control_material_weakness_disclosed": settings.GSA_MIGRATION,
+        "is_material_noncompliance_disclosed": settings.GSA_MIGRATION,
+        "is_aicpa_audit_guide_included": settings.GSA_MIGRATION,
+        "is_low_risk_auditee": settings.GSA_MIGRATION,
+        "agencies": [settings.GSA_MIGRATION],
+    }
+    if actual:
+        for key, value in actual.items():
+            if value:
+                default[key] = value
+    # Tracking invalid records
+    invalid_records = []
+    for mapping in mappings:
+        in_dissem = (
+            mapping.in_dissem
+            if mapping.in_dissem != FormFieldInDissem
+            else mapping.in_form
+        )
+        track_invalid_records(
+            [
+                (mapping.in_db, getattr(audit_header, mapping.in_db)),
+            ],
+            in_dissem,
+            default[mapping.in_form],
+            invalid_records,
+        )
+    if invalid_records:
+        InvalidRecord.append_invalid_migration_tag(
+            INVALID_MIGRATION_TAGS.ACE_AUDIT_REPORT,
+        )
+
+    return default

backend/census_historical_migration/sac_general_lib/general_information.py

@@ -21,7 +21,7 @@
 from jsonschema import validate
 from jsonschema.exceptions import ValidationError
 from ..change_record import InspectionRecord, CensusRecord, GsaFacRecord
-
+from ..report_type_flag import AceFlag
 
 PERIOD_DICT = {"A": "annual", "B": "biennial", "O": "other"}
 AUDIT_TYPE_DICT = {
@@ -351,10 +351,8 @@ def xform_audit_type(general_information):
         value_in_db = general_information["audit_type"]
         audit_type = _census_audit_type(value_in_db.upper())
         if audit_type == AUDIT_TYPE_DICT["A"]:
-            raise DataMigrationError(
-                "Skipping ACE audit",
-                "skip_ace_audit",
-            )
+            audit_type = settings.GSA_MIGRATION
+            AceFlag.set_ace_report_flag(True)
         general_information["audit_type"] = audit_type
         track_transformations(
             "AUDITTYPE",

backend/census_historical_migration/sac_general_lib/utils.py

@@ -2,6 +2,8 @@
 from datetime import datetime, timezone
 import sys
 
+from census_historical_migration.report_type_flag import AceFlag
+
 from ..transforms.xform_string_to_string import (
     string_to_string,
 )
@@ -22,7 +24,7 @@ def create_json_from_db_object(gobj, mappings):
             value = mapping.default
         # Fields with a value of None are skipped to maintain consistency with the logic
         # used in workbook data extraction and to prevent converting None into the string "None".
-        if value is None:
+        if value is None or (value == "" and AceFlag.get_ace_report_flag()):
             continue
         # Check for the type and apply the correct conversion method
         if mapping.type is str:

backend/census_historical_migration/test_audit_information_xforms.py

@@ -1,6 +1,8 @@
+from unittest.mock import MagicMock, patch
 from django.test import SimpleTestCase
 
 from .sac_general_lib.audit_information import (
+    ace_audit_information,
     xform_build_sp_framework_gaap_results,
     xform_framework_basis,
     xform_sp_framework_required,
@@ -205,3 +207,76 @@ def test_lowrisk_blank(self):
         audit_header.LOWRISK = ""
         xform_lowrisk(audit_header)
         self.assertEqual(audit_header.LOWRISK, settings.GSA_MIGRATION)
+
+
+class TestAceAuditInformation(SimpleTestCase):
+
+    def setUp(self):
+        self.audit_header = MagicMock()
+        self.audit_header.some_field = "test_value"
+
+        self.default_values = {
+            "dollar_threshold": settings.GSA_MIGRATION_INT,
+            "gaap_results": [settings.GSA_MIGRATION],
+            "is_going_concern_included": settings.GSA_MIGRATION,
+            "is_internal_control_deficiency_disclosed": settings.GSA_MIGRATION,
+            "is_internal_control_material_weakness_disclosed": settings.GSA_MIGRATION,
+            "is_material_noncompliance_disclosed": settings.GSA_MIGRATION,
+            "is_aicpa_audit_guide_included": settings.GSA_MIGRATION,
+            "is_low_risk_auditee": settings.GSA_MIGRATION,
+            "agencies": [settings.GSA_MIGRATION],
+        }
+
+    @patch(
+        "census_historical_migration.sac_general_lib.audit_information.create_json_from_db_object"
+    )
+    def test_ace_audit_information_with_actual_values(self, mock_create_json):
+        """Test that the function returns the correct values when all fields are present."""
+        mock_create_json.return_value = {
+            "dollar_threshold": 1000,
+            "is_low_risk_auditee": True,
+            "new_field": "new_value",
+        }
+
+        result = ace_audit_information(self.audit_header)
+
+        expected_result = self.default_values.copy()
+        expected_result.update(
+            {
+                "dollar_threshold": 1000,
+                "is_low_risk_auditee": True,
+                "new_field": "new_value",
+            }
+        )
+
+        self.assertEqual(result, expected_result)
+
+    @patch(
+        "census_historical_migration.sac_general_lib.audit_information.create_json_from_db_object"
+    )
+    def test_ace_audit_information_with_default_values(self, mock_create_json):
+        """Test that the function returns the correct values when all fields are missing."""
+        mock_create_json.return_value = {}
+
+        result = ace_audit_information(self.audit_header)
+
+        expected_result = self.default_values
+
+        self.assertEqual(result, expected_result)
+
+    @patch(
+        "census_historical_migration.sac_general_lib.audit_information.create_json_from_db_object"
+    )
+    def test_ace_audit_information_with_mixed_values(self, mock_create_json):
+        """Test that the function returns the correct values when some fields are missing."""
+        mock_create_json.return_value = {
+            "dollar_threshold": None,
+            "is_low_risk_auditee": True,
+        }
+
+        result = ace_audit_information(self.audit_header)
+
+        expected_result = self.default_values.copy()
+        expected_result.update({"is_low_risk_auditee": True})
+
+        self.assertEqual(result, expected_result)