csv use empty string instead of nulls (#1766)

Author: gerrycampion

cdisc_rules_engine/services/reporting/base_report_data.py

@@ -39,17 +39,19 @@ def __init__(
         )
 
     @staticmethod
-    def process_values(values: list[str]) -> list[str]:
-        if not values or values is None:
-            return ["null"]
+    def process_values(
+        values: list[str | None], null_placeholder: str = "null"
+    ) -> list[str]:
+        if not values:
+            return [null_placeholder]
         processed_values = []
         for value in values:
             if value is None:
-                processed_values.append("null")
+                processed_values.append(null_placeholder)
                 continue
             value = value.strip()
             if value == "" or value.lower() == "nan":
-                processed_values.append("null")
+                processed_values.append(null_placeholder)
             else:
                 processed_values.append(value)
         return processed_values

cdisc_rules_engine/services/reporting/sdtm_report_data.py

@@ -356,7 +356,8 @@ def get_csv_rows(self) -> tuple[list[str], list[list[str]]]:
             variables = issue.get("variables") or []
             values = issue.get("values") or []
             for variable, value in zip(variables, values):
-                rows.append([dataset, record, variable, str(value)])
+                csv_value = "" if value in (None, "null") else value
+                rows.append([dataset, record, variable, csv_value])
         return header, rows
 
     def get_rules_report_data(self) -> list[dict]:

cdisc_rules_engine/services/reporting/usdm_report_data.py

@@ -253,7 +253,8 @@ def get_csv_rows(self) -> tuple[list[str], list[list[str]]]:
             attributes = issue.get("attributes") or []
             values = issue.get("values") or []
             for attribute, value in zip(attributes, values):
-                rows.append([path, attribute, str(value)])
+                csv_value = "" if value in (None, "null") else value
+                rows.append([path, attribute, csv_value])
         return header, rows
 
     def get_rules_report_data(self) -> list[dict]:

tests/unit/test_services/test_reporting/test_sdtm_report.py

@@ -187,6 +187,38 @@ def test_get_csv_rows_empty_results():
     assert rows == []
 
 
+def test_get_csv_rows_preserves_blank_values_for_none_and_empty_string(
+    mock_validation_results,
+):
+    mock_validation_results[0].results[0]["errors"][0]["value"]["AESTDY"] = None
+    mock_validation_results[0].results[0]["errors"][0]["value"]["DOMAIN"] = ""
+    report = SDTMReportData(
+        [],
+        ["test"],
+        mock_validation_results,
+        10.1,
+        MagicMock(define_xml_path=None, max_errors_per_rule=(None, False)),
+    )
+
+    _, rows = report.get_csv_rows()
+    assert [
+        row[3] for row in rows if row[1] == "1" and row[2] in {"AESTDY", "DOMAIN"}
+    ] == [
+        "",
+        "",
+    ]
+
+    details = report.get_detailed_data()
+    detail_row = next(
+        row
+        for row in details
+        if row["core_id"] == "CORE1"
+        and row["row"] == 1
+        and row["variables"] == ["AESTDY", "DOMAIN"]
+    )
+    assert detail_row["values"] == ["null", "null"]
+
+
 def test_no_errors_when_none_value_in_one_of_the_records(mock_validation_results):
     # forcing None and str comparison in summary and details
     mock_validation_results[0].id = None

tests/unit/test_services/test_reporting/test_usdm_report.py

@@ -184,6 +184,32 @@ def test_get_csv_rows_empty_results():
     assert rows == []
 
 
+def test_get_csv_rows_preserves_blank_values_for_none_and_empty_string(
+    mock_validation_results,
+):
+    mock_validation_results[0].results[0]["errors"][0]["value"]["AESTDY"] = None
+    mock_validation_results[0].results[0]["errors"][0]["value"]["DOMAIN"] = ""
+    report = USDMReportData(
+        [],
+        ["test"],
+        mock_validation_results,
+        10.1,
+        MagicMock(define_xml_path=None, max_errors_per_rule=(None, False)),
+    )
+
+    _, rows = report.get_csv_rows()
+    assert any(row[1] == "AESTDY" and row[2] == "" for row in rows)
+    assert any(row[1] == "DOMAIN" and row[2] == "" for row in rows)
+
+    details = report.get_detailed_data()
+    detail_row = next(
+        row
+        for row in details
+        if row["core_id"] == "CORE1" and row["attributes"] == ["AESTDY", "DOMAIN"]
+    )
+    assert detail_row["values"] == ["null", "null"]
+
+
 def test_no_errors_when_none_value_in_one_of_the_records(mock_validation_results):
     # forcing None and str comparison in summary and details
     mock_validation_results[0].id = None