Updating compartment 1 to obtain nhs_numbers of the batches from the DB instead of the downloaded file

Edited compartment 1 so IOM can be toggled on/off as currently we have a data issue

Author: adrianoaru-nhs

tests/Smokescreen/test_compartment_1.py

@@ -4,20 +4,19 @@
 from utils.batch_processing import batch_processing
 from utils.oracle import OracleDB
 
-# To Do:
-# Create a common click() function -> this aims to solve an issue where sometimes it thinks it has clicked the element but the page does not change
-
 @pytest.mark.smoke
 @pytest.mark.smokescreen
-@pytest.mark.wip
 def test_compartment_1(page: Page) -> None:
+    execute_iom = False
+
     page.goto("/")
     BcssLoginPage(page).login_as_user("BCSS401")
 
     # Create plan
+    # England
     MainMenu(page).go_to_call_and_recall_page()
     CallAndRecall(page).go_to_planning_and_monitoring_page()
-    InvitationsMonitoring(page).go_to_bcss001_invitations_plan_page() # This code is for bcss001 (IOM)
+    InvitationsMonitoring(page).go_to_bcss001_invitations_plan_page()
     InvitationsPlans(page).go_to_create_a_plan_page()
     CreateAPlan(page).click_set_all_button()
     CreateAPlan(page).fill_daily_invitation_rate_field("10")
@@ -27,18 +26,20 @@ def test_compartment_1(page: Page) -> None:
     CreateAPlan(page).fill_note_field("test data")
     CreateAPlan(page).click_saveNote_button()
     InvitationsPlans(page).invitations_plans_title.wait_for()
-    NavigationBar(page).click_back_link()
 
-    InvitationsMonitoring(page).go_to_bcss009_invitations_plan_page() # This code is Replicated for bcss009 (IOM)
-    InvitationsPlans(page).go_to_create_a_plan_page()
-    CreateAPlan(page).click_set_all_button()
-    CreateAPlan(page).fill_daily_invitation_rate_field("1")
-    CreateAPlan(page).click_update_button()
-    CreateAPlan(page).click_confirm_button()
-    CreateAPlan(page).click_save_button()
-    CreateAPlan(page).fill_note_field("test data")
-    CreateAPlan(page).click_saveNote_button()
-    InvitationsPlans(page).invitations_plans_title.wait_for()
+    # Isle Of Man
+    if execute_iom:
+        NavigationBar(page).click_back_link()
+        InvitationsMonitoring(page).go_to_bcss009_invitations_plan_page()
+        InvitationsPlans(page).go_to_create_a_plan_page()
+        CreateAPlan(page).click_set_all_button()
+        CreateAPlan(page).fill_daily_invitation_rate_field("1") # This step will fail as currently there are insufficient subjects in the next invitations shortlist (we are working on this with  compartment 0)
+        CreateAPlan(page).click_update_button()
+        CreateAPlan(page).click_confirm_button()
+        CreateAPlan(page).click_save_button()
+        CreateAPlan(page).fill_note_field("test data")
+        CreateAPlan(page).click_saveNote_button()
+        InvitationsPlans(page).invitations_plans_title.wait_for()
 
     # Generate Invitations
     NavigationBar(page).click_main_menu_link()
@@ -49,24 +50,26 @@ def test_compartment_1(page: Page) -> None:
 
     # Print the batch of Pre-Invitation Letters - England
     batch_processing(page, "S1", "Pre-invitation (FIT) (digital leaflet)", "S9 - Pre-invitation Sent")
-    s1_nhs_numbers = batch_processing(page, "S1", "Pre-invitation (FIT)", "S9 - Pre-invitation Sent") # New batch processing for S1 IOM TODO
-    for nhs_no in range(len(s1_nhs_numbers)): # Change this to get nhs numbers from DF instead of list TODO
-        OracleDB().exec_bcss_timed_events(s1_nhs_numbers[nhs_no-1])
+    nhs_number_df = batch_processing(page, "S1", "Pre-invitation (FIT)", "S9 - Pre-invitation Sent")
+    for index, row in nhs_number_df.iterrows():
+        OracleDB().exec_bcss_timed_events(row["subject_nhs_number"])
 
-    # Print the batch of Pre-Invitation Letters - IsleMan
-    s1_nhs_numbers = batch_processing(page, "S1", "Pre-invitation (gFOBT)", "S9 - Pre-invitation Sent") # New batch processing for S1 IOM TODO
-    for nhs_no in range(len(s1_nhs_numbers)): # Change this to get nhs numbers from DF instead of list TODO
-        OracleDB().exec_bcss_timed_events(s1_nhs_numbers[nhs_no-1])
+    # Print the batch of Pre-Invitation Letters - Isle Of Man
+    if execute_iom:
+        nhs_number_df = batch_processing(page, "S1", "Pre-invitation (gFOBT)", "S9 - Pre-invitation Sent")
+        for index, row in nhs_number_df.iterrows():
+            OracleDB().exec_bcss_timed_events(row["subject_nhs_number"])
 
     # Print the batch of Invitation & Test Kit Letters - England
