Converting functions in test_compartment_1.py into utils

adrianoaru-nhs · adrianoaru-nhs · commit a8f574118c23 · 2025-03-06T09:37:20.000Z
Adding functionality of execution bcss_time_events to progress subjects instead of updating parameters on the DB
diff --git a/tests/Smokescreen/my_pages.py b/tests/Smokescreen/my_pages.py
@@ -13,4 +13,3 @@
 from pages.invitations_plans_page import *
 from pages.generate_invitations_page import *
 from pages.log_out_page import *
-from utils.oracle import *
diff --git a/tests/Smokescreen/test_compartment_1.py b/tests/Smokescreen/test_compartment_1.py
@@ -1,17 +1,10 @@
 import pytest
-from pypdf import PdfReader
-from playwright.sync_api import Page, expect
-import pandas as pd
-import os
+from playwright.sync_api import Page
 from my_pages import *
+from utils.batch_processing import batch_processing
+from utils.oracle import exec_bcss_timed_events
 
 # To Do:
-# Create more POMs
-# Add more fail states
-# Add more logging to understand what is going on
-# Convert import pages * into 1 line
-# Move functions to utils
-# Remove as many hard coded timeouts as possible
 # Create a generic 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.wip
@@ -41,113 +34,17 @@ def test_example(page: Page) -> None:
     GenerateInvitations(page).wait_for_invitation_generation_complete()
 
     # Print the batch of Pre-Invitation Letters
-    batch_processing(page, "S1", "Pre-invitation (FIT)", "S9 - Pre-invitation Sent")
+    s1_nhs_no = batch_processing(page, "S1", "Pre-invitation (FIT)", "S9 - Pre-invitation Sent")
+    exec_bcss_timed_events(s1_nhs_no)
     batch_processing(page, "S1", "Pre-invitation (FIT) (digital leaflet)", "S9 - Pre-invitation Sent")
 
-    database_connection_exec("bcss_timed_events")
-
     # Print the batch of Invitation & Test Kit Letters
-    batch_processing(page, "S9", "Invitation & Test Kit (FIT)", "S10 - Invitation & Test Kit Sent")
+    s9_nhs_no = batch_processing(page, "S9", "Invitation & Test Kit (FIT)", "S10 - Invitation & Test Kit Sent")
+    exec_bcss_timed_events(s9_nhs_no)
 
     # Print a set of reminder letters
     batch_processing(page, "S10", "Test Kit Reminder", "S19 - Reminder of Initial Test Sent")
 
     # Log out
     NavigationBar(page).click_log_out_link()
     Logout(page).verify_log_out_page()
