Addressing SonarQube issues and altering logic

Author: adrianoaru-nhs

classes/subject.py

@@ -1251,93 +1251,105 @@ def from_dataframe_row(row: pd.Series) -> "Subject":
         Only fields present in the SQL query are populated.
         """
 
-        def parse_date(
-            val: Optional[Union[pd.Timestamp, str, datetime, date]],
-        ) -> Optional[date]:
-            """
-            Converts a value to a Python date object if possible.
-
-            Args:
-                val: The value to convert (can be pandas.Timestamp, string, datetime, date, or None).
-
-            Returns:
-                Optional[date]: The converted date object, or None if conversion fails.
-            """
-            if pd.isnull(val):
-                return None
-            if isinstance(val, pd.Timestamp):
-                return val.to_pydatetime().date()
-            if isinstance(val, str):
-                try:
-                    return datetime.strptime(val[:10], "%Y-%m-%d").date()
-                except Exception:
-                    return None
-            if isinstance(val, datetime):
-                return val.date()
-            if isinstance(val, date):
-                return val
-            return None
-
-        def parse_datetime(
-            val: Optional[Union[pd.Timestamp, str, datetime, date]],
-        ) -> Optional[datetime]:
-            """
-            Converts a value to a Python datetime object if possible.
-
-            Args:
-                val: The value to convert (can be pandas.Timestamp, string, datetime, or None).
-
-            Returns:
-                Optional[datetime]: The converted datetime object, or None if conversion fails.
-            """
-            if pd.isnull(val):
-                return None
-            if isinstance(val, pd.Timestamp):
-                return val.to_pydatetime()
-            if isinstance(val, str):
-                for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
-                    try:
-                        return datetime.strptime(val[:19], fmt)
-                    except Exception:
-                        continue
-                return None
-            if isinstance(val, datetime):
-                return val
-            return None
-
         field_map = {
             "screening_subject_id": row.get("screening_subject_id"),
             "nhs_number": row.get("subject_nhs_number"),
             "surname": row.get("person_family_name"),
             "forename": row.get("person_given_name"),
-            "datestamp": parse_datetime(row.get("datestamp")),
+            "datestamp": Subject.parse_datetime(row.get("datestamp")),
             "screening_status_id": row.get("screening_status_id"),
             "screening_status_change_reason_id": row.get("ss_reason_for_change_id"),
-            "screening_status_change_date": parse_date(
+            "screening_status_change_date": Subject.parse_date(
                 row.get("screening_status_change_date")
             ),
-            "screening_due_date": parse_date(row.get("screening_due_date")),
+            "screening_due_date": Subject.parse_date(row.get("screening_due_date")),
             "screening_due_date_change_reason_id": row.get("sdd_reason_for_change_id"),
-            "screening_due_date_change_date": parse_date(row.get("sdd_change_date")),
-            "calculated_screening_due_date": parse_date(row.get("calculated_sdd")),
-            "surveillance_screening_due_date": parse_date(
+            "screening_due_date_change_date": Subject.parse_date(
+                row.get("sdd_change_date")
+            ),
+            "calculated_screening_due_date": Subject.parse_date(
+                row.get("calculated_sdd")
+            ),
+            "surveillance_screening_due_date": Subject.parse_date(
                 row.get("surveillance_screen_due_date")
             ),
-            "calculated_surveillance_due_date": parse_date(row.get("calculated_ssdd")),
+            "calculated_surveillance_due_date": Subject.parse_date(
+                row.get("calculated_ssdd")
+            ),
             "surveillance_due_date_change_reason_id": row.get(
                 "surveillance_sdd_rsn_change_id"
             ),
-            "surveillance_due_date_change_date": parse_date(
+            "surveillance_due_date_change_date": Subject.parse_date(
                 row.get("surveillance_sdd_change_date")
             ),
-            "lynch_due_date": parse_date(row.get("lynch_screening_due_date")),
+            "lynch_due_date": Subject.parse_date(row.get("lynch_screening_due_date")),
             "lynch_due_date_change_reason_id": row.get(
                 "lynch_sdd_reason_for_change_id"
             ),
-            "lynch_due_date_change_date": parse_date(row.get("lynch_sdd_change_date")),
-            "calculated_lynch_due_date": parse_date(row.get("lynch_calculated_sdd")),
-            "date_of_birth": parse_date(row.get("date_of_birth")),
-            "date_of_death": parse_date(row.get("date_of_death")),
+            "lynch_due_date_change_date": Subject.parse_date(
+                row.get("lynch_sdd_change_date")
+            ),
+            "calculated_lynch_due_date": Subject.parse_date(
+                row.get("lynch_calculated_sdd")
+            ),
+            "date_of_birth": Subject.parse_date(row.get("date_of_birth")),
+            "date_of_death": Subject.parse_date(row.get("date_of_death")),
         }
 
         return Subject(**field_map)
+
+    @staticmethod
+    def parse_date(
+        val: Optional[Union[pd.Timestamp, str, datetime, date]],
+    ) -> Optional[date]:
+        """
+        Converts a value to a Python date object if possible.
+
+        Args:
+            val: The value to convert (can be pandas.Timestamp, string, datetime, date, or None).
+
+        Returns:
+            Optional[date]: The converted date object, or None if conversion fails.
+        """
+        if pd.isnull(val):
+            return None
+        if isinstance(val, pd.Timestamp):
+            return val.to_pydatetime().date()
+        if isinstance(val, str):
+            try:
+                return datetime.strptime(val[:10], "%Y-%m-%d").date()
+            except Exception:
+                return None
+        if isinstance(val, datetime):
+            return val.date()
+        if isinstance(val, date):
+            return val
+        return None
+
+    @staticmethod
+    def parse_datetime(
+        val: Optional[Union[pd.Timestamp, str, datetime, date]],
+    ) -> Optional[datetime]:
+        """
+        Converts a value to a Python datetime object if possible.
+
+        Args:
+            val: The value to convert (can be pandas.Timestamp, string, datetime, or None).
+
+        Returns:
+            Optional[datetime]: The converted datetime object, or None if conversion fails.
+        """
+        if pd.isnull(val):
+            return None
+        if isinstance(val, pd.Timestamp):
+            return val.to_pydatetime()
+        if isinstance(val, str):
+            for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
+                try:
+                    return datetime.strptime(val[:19], fmt)
+                except Exception:
+                    continue
+            return None
+        if isinstance(val, datetime):
+            return val
+        return None

docs/utility-guides/SubjectAssertion.md

@@ -22,7 +22,7 @@ It is designed to assert that a subject with a given NHS number matches specifie
 ## Overview
 
 The `subject_assertion` function is used to verify that a subject in the database matches a set of criteria.
-If the subject does not match all criteria, the function will iteratively remove criteria (except NHS number) and retry, logging any criteria that caused the assertion to fail.
+If the subject does not match all criteria, the function will iteratively loop through each criteria (except NHS number), logging any criteria that caused the assertion to fail.
 
 ---
 
@@ -36,8 +36,8 @@ If the subject does not match all criteria, the function will iteratively remove
 ## How It Works
 
 1. The function first checks if the subject with the given NHS number matches all provided criteria.
-2. If not, it removes one criterion at a time (except NHS number) and retries the assertion.
-3. This process continues until either a match is found or all criteria (except NHS number) have been removed.
+2. If not, it removes checks one criterion at a time and retries the assertion.
+3. This process continues until all criteria have been checked.
 4. If a match is found only after removing criteria, the failed criteria are logged.
 5. The function returns `True` only if all criteria match on the first attempt; otherwise, it returns `False`.
 

tests_utils/test_subject_assertion_util.py

@@ -3,20 +3,27 @@
 
 pytestmark = [pytest.mark.utils_local]
 
+nhs_number = "9233639266"
+
 
 def test_subject_assertion_true():
-    nhs_number = "9233639266"
     criteria = {"screening status": "Inactive", "subject age": "> 28"}
     assert subject_assertion(nhs_number, criteria) is True
 
 
 def test_subject_assertion_false():
-    nhs_number = "9233639266"
     criteria = {"screening status": "Call", "subject age": "< 28"}
     assert subject_assertion(nhs_number, criteria) is False
 
 
 def test_subject_assertion_false_with_some_true():
-    nhs_number = "9233639266"
-    criteria = {"screening status": "Inactive", "subject age": "< 28"}
+    criteria = {
+        "screening status": "Inactive",
+        "subject age": "> 28",
+        "latest episode type": "FOBT",
+        "latest episode status": "Open",
+        "latest episode has referral date": "Past",
+        "latest episode has diagnosis date": "No",
+        "latest episode diagnosis date reason": "NULL",
+    }
     assert subject_assertion(nhs_number, criteria) is False

utils/subject_assertion.py

@@ -14,7 +14,9 @@ def subject_assertion(nhs_number: str, criteria: dict) -> bool:
     Returns:
         bool: True if the subject matches the provided criteria, False if it does not.
     """