-    s9_nhs_numbers = batch_processing(page, "S9", "Invitation & Test Kit (FIT)", "S10 - Invitation & Test Kit Sent") # New batch processing for S9 IOM TODO
-    for nhs_no in range(len(s9_nhs_numbers)):
-        OracleDB().exec_bcss_timed_events(s9_nhs_numbers[nhs_no-1]) # Change this to get nhs numbers from DF instead of list TODO
+    nhs_number_df = batch_processing(page, "S9", "Invitation & Test Kit (FIT)", "S10 - Invitation & Test Kit Sent")
+    for index, row in nhs_number_df.iterrows():
+        OracleDB().exec_bcss_timed_events(row["subject_nhs_number"])
 
-    # Print the batch of Pre-Invitation Letters - IsleMan
-    s9_nhs_numbers = batch_processing(page, "S9", "Invitation & Test Kit (gFOBT)", "S10 - Invitation & Test Kit Sent") # New batch processing for S9 IOM TODO
-    for nhs_no in range(len(s9_nhs_numbers)):
-        OracleDB().exec_bcss_timed_events(s9_nhs_numbers[nhs_no-1]) # Change this to get nhs numbers from DF instead of list TODO
+    # Print the batch of Pre-Invitation Letters - Isle Of Man
+    if execute_iom:
+        nhs_number_df = batch_processing(page, "S9", "Invitation & Test Kit (gFOBT)", "S10 - Invitation & Test Kit Sent")
+        for index, row in nhs_number_df.iterrows():
+            OracleDB().exec_bcss_timed_events(row["subject_nhs_number"])
 
     # Print a set of reminder letters
     batch_processing(page, "S10", "Test Kit Reminder", "S19 - Reminder of Initial Test Sent")

utils/batch_processing.py

@@ -6,6 +6,7 @@
 from pages.archived_batch_list_page import ArchivedBatchList
 from utils.pdf_reader import extract_nhs_no_from_pdf
 from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
+from utils.get_nhs_no_from_batch_id import get_nhs_no_from_batch_id
 import os
 import pytest
 from playwright.sync_api import Page
@@ -28,18 +29,18 @@ def batch_processing(page: Page, batch_type: str, batch_description: str, latest
         row = batch_description_cells.nth(i).locator("..")  # Get the parent row
 
         # Check if the row contains "Prepared" or "Open"
-        if row.locator("td", has_text="Prepared").count() > 0 or row.locator("td", has_text="Open").count() > 0: #Within if statement get nhs_number from DB TODO
+        if row.locator("td", has_text="Prepared").count() > 0 or row.locator("td", has_text="Open").count() > 0:
             # Find the first link in that row and click it
             link = row.locator("a").first
             link_text = link.inner_text()  # Get the batch id dynamically
-            # new code here query DB (getSubjectsFromLetterBatchID function in selenium code) passing in a batch id to get the nhs numbers of the subjects TODO
+            nhs_no_df = get_nhs_no_from_batch_id(link_text)
             link.click()
             break
         else:
             pytest.fail(f"No open/prepared '{batch_type} - {batch_description}' batch found")
 
     ManageActiveBatch(page).click_prepare_button()
-
+    page.wait_for_timeout(1000) # This one second timeout does not affect the time to execute, as it is just used to ensure the reprepare batch button is clicked and does not instantly advance to the next step
     ManageActiveBatch(page).reprepare_batch_text.wait_for()
 
     # This loops through each Retrieve button and clicks each one
@@ -55,8 +56,6 @@ def batch_processing(page: Page, batch_type: str, batch_description: str, latest
         # Wait for the download process to complete and save the downloaded file in a temp folder
         download_file.save_as(file)
         if file.endswith(".pdf"):
-            nhs_numbers = extract_nhs_no_from_pdf(file) # Remove pdf_reader TODO
-            first_nhs_no = nhs_numbers[0]
             os.remove(file) # Deletes the file after extracting the necessary data
         elif file.endswith(".csv"):
             os.remove(file) # Deletes the file after extracting the necessary data
@@ -74,5 +73,6 @@ def batch_processing(page: Page, batch_type: str, batch_description: str, latest
     ArchivedBatchList(page).enter_id_filter(link_text)
     ArchivedBatchList(page).verify_table_data(link_text)
 
-    verify_subject_event_status_by_nhs_no(page, first_nhs_no, latest_event_status) # This needs to be changed to pass an NHS number from the new DF first_nhs_no = df["COL NAME"].iloc[0] TODO
-    return nhs_numbers
+    first_nhs_no = nhs_no_df["subject_nhs_number"].iloc[0]
+    verify_subject_event_status_by_nhs_no(page, first_nhs_no, latest_event_status)
+    return nhs_no_df

utils/get_nhs_no_from_batch_id.py

@@ -0,0 +1,15 @@
+import pandas as pd
+from oracle import OracleDB
+
+def get_nhs_no_from_batch_id(batch_id):
+    nhs_number_df = OracleDB().execute_query(f"""
+    SELECT SUBJECT_NHS_NUMBER
+    FROM SCREENING_SUBJECT_T ss
+    INNER JOIN sd_contact_t c ON ss.subject_nhs_number = c.nhs_number
+    INNER JOIN LETT_BATCH_RECORDS lbr
+    ON ss.SCREENING_SUBJECT_ID = lbr.SCREENING_SUBJECT_ID
+    WHERE lbr.BATCH_ID IN {batch_id}
+    AND ss.screening_status_id != 4008
+    ORDER BY ss.subject_nhs_number
+    FETCH FIRST 5 ROWS ONLY""")
+    return nhs_number_df