-
-def batch_processing(page: Page, batch_type: str, batch_description: str, latest_event_status: str):
-    NavigationBar(page).click_main_menu_link()
-    MainMenu(page).go_to_communications_production_page()
-    CommunicationsProduction(page).go_to_active_batch_list_page()
-    ActiveBatchList(page).enter_event_code_filter(batch_type)
-
-    batch_description_cells = page.locator(f"//td[text()='{batch_description}']")
-
-    if batch_description_cells.count() == 0 and batch_description == "Pre-invitation (FIT) (digital leaflet)":
-        print(f"No S1 Pre-invitation (FIT) (digital leaflet) batch found. Skipping to next step")
-        return
-    elif batch_description_cells.count() == 0 and page.locator("td", has_text="No matching records found"):
-        pytest.fail(f"No {batch_type} {batch_description} batch found")
-
-    for i in range(batch_description_cells.count()):
-        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:
-            # 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
-            link.click()
-            break
-        else:
-            pytest.fail(f"No open/prepared '{batch_type} - {batch_description}' batch found")
-
-    ManageActiveBatch(page).click_prepare_button()
-
-    ManageActiveBatch(page).reprepare_batch_text.wait_for()
-
-    # This loops through each Retrieve button and clicks each one
-    retrieve_button_count = 0
-    for retrieve_button in range (ManageActiveBatch(page).retrieve_button.count()):
-        retrieve_button_count += 1
-        # Start waiting for the pdf download
-        with page.expect_download() as download_info:
-            # Perform the action that initiates download
-            ManageActiveBatch(page).retrieve_button.nth(retrieve_button-1).click()
-        download_file = download_info.value
-        file = download_file.suggested_filename
-        # 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_no = pdf_Reader(file)
-            os.remove(file) # Deletes the file after extracting the necessary data
-        elif file.endswith(".csv"):
-            # csv_df = csv_Reader(file) # Currently no use in compartment 1, will be necessary for future compartments
-            os.remove(file) # Deletes the file after extracting the necessary data
-
-    # This loops through each Confirm printed button and clicks each one
-    for _ in range (retrieve_button_count):
-        page.on("dialog", lambda dialog: dialog.accept())
-        ManageActiveBatch(page).confirm_button.nth(0).click()
-
-    ActiveBatchList(page).batch_successfully_archived_msg.wait_for()
-
-    NavigationBar(page).click_main_menu_link()
-    MainMenu(page).go_to_communications_production_page()
-    CommunicationsProduction(page).go_to_archived_batch_list_page()
-    ArchivedBatchList(page).enter_id_filter(link_text)
-    ArchivedBatchList(page).verify_table_data(link_text)
-
-    subject_search_by_nhs_no(page, nhs_no, latest_event_status)
-
-def pdf_Reader(file: str):
-    reader = PdfReader(file)
-
-    # For loop looping through all pages of the file to find the NHS Number
-    for pages in reader.pages:
-        text = pages.extract_text()
-        if "NHS No" in text:
-            # If NHS number is found split the text by every new line into a list
-            text = text.splitlines(True)
-            for split_text in text:
-                if "NHS No" in split_text:
-                    # If a string is found containing "NHS No" only digits are stored into nhs_no
-                    nhs_no = "".join([ele for ele in split_text if ele.isdigit()])
-                    break
-    return nhs_no
-
-def csv_Reader(file: str):
-    csv_df = pd.read_csv(file)
-    return csv_df
-
-def subject_search_by_nhs_no(page: Page, nhs_no: str, latest_event_status: str):
-    NavigationBar(page).click_main_menu_link()
-    MainMenu(page).go_to_screening_subject_search_page()
-    SubjectScreeningPage(page).click_nhs_number_filter()
-    SubjectScreeningPage(page).nhs_number_filter.fill(nhs_no)
-    SubjectScreeningPage(page).nhs_number_filter.press("Enter")
-    SubjectScreeningPage(page).click_search_button()
-    SubjectScreeningSummary(page).verify_subject_screening_summary()
-    SubjectScreeningSummary(page).verify_latest_event_status_header()
-    SubjectScreeningSummary(page).verify_latest_event_status_value(latest_event_status)
diff --git a/utils/batch_processing.py b/utils/batch_processing.py
@@ -0,0 +1,79 @@
+from pages.navigation_bar_links import NavigationBar
+from pages.bcss_home_page import MainMenu
+from pages.communications_production_page import CommunicationsProduction
+from pages.active_batch_list_page import ActiveBatchList
+from pages.manage_active_batch_page import ManageActiveBatch
+from pages.archived_batch_list_page import ArchivedBatchList
+from utils.extract_nhs_no_from_pdf import pdf_reader
+from utils.csv_reader import csv_reader
+from utils.subject_search_by_nhs_no import subject_search_by_nhs_no
+from utils.oracle import database_connection_exec, database_connection_query
+import os
+import pytest
+from playwright.sync_api import Page
+
+def batch_processing(page: Page, batch_type: str, batch_description: str, latest_event_status: str):
+    NavigationBar(page).click_main_menu_link()
+    MainMenu(page).go_to_communications_production_page()
+    CommunicationsProduction(page).go_to_active_batch_list_page()
+    ActiveBatchList(page).enter_event_code_filter(batch_type)
+
+    batch_description_cells = page.locator(f"//td[text()='{batch_description}']")
+
+    if batch_description_cells.count() == 0 and batch_description == "Pre-invitation (FIT) (digital leaflet)":
+        print(f"No S1 Pre-invitation (FIT) (digital leaflet) batch found. Skipping to next step")
+        return
+    elif batch_description_cells.count() == 0 and page.locator("td", has_text="No matching records found"):
+        pytest.fail(f"No {batch_type} {batch_description} batch found")
+
+    for i in range(batch_description_cells.count()):
+        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:
+            # 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
+            link.click()
+            break
+        else:
+            pytest.fail(f"No open/prepared '{batch_type} - {batch_description}' batch found")
+
+    ManageActiveBatch(page).click_prepare_button()
+
+    ManageActiveBatch(page).reprepare_batch_text.wait_for()
+
+    # This loops through each Retrieve button and clicks each one
+    retrieve_button_count = 0
+    for retrieve_button in range (ManageActiveBatch(page).retrieve_button.count()):
+        retrieve_button_count += 1
+        # Start waiting for the pdf download
+        with page.expect_download() as download_info:
+            # Perform the action that initiates download
+            ManageActiveBatch(page).retrieve_button.nth(retrieve_button-1).click()
+        download_file = download_info.value
+        file = download_file.suggested_filename
+        # 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_no = pdf_reader(file)
+            os.remove(file) # Deletes the file after extracting the necessary data
+        elif file.endswith(".csv"):
+            csv_df = csv_reader(file) # Currently no use in compartment 1, will be necessary for future compartments
+            os.remove(file) # Deletes the file after extracting the necessary data
+
+    # This loops through each Confirm printed button and clicks each one
+    for _ in range (retrieve_button_count):
+        page.on("dialog", lambda dialog: dialog.accept())
+        ManageActiveBatch(page).confirm_button.nth(0).click()
+
+    ActiveBatchList(page).batch_successfully_archived_msg.wait_for()
+
+    NavigationBar(page).click_main_menu_link()
+    MainMenu(page).go_to_communications_production_page()
+    CommunicationsProduction(page).go_to_archived_batch_list_page()
+    ArchivedBatchList(page).enter_id_filter(link_text)
+    ArchivedBatchList(page).verify_table_data(link_text)
+
+    subject_search_by_nhs_no(page, nhs_no, latest_event_status)
+    return nhs_no
diff --git a/utils/csv_reader.py b/utils/csv_reader.py
@@ -0,0 +1,5 @@
+import pandas as pd
+
+def csv_reader(file: str):
+    csv_df = pd.read_csv(file)
+    return csv_df
diff --git a/utils/extract_nhs_no_from_pdf.py b/utils/extract_nhs_no_from_pdf.py
@@ -0,0 +1,17 @@
+from pypdf import PdfReader
+
+def pdf_reader(file: str):
+    reader = PdfReader(file)
+
+    # For loop looping through all pages of the file to find the NHS Number
+    for pages in reader.pages:
+        text = pages.extract_text()
+        if "NHS No" in text:
+            # If NHS number is found split the text by every new line into a list
+            text = text.splitlines(True)
+            for split_text in text:
+                if "NHS No" in split_text:
+                    # If a string is found containing "NHS No" only digits are stored into nhs_no
+                    nhs_no = "".join([ele for ele in split_text if ele.isdigit()])
+                    break
+    return nhs_no
diff --git a/utils/oracle.py b/utils/oracle.py
@@ -16,7 +16,7 @@ def database_connection_exec(procedure: str): # To use when "exec xxxx" (stored
             print(f"Attempting to execute stored procedure: {procedure}")
             cursor = conn.cursor()
             cursor.callproc(procedure)
-            print(conn.version, "DB write successful!")
+            print(conn.version, "stored procedure execution successful!")
         except Exception as executionError:
             print(f"Failed to execute stored procedure with execution error: {executionError}")
         finally:
@@ -25,7 +25,39 @@ def database_connection_exec(procedure: str): # To use when "exec xxxx" (stored
     except Exception as connectionError:
             print(f"Failed to connect to the DB! with connection error: {connectionError}")
 
-def database_connection_query(query: str): # To use when "exec xxxx" (stored procedures)
+def exec_bcss_timed_events(nhs_no: str): # To use when "exec xxxx" (stored procedures)
+    load_dotenv()
+    un = os.getenv("un")
+    cs = os.getenv("cs")
+    pw = os.getenv("pw")
+
+    try:
+        print("Attempting DB connection...")
+        conn = oracledb.connect(user=un, password=pw, dsn=cs)
+        print(conn.version, "DB connection successful!")
+        try:
+            print(f"Attempting to get subject_id from nhs number: {nhs_no}")
+            cursor = conn.cursor()
+            cursor.execute(f"SELECT SCREENING_SUBJECT_ID FROM SCREENING_SUBJECT_T WHERE SUBJECT_NHS_NUMBER = {int(nhs_no)}")
+            result = cursor.fetchall()
+            subject_id = result[0][0]
+            print(conn.version, "Able to extract subject ID")
+            try:
+                print(f"Attempting to execute stored procedure: {f"'bcss_timed_events', [subject_id,'Y']"}")
+                cursor = conn.cursor()
+                cursor.callproc('bcss_timed_events', [subject_id,'Y'])
+                print(conn.version, "stored procedure execution successful!")
+            except Exception as spExecutionError:
+                print(f"Failed to execute stored procedure with execution error: {spExecutionError}")
+        except Exception as queryExecutionError:
+            print(f"Failed to to extract subject ID with error: {queryExecutionError}")
+        finally:
+            if conn is not None:
+                conn.close()
+    except Exception as connectionError:
+            print(f"Failed to connect to the DB! with connection error: {connectionError}")
+
+def database_connection_query(query: str,): # To use when "exec xxxx" (stored procedures)
     load_dotenv()
     un = os.getenv("un")
     cs = os.getenv("cs")