-    nhs_no_criteria = {"nhs number": nhs_number}
+    nhs_number_string = "nhs number"
+    subject_nhs_number_string = "subject_nhs_number"
+    nhs_no_criteria = {nhs_number_string: nhs_number}
     subject = Subject()
     user = User()
     builder = SubjectSelectionQueryBuilder()
@@ -29,40 +31,45 @@ def subject_assertion(nhs_number: str, criteria: dict) -> bool:
     subject_df = OracleDB().execute_query(query, bind_vars)
     subject = Subject.from_dataframe_row(subject_df.iloc[0])
 
-    criteria["nhs number"] = nhs_number
+    criteria[nhs_number_string] = nhs_number
+
+    # Check all criteria together first
     query, bind_vars = builder.build_subject_selection_query(
         criteria=criteria,
         user=user,
         subject=subject,
         subjects_to_retrieve=1,
     )
-
     df = OracleDB().execute_query(query, bind_vars)
-
-    if nhs_number in df["subject_nhs_number"].values:
+    if nhs_number in df[subject_nhs_number_string].values:
         return True
 
-    # Try removing criteria one by one (except nhs number)
+    # Check each criterion independently
     failed_criteria = []
-    criteria_keys = [key for key in criteria if key != "nhs number"]
+    criteria_keys = [key for key in criteria if key != nhs_number_string]
     for key in criteria_keys:
-        reduced_criteria = {key: value for key, value in criteria.items() if key != key}
+        single_criteria = {nhs_number_string: nhs_number, key: criteria[key]}
         query, bind_vars = builder.build_subject_selection_query(
-            criteria=reduced_criteria,
+            criteria=single_criteria,
             user=user,
             subject=subject,
             subjects_to_retrieve=1,
         )
         df = OracleDB().execute_query(query, bind_vars)
-        if nhs_number in df["subject_nhs_number"].values:
-            failed_criteria.append((key, criteria[key]))
-            break
-        else:
+        if (
+            subject_nhs_number_string not in df.columns
+            or nhs_number not in df[subject_nhs_number_string].values
+        ):
             failed_criteria.append((key, criteria[key]))
 
     if failed_criteria:
         log_message = "Subject Assertion Failed\nFailed criteria:\n" + "\n".join(
-            [f"Key: '{key}' - Value: '{value}'" for key, value in failed_criteria]
+            [f"{key}, {value}" for key, value in failed_criteria]
         )
         logging.error(log_message)
+    else:
+        logging.error(
+            "Subject Assertion Failed: Criteria combination is invalid or conflicting."
+        )
+
     return False