Refactoring Code and adding queries

Author: adrianoaru-nhs

pages/contacts_list/maintain_contacts_page.py

@@ -50,22 +50,18 @@ def click_search_button(self) -> None:
         """Click the search button to perform a search"""
         self.click(self.search_button)
 
-    def click_person_link_from_suranme(self, surname: str) -> None:
+    def click_person_link_from_surname(self, surname: str) -> None:
         """
         Clicks on the link containing the person's surname to go to the edit contact page
         Args:
             surname (str): The surname of the subject
         """
-        self.click(self.page.get_by_role("link", name=surname))
+        self.click(self.page.get_by_role("link", name=surname).last)
 
     def click_person_link_from_forename(self, forename: str) -> None:
         """
         Clicks on the link containing the person's surname to go to the edit contact page
         Args:
             forename (str): The forename of the subject
         """
-        self.click(self.page.get_by_role("link", name=forename))
-
-    def click_first_correct_link(self) -> None:
-        """Clicks on the first link in the search results that is a BCSS user and has a code"""
-        self.search_table.click_surname_if_bcss_user_and_has_code()
+        self.click(self.page.get_by_role("link", name=forename).last)

tests/regression/regression_tests/test_regression_test_setup_steps.py

@@ -15,10 +15,10 @@
 from utils.oracle.oracle_specific_functions import (
     set_org_parameter_value,
     check_parameter,
+    build_accredited_screening_colonoscopist_query,
+    get_accredited_screening_colonoscopist_in_bcs001,
 )
 
-from utils.oracle.oracle import OracleDB
-
 
 def test_allow_10_minute_colonsocopy_assessment_appointments(page: Page) -> None:
     """
@@ -50,36 +50,31 @@ def test_ensure_the_is_accredited_screening_colonoscopist_with_current_resect_an
     Scenario: 2: Ensure there is an Accredited Screening Colonoscopist with current Resect & Discard accreditation
     """
     UserTools.user_login(page, "BCSS Bureau Staff at X26")
+
+    current_df = build_accredited_screening_colonoscopist_query("Current")
+    if current_df.empty:
+        pytest.skip(
+            "No Accredited Screening Colonoscopist with current Resect & Discard accreditation found."
+        )
+    expired_df = build_accredited_screening_colonoscopist_query("Expiring soon")
+    if expired_df.empty:
+        pytest.skip(
+            "No Accredited Screening Colonoscopist with expiring Resect & Discard accreditation found."
+        )
+    person_df = get_accredited_screening_colonoscopist_in_bcs001()
+    if person_df.empty:
+        pytest.fail("No Accredited Screening Colonoscopist found in the database.")
+
     BasePage(page).go_to_contacts_list_page()
     ContactsListPage(page).go_to_maintain_contacts_page()
-    query = """
-    SELECT
-        prs.prs_id,
-        prs.person_family_name,
-        prs.person_given_name,
-        prs.gmc_code,
-        pio.pio_id,
-        pio.role_id,
-        org.org_code,
-        org.org_name
-    FROM person prs
-    INNER JOIN person_in_org pio ON prs.prs_id = pio.prs_id
-    INNER JOIN org ON org.org_id = pio.org_id
-    WHERE 1=1
-    AND pio.role_id = (SELECT valid_value_id FROM valid_values WHERE description = 'Accredited Screening Colonoscopist')
-    AND org.org_code = 'BCS001'
-    AND TRUNC(SYSDATE) BETWEEN TRUNC(pio.start_date) AND NVL(pio.end_date, SYSDATE)
-    AND pio.is_bcss_user = 1
-    ORDER BY prs.prs_id
-    FETCH FIRST 1 ROWS ONLY
-    """
-    person_df = OracleDB().execute_query(query)
+
     surname = person_df.iloc[0]["person_family_name"]
     forename = person_df.iloc[0]["person_given_name"]
     MaintainContactsPage(page).fill_surname_input_field(surname)
     MaintainContactsPage(page).fill_forenames_input_field(forename)
     MaintainContactsPage(page).click_search_button()