@@ -39,12 +71,14 @@ def database_connection_query(query: str): # To use when "exec xxxx" (stored pro
             print(f"Attempting to execute query: {query}")
             cursor = conn.cursor()
             cursor.execute(query)
-            print(conn.version, "DB write successful!")
+            result = cursor.fetchall()
+            print(conn.version, "query execution successful!")
         except Exception as executionError:
             print(f"Failed to execute query with execution error {executionError}")
         finally:
             if conn is not None:
                 conn.close()
+                return result
     except Exception as connectionError:
             print(f"Failed to connect to the DB! with connection error {connectionError}")
 
diff --git a/utils/subject_search_by_nhs_no.py b/utils/subject_search_by_nhs_no.py
@@ -0,0 +1,16 @@
+from pages.navigation_bar_links import NavigationBar
+from pages.bcss_home_page import MainMenu
+from pages.subject_screening_page import SubjectScreeningPage
+from pages.subject_screening_summary import SubjectScreeningSummary
+from playwright.sync_api import Page
+
+def subject_search_by_nhs_no(page: Page, nhs_no: str, latest_event_status: str):
+    NavigationBar(page).click_main_menu_link()
+    MainMenu(page).go_to_screening_subject_search_page()
+    SubjectScreeningPage(page).click_nhs_number_filter()
+    SubjectScreeningPage(page).nhs_number_filter.fill(nhs_no)
+    SubjectScreeningPage(page).nhs_number_filter.press("Enter")
+    SubjectScreeningPage(page).click_search_button()
+    SubjectScreeningSummary(page).verify_subject_screening_summary()
+    SubjectScreeningSummary(page).verify_latest_event_status_header()
+    SubjectScreeningSummary(page).verify_latest_event_status_value(latest_event_status)
