Added logging to C1 and added pytest parks to pytest.ini

adrianoaru-nhs · adrianoaru-nhs · commit f0d3c7de28db · 2025-03-24T14:58:13.000Z
diff --git a/pages/generate_invitations_page.py b/pages/generate_invitations_page.py
@@ -1,6 +1,6 @@
 from playwright.sync_api import Page, expect
 import pytest
-
+import logging
 
 class GenerateInvitations:
     def __init__(self, page: Page):
@@ -32,6 +32,7 @@ def wait_for_invitation_generation_complete(self):
         elapsed = 0
 
         # Loop until the table no longer contains "Queued"
+        logging.info(f"Waiting for successful generation")
         while elapsed < timeout: # there may be a stored procedure to speed this process up
             table_text = self.displayRS.text_content()
             if "Queued" in table_text or "In Progress" in table_text:
@@ -45,8 +46,13 @@ def wait_for_invitation_generation_complete(self):
                 break
 
         # Final check: ensure that the table now contains "Completed"
-        expect(self.displayRS).to_contain_text("Completed")
+        try:
+            expect(self.displayRS).to_contain_text("Completed")
+            logging.info("Invitations successfully generated")
+            logging.info(f"Invitations took {elapsed/1000} seconds to generate")
+        except Exception as e:
+            pytest.fail("Invitations not generated successfully")
 
         value = self.planned_invitations_total.text_content().strip()  # Get text and remove extra spaces
-        if int(value) <= 5:
+        if int(value) < 5:
             pytest.fail("There are less than 5 invitations generated")
diff --git a/pytest.ini b/pytest.ini
@@ -27,3 +27,7 @@ markers =
     utils: test setup and support methods
     smoke: tests designed to run as part of the smokescreen regression test suite
     wip: tests that are currently in progress
+    smokescreen: all compartments to be run as part of the smokescreen
+    compartment1: only for compartment 1
+    compartment2: only for compartment 2
+    compartment3: only for compartment 3
diff --git a/tests/Smokescreen/test_compartment_1.py b/tests/Smokescreen/test_compartment_1.py
@@ -6,16 +6,16 @@
 
 @pytest.mark.smoke
 @pytest.mark.smokescreen
+@pytest.mark.compartment1
 def test_compartment_1(page: Page) -> None:
-
-    page.goto("/")
-    BcssLoginPage(page).login_as_user("BCSS401")
+    UserTools.user_login(page, "Hub Manager State Registered")
 
     # 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()
     InvitationsPlans(page).go_to_create_a_plan_page()
+    logging.info("Setting daily invitation rate")
     CreateAPlan(page).click_set_all_button()
     CreateAPlan(page).fill_daily_invitation_rate_field("10")
     CreateAPlan(page).click_update_button()
@@ -29,24 +29,24 @@ def test_compartment_1(page: Page) -> None:
     NavigationBar(page).click_main_menu_link()
     MainMenu(page).go_to_call_and_recall_page()
     CallAndRecall(page).go_to_generate_invitations_page()
+    logging.info("Generating Invitations")
     GenerateInvitations(page).click_generate_invitations_button()
     GenerateInvitations(page).wait_for_invitation_generation_complete()
 
     # Print the batch of Pre-Invitation Letters - England
     batch_processing(page, "S1", "Pre-invitation (FIT) (digital leaflet)", "S9 - Pre-invitation Sent")
     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"])
+    OracleDB().exec_bcss_timed_events(nhs_number_df)
 
     # Print the batch of Invitation & Test Kit Letters - England
     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"])
+    OracleDB().exec_bcss_timed_events(nhs_number_df)
 
     # Print a set of reminder letters
     batch_processing(page, "S10", "Test Kit Reminder", "S19 - Reminder of Initial Test Sent")
 
     # Log out
+    logging.info("Logging Out")
     NavigationBar(page).click_log_out_link()
     Logout(page).verify_log_out_page()
     page.close()
diff --git a/tests/Smokescreen/test_compartment_2.py b/tests/Smokescreen/test_compartment_2.py
@@ -8,9 +8,9 @@
 
 
 @pytest.mark.smokescreen
+@pytest.mark.compartment2
 def test_compartment_2(page: Page) -> None:
-    page.goto("/")
-    BcssLoginPage(page).login_as_user("BCSS401")
+    UserTools.user_login(page, "Hub Manager State Registered")
 
     MainMenu(page).go_to_fit_test_kits_page()
     FITTestKits(page).go_to_log_devices_page()
diff --git a/tests/Smokescreen/test_compartment_3.py b/tests/Smokescreen/test_compartment_3.py
@@ -9,8 +9,8 @@
 from utils.oracle import OracleDB
 from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
 
-
-@pytest.mark.wip2
+@pytest.mark.smokescreen
+@pytest.mark.compartment3
 def test_compartment_3(page: Page) -> None:
     UserTools.user_login(page, "Hub Manager State Registered")
 
@@ -45,49 +45,17 @@ def test_compartment_3(page: Page) -> None:
             logging.error(f"Verification failed for NHS number {nhs_number} with status {expected_status}: {str(e)}")
             raise
 
-
-    # Navigate to log devices page
-    MainMenu(page).go_to_fit_test_kits_page()
-    FITTestKits(page).go_to_log_devices_page()
-
-    # Get a fit device id
-    subjectdf = create_fit_id_df()
-
-    # Log fit device
-    for subject in range(4):
-        fit_device_id = subjectdf["fit_device_id"].iloc[subject]
-        LogDevices(page).fill_fit_device_id_field(fit_device_id)
-        sample_date = datetime.now().strftime("%#d %b %Y")
-        LogDevices(page).fill_sample_date_field(sample_date)
-        LogDevices(page).verify_successfully_logged_device_text()
-
-    # Check statuses of 'normal' FIT kit subjects has moved to S2
-    nhs_no = subjectdf["subject_nhs_number"].iloc[0]
-    verify_subject_event_status_by_nhs_no(page, nhs_no, "S2 - Normal")
-
-    # Check status of 'abnormal' FIT kit subjects has moved to A8
-    nhs_no = subjectdf["subject_nhs_number"].iloc[0]
-    verify_subject_event_status_by_nhs_no(page, nhs_no, "A8 - Abnormal")
-
     # (Step 12) - Process S2 batch
     # Run batch processing function on S2 batch
-    batch_processing(page, "S2", "Normal", "S2 - Normal")
-    nhs_number_df = batch_processing(page, "S2", "Normal", "S2 - Normal")
-    for index, row in nhs_number_df.iterrows():
-        OracleDB().exec_bcss_timed_events(row["subject_nhs_number"])
-
-    # Verify Status of subject is updated to S158
-    nhs_no = subjectdf["subject_nhs_number"].iloc[0]
-    verify_subject_event_status_by_nhs_no(page, nhs_no, "S158")
+    nhs_number_df = batch_processing(page, "S2", "Subject Result (Normal)", "S158 - Subject Discharge Sent (Normal)")
+    OracleDB().exec_bcss_timed_events(nhs_number_df)
 
     # (Step 13) - Process S158 batch
     # Run batch processing function on S158 batch
-    batch_processing(page, "S158", "Normal result letter sent to subject", "S158 - Normal result letter sent to subject")
-    nhs_number_df = batch_processing(page, "S158", "Normal result letter sent to subject", "S158 - Normal result letter sent to subject")
-    for index, row in nhs_number_df.iterrows():
-        OracleDB().exec_bcss_timed_events(row["subject_nhs_number"])
-
-    # Verify Status of subject is updated to S159
-    nhs_no = subjectdf["subject_nhs_number"].iloc[0]
-    verify_subject_event_status_by_nhs_no(page, nhs_no, "S159 - Normal result letter sent to GP")
+    nhs_number_df = batch_processing(page, "S158", "GP Result (Normal)", "S159 - GP Discharge Sent (Normal)")
+    OracleDB().exec_bcss_timed_events(nhs_number_df)
 
+    # Log out
+    NavigationBar(page).click_log_out_link()
+    Logout(page).verify_log_out_page()
+    page.close()
diff --git a/utils/batch_processing.py b/utils/batch_processing.py
@@ -4,14 +4,15 @@
 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.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
+import logging
 
 def batch_processing(page: Page, batch_type: str, batch_description: str, latest_event_status: str):