-    MaintainContactsPage(page).click_first_correct_link()
+    MaintainContactsPage(page).click_person_link_from_surname(surname)
+
     EditContactPage(page).click_view_resect_and_discard_link()
     ResectAndDiscardAccreditationHistoryPage(page).verify_heading_is_correct()
     ResectAndDiscardAccreditationHistoryPage(
@@ -89,6 +84,7 @@ def test_ensure_the_is_accredited_screening_colonoscopist_with_current_resect_an
     ResectAndDiscardAccreditationHistoryPage(
         page
     ).add_new_period_of_resect_and_discard_accerditation(date=yesterday)
+
     BasePage(page).click_back_button()
     end_date = yesterday + relativedelta(years=2)
     EditContactPage(page).assert_value_for_label_in_edit_contact_table(

utils/oracle/oracle_specific_functions.py

@@ -737,6 +737,112 @@ def get_investigation_dataset_polyp_algorithm_size(
     return polyp_algorithm_size
 
 
+def build_accredited_screening_colonoscopist_query(query_type: str) -> pd.DataFrame:
+    """
+    Builds a SQL query for either 'Current' or 'Expiring soon' accreditations.
+
+    Args:
+        query_type (str): Either "Current" or "Expiring soon".
+
+    Returns:
+        str: The SQL query string.
+
+    Raises:
+        ValueError: If query_type is invalid.
+    """
+    base_query = """
+    SELECT
+        prs.prs_id,
+        prs.person_family_name,
+        prs.person_given_name,
+        prs.gmc_code,
+        pio305730.pio_id
+    FROM person prs
+    LEFT OUTER JOIN person_in_org pio305730 ON pio305730.prs_id = prs.prs_id
+        AND NOT EXISTS (
+            SELECT 1
+            FROM person_in_org pio305730_2
+            WHERE pio305730.prs_id = pio305730_2.prs_id
+            AND pio305730.role_id = pio305730_2.role_id
+            AND pio305730.pio_id != pio305730_2.pio_id
+        )
+        AND pio305730.role_id = 305730
+        AND TRUNC(SYSDATE) BETWEEN pio305730.start_date AND NVL(pio305730.end_date, TO_DATE('31/12/3999','DD/MM/YYYY'))
+        AND pio305730.org_id IN (
+            SELECT oio305730.org_id
+            FROM org oio305730
+            WHERE oio305730.org_code = 'BCS001'
+        )
+    INNER JOIN person_accreditation pa ON pa.prs_id = prs.prs_id
+        AND pa.accreditation_id = (
+            SELECT pax.accreditation_id
+            FROM person_accreditation pax
+            WHERE pax.prs_id = prs.prs_id
+            AND pax.accreditation_type_id = 305730
+            ORDER BY pax.start_date DESC
+            FETCH FIRST 1 ROW ONLY
+        )
+        AND pa.accreditation_type_id = 305730
+        AND pa.start_date <= TRUNC(SYSDATE)
+    """
+
+    # condition for accreditation end date
+    if query_type == "Expiring soon":
+        condition = """    AND pa.end_date > TRUNC(SYSDATE)
+    AND pa.end_date <= ADD_MONTHS(TRUNC(SYSDATE), 5)"""
+    elif query_type == "Current":
+        condition = """    AND pa.end_date > ADD_MONTHS(TRUNC(SYSDATE), 5)"""
+    else:
+        raise ValueError("query_type must be either 'Current' or 'Expiring soon'")
+
+    full_query = f"""{base_query}
+        {condition}
+        WHERE 1=1
+        AND pio305730.pio_id IS NOT NULL
+        FETCH FIRST 1 ROWS ONLY"""
+
+    df = OracleDB().execute_query(full_query)
+    return df
+
+
+def get_accredited_screening_colonoscopist_in_bcs001() -> pd.DataFrame:
+    query = """
+        SELECT
+            prs.prs_id,
+            prs.person_family_name,
+            prs.person_given_name,
+            prs.gmc_code,
+            pio305730.pio_id
+        FROM person prs
+        LEFT OUTER JOIN person_in_org pio305730 ON pio305730.prs_id = prs.prs_id
+            AND NOT EXISTS (
+                SELECT 1
+                FROM person_in_org pio305730_2
+                WHERE pio305730.prs_id = pio305730_2.prs_id
+                AND pio305730.role_id = pio305730_2.role_id
+                AND pio305730.pio_id != pio305730_2.pio_id
+            )
+            AND pio305730.role_id = 305730
+            AND TRUNC(SYSDATE) BETWEEN pio305730.start_date AND NVL(pio305730.end_date, TO_DATE('31/12/3999','DD/MM/YYYY'))
+            AND pio305730.org_id IN (
+                SELECT oio305730.org_id
+                FROM org oio305730
+                WHERE oio305730.org_code = 'BCS001'
+            )
+        INNER JOIN person_accreditation pa ON pa.prs_id = prs.prs_id
+            AND pa.accreditation_id = (
+                SELECT pax.accreditation_id
+                FROM person_accreditation pax
+                WHERE pax.prs_id = prs.prs_id
+                AND pax.accreditation_type_id = 305730
+                ORDER BY pax.start_date DESC
+                FETCH FIRST 1 ROW ONLY
+            )
+    """
+    df = OracleDB().execute_query(query)
+    return df
+
+
 class SubjectSelector:
     """
     Provides helper methods for selecting screening subjects based on preconditions

utils/table_util.py

@@ -365,51 +365,6 @@ def get_row_where(self, criteria: dict[str, str]) -> Locator | None:
                 return self.pick_row(i)
         return None
 
-    def click_surname_if_bcss_user_and_has_code(self):
-        """
-        Clicks on the surname link of the first row where:
-        - 'BCSS User?' column is 'Yes'
-        - 'User Code' is not '-'
-        """
-        surname_col = self.get_column_index("Surname")
-        user_code_col = self.get_column_index("User Code")
-        bcss_col = self.get_column_index("BCSS User?")
-
-        if -1 in (surname_col, user_code_col, bcss_col):
-            raise ValueError("One or more required columns not found")
-
-        rows = self.table.locator(self.tbody_tr_string)
-
-        for i in range(rows.count()):
-            row = rows.nth(i)
-
-            # Skip rows with no <td> cells
-            cell_count = row.locator("td").count()
-            if cell_count == 0:
-                continue
-
-            # Skip rows that don’t have enough columns
-            if cell_count < max(user_code_col, bcss_col, surname_col):
-                continue
-
-            user_code = (
-                row.locator(f"td:nth-child({user_code_col})").inner_text().strip()
-            )
-            bcss_value = row.locator(f"td:nth-child({bcss_col})").inner_text().strip()
-
-            if bcss_value.lower() == "yes" and user_code != "-":
-                surname_link = row.locator(f"td:nth-child({surname_col}) a")
-                if surname_link.count() > 0:
-                    surname_link.click()
-                    return
-                else:
-                    logging.error(f"No clickable surname link found in row {i+1}")
-                    return
-
-        logging.warning(
-            "No matching row found where BCSS User? == 'Yes' and User Code != '-'"
-        )
-
     def verify_value_for_label(self, label_text: str, expected_text: str) -> None:
         """
         Verifies that the value cell (2nd <td>) in the row where the first <td>