+    logging.info(f"Processing {batch_type} - {batch_description} batch")
     NavigationBar(page).click_main_menu_link()
     MainMenu(page).go_to_communications_production_page()
     CommunicationsProduction(page).go_to_active_batch_list_page()
@@ -20,59 +21,81 @@ def batch_processing(page: Page, batch_type: str, batch_description: str, latest
     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")
+        logging.warning("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:
+        # Check if the row contains "Open"
+        if 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
-            nhs_no_df = get_nhs_no_from_batch_id(link_text)
+            try:
+                logging.info(f"Attempting to get NHS Numbers for batch: {link_text} from the DB")
+                nhs_no_df = get_nhs_no_from_batch_id(link_text)
+                logging.info(f"Successfully retrieved NHS Numbers from batch: {link_text}")
+            except Exception as e:
+                logging.error(f"Failed to retrieve NHS Numbers from batch: {link_text}")
             link.click()
             break
         else:
-            pytest.fail(f"No open/prepared '{batch_type} - {batch_description}' batch found")
+            pytest.fail(f"No open '{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
     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"):
-            os.remove(file) # Deletes the file after extracting the necessary data
-        elif file.endswith(".csv"):
+    try:
+        for retrieve_button in range (ManageActiveBatch(page).retrieve_button.count()):
+            retrieve_button_count += 1
+            logging.info(f"Clicking retrieve button {retrieve_button_count}")
+            # 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)
             os.remove(file) # Deletes the file after extracting the necessary data
+    except Exception as e:
+        logging.error(f"No retrieve button available to click: {str(e)}")
 
     # 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()
+    try:
+        for confirm_printed_button in range (retrieve_button_count):
+            logging.info(f"Clicking confirm printed button {confirm_printed_button+1}")
+            page.once("dialog", lambda dialog: dialog.accept())
+            ManageActiveBatch(page).confirm_button.nth(0).click()
+    except Exception as e:
+        logging.error(f"No confirm printed button available to click: {str(e)}")
 
-    ActiveBatchList(page).batch_successfully_archived_msg.wait_for()
+    try:
+        ActiveBatchList(page).batch_successfully_archived_msg.wait_for()
+        logging.info(f"Batch {link_text} successfully archived")
+    except Exception as e:
+        logging.error(f"Batch successfully archived message is not shown: {str(e)}")
 
     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)
+    try:
+        ArchivedBatchList(page).verify_table_data(link_text)
+        logging.info(f"Batch {link_text} visible in archived batch list")
+    except Exception:
+        logging.error(f"Batch {link_text} not visible in archived batch list")
 
     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)
+    try:
+        verify_subject_event_status_by_nhs_no(page, first_nhs_no, latest_event_status)
+        logging.info(f"Successfully verified NHS number {first_nhs_no} with status {latest_event_status}")
+    except Exception as e:
+        logging.error(f"Verification failed for NHS number {first_nhs_no} with status {latest_event_status}: {str(e)}")
     return nhs_no_df
diff --git a/utils/fit_kit_logged.py b/utils/fit_kit_logged.py
@@ -2,7 +2,7 @@
 import pandas as pd
 from datetime import datetime
 import logging
-
+import pytest
 
 
 def process_kit_data():
@@ -21,7 +21,7 @@ def process_kit_data():
         logging.info(f"Processing normal kit with Device ID: {device_id}")  # Logging normal device_id
         device_ids.append((device_id, True))  # Add to the list with normal flag
     else:
-        logging.warning("No normal kits found for processing.")  # Log warning
+        pytest.fail("No normal kits found for processing.")
 
         # Process abnormal kits (multiple, loop through)
     if not abnormal_fit_kit_df.empty:
@@ -30,7 +30,7 @@ def process_kit_data():
             logging.info(f"Processing abnormal kit with Device ID: {device_id}")  # Logging abnormal device_id
             device_ids.append((device_id, False))  # Add to the list with abnormal flag
     else:
-        logging.warning("No abnormal kits found for processing.")  # Log warning
+        pytest.fail("No abnormal kits found for processing.")
 
     return device_ids
 
diff --git a/utils/oracle.py b/utils/oracle.py
