Merge branch 'main' into feature/BCSS-20362-c5-pom-for-colonoscopy-datasets

Author: Andyg79

.github/actions/run-playwright-tests/action.yml

@@ -8,7 +8,7 @@ inputs:
     required: true
     type: string
   marker_to_use:
-    description: "The test marker to use when running tests (e.g. smokescreen)"
+    description: "The test marker to use when running tests (e.g. smoke)"
     required: true
     type: string
 
@@ -29,13 +29,10 @@ runs:
       run: python -m playwright install --with-deps
     - name: Run specified tests
       shell: bash
-      run: pytest -m ${{ inputs.marker_to_use }} --base-url=$URL_TO_USE --ignore=tests_utils/
-      env:
-        BCSS_PASS: ${{ secrets.BCSS_PASS }}
-        ORACLE_DB: ${ ${{ secrets.ORACLE_DB }}/<REPLACE>/${{ inputs.bcss_cloud_environment }} }
-        ORACLE_USERNAME: ${{ secrets.ORACLE_USERNAME }}
-        ORACLE_PASS: ${{ secrets.ORACLE_PASS }}
-        URL_TO_USE: ${ ${{ vars.CLOUD_ENV_URL }}/<REPLACE>/${{ inputs.bcss_cloud_environment }} }
+      run: |
+        URL_TO_USE="${URL_TO_USE_DEFAULT//<REPLACE>/${{ inputs.bcss_cloud_environment }}}"
+        ORACLE_DB="${ORACLE_DB_DEFAULT//<REPLACE>/${{ inputs.bcss_cloud_environment }}}"
+        pytest -m ${{ inputs.marker_to_use }} --base-url=https://$URL_TO_USE --ignore=tests_utils/
     - uses: actions/upload-artifact@v4
       if: ${{ !cancelled() }}
       with:

.github/workflows/execute-tests.yaml

@@ -1,5 +1,10 @@
 name: "Test Runner"
 
+# This workflow is triggered manually and allows the user to specify the environment and test marker to run.
+# It is functional, however will not work against BCSS test environments until we configure some self-hosted
+# GitHub runners, as the GitHub-hosted runners are based outside of the UK so get blocked by the WAF on the
+# environments.
+
 on:
   workflow_dispatch:
     inputs:
@@ -17,12 +22,19 @@ jobs:
   run-tests:
     name: "Run Specified Tests"
     runs-on: ubuntu-latest
-    timeout-minutes: 3
+    timeout-minutes: 10
     steps:
       - name: "Checkout code"
         uses: actions/checkout@v4
       - name: "Run Tests"
+        id: run-tests
         uses: ./.github/actions/run-playwright-tests
         with:
           bcss_cloud_environment: "${{ inputs.bcss_cloud_environment }}"
           marker_to_use: "${{ inputs.marker_to_use }}"
+        env:
+          BCSS_PASS: ${{ secrets.BCSS_PASS }}
+          ORACLE_DB_DEFAULT: ${{ secrets.ORACLE_DB }}
+          ORACLE_USERNAME: ${{ secrets.ORACLE_USERNAME }}
+          ORACLE_PASS: ${{ secrets.ORACLE_PASS }}
+          URL_TO_USE_DEFAULT: ${{ vars.CLOUD_ENV_URL }}

pages/screening_practitioner_appointments/appointment_detail_page.py

@@ -8,15 +8,15 @@ def __init__(self, page: Page):
         self.page = page
         # Appointment Detail - page filters
         self.attendance_radio = self.page.get_by_role("radio", name="Attendance")
-        self.attendented_check_box = self.page.locator("#UI_ATTENDED")
+        self.attended_check_box = self.page.locator("#UI_ATTENDED")
         self.calendar_button = self.page.get_by_role("button", name="Calendar")
         self.save_button = self.page.get_by_role("button", name="Save")
 
     def check_attendance_radio(self) -> None:
         self.attendance_radio.check()
 
-    def check_attendented_check_box(self) -> None:
-        self.attendented_check_box.check()
+    def check_attended_check_box(self) -> None:
+        self.attended_check_box.check()
 
     def click_calendar_button(self) -> None:
         self.click(self.calendar_button)

pages/screening_subject_search/attend_diagnostic_test_page.py

@@ -0,0 +1,21 @@
+from playwright.sync_api import Page
+from pages.base_page import BasePage
+
+
+class AttendDiagnosticTest(BasePage):
+    def __init__(self, page: Page):
+        super().__init__(page)
+        self.page = page
+        # Advance Diagnostic Test - page locators
+        self.actual_type_of_test_dropdown = self.page.locator("#UI_CONFIRMED_TYPE_OF_TEST")
+        self.calendar_button = self.page.get_by_role("button", name="Calendar")
+        self.save_button =self.page.get_by_role("button", name="Save")
+
+    def select_actual_type_of_test_dropdown_option(self, text: str) -> None:
+        self.actual_type_of_test_dropdown.select_option(label=text)
+
+    def click_calendar_button(self) -> None:
+        self.click(self.calendar_button)
+
+    def click_save_button(self) -> None:
+        self.click(self.save_button)

requirements.txt

@@ -7,3 +7,4 @@ pandas~=2.2.3
 python-dotenv~=1.0.1
 sqlalchemy>=2.0.38
 jproperties~=2.1.2
+pypdf>=5.3.0

tests/smokescreen/test_compartment_1.py

@@ -88,7 +88,7 @@ def test_compartment_1(page: Page, smokescreen_properties: dict) -> None:
             "Skipping S1 Pre-invitation (FIT) (digital leaflet) as no self referral invitations were generated"
         )
     batch_processing(
-        page, "S1", "Pre-invitation (FIT)", "S9 - Pre-invitation Sent", True
+        page, "S1", "Pre-invitation (FIT)", "S9 - Pre-invitation Sent", True, True
     )
 
     # Print the batch of Invitation & Test Kit Letters - England

tests/smokescreen/test_compartment_5.py

@@ -28,6 +28,9 @@
 from pages.screening_practitioner_appointments.appointment_calendar_page import (
     AppointmentCalendar,
 )
+from pages.screening_subject_search.attend_diagnostic_test_page import (
+    AttendDiagnosticTest,
+)
 from utils.user_tools import UserTools
 from utils.load_properties_file import PropertiesFile
 from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
@@ -84,7 +87,7 @@ def test_compartment_5(page: Page, smokescreen_properties: dict) -> None:
 
     # Select Attendance radio button, tick Attended checkbox, set Attended Date to yesterday's (system) date and then press Save
     AppointmentDetail(page).check_attendance_radio()
-    AppointmentDetail(page).check_attendented_check_box()
+    AppointmentDetail(page).check_attended_check_box()
     AppointmentDetail(page).click_calendar_button()
     CalendarPicker(page).v1_calender_picker(datetime.today() - timedelta(1))
     AppointmentDetail(page).click_save_button()
@@ -148,10 +151,10 @@ def test_compartment_5(page: Page, smokescreen_properties: dict) -> None:
     AdvanceFOBTScreeningEpisode(page).click_attend_diagnostic_test_button()
 
     # Select Colonoscopy from drop down list. Enter the actual appointment date as today's date and select 'Save'
-    page.locator("#UI_CONFIRMED_TYPE_OF_TEST").select_option(label="Colonoscopy")
-    page.get_by_role("button", name="Calendar").click()
+    AttendDiagnosticTest.select_actual_type_of_test_dropdown_option("Colonoscopy")
+    AttendDiagnosticTest.click_calendar_button()
     CalendarPicker(page).v1_calender_picker(datetime.today())
-    page.get_by_role("button", name="Save").click()
+    AttendDiagnosticTest.click_save_button()
     SubjectScreeningSummary(page).verify_latest_event_status_value(
         "A259 - Attended Diagnostic Test"
     )

utils/batch_processing.py

@@ -10,10 +10,12 @@
 from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
 from utils.oracle.oracle_specific_functions import get_nhs_no_from_batch_id
 from utils.oracle.oracle import OracleDB
+from utils.pdf_reader import extract_nhs_no_from_pdf
 import os
 import pytest
 from playwright.sync_api import Page
 import logging
+import pandas as pd
 
 
 def batch_processing(
@@ -22,6 +24,7 @@ def batch_processing(
     batch_description: str,
     latest_event_status: str,
     run_timed_events: bool = False,
+    get_subjects_from_pdf: bool = False,
 ) -> None:
     """
     This util is used to process batches. It expects the following inputs:
@@ -30,6 +33,7 @@ def batch_processing(
     - batch_description: This is the description of the batch. E.g. Pre-invitation (FIT)
     - latest_event_status: This is the status the subject will get updated to after the batch has been processed.
     - run_timed_events: This is an optional input that executes bcss_timed_events if set to True
+    - get_subjects_from_pdf: This is an optial input to change the method of retrieving subjects from the batch from the Db to the PDF file.
     """
     logging.info(f"Processing {batch_type} - {batch_description} batch")
     BasePage(page).click_main_menu_link()
@@ -39,8 +43,9 @@ def batch_processing(
 
     batch_description_cells = page.locator(f"//td[text()='{batch_description}']")
 
-    if batch_description_cells.count() == 0 and page.locator(
-        "td", has_text="No matching records found"
+    if (
+        batch_description_cells.count() == 0
+        and page.locator("td", has_text="No matching records found").is_visible()
     ):
         pytest.fail(f"No {batch_type} {batch_description} batch found")
 
@@ -55,41 +60,31 @@ def batch_processing(
             logging.info(
                 f"Successfully found open '{batch_type} - {batch_description}' batch"
             )
-            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:
-                pytest.fail(
-                    f"Failed to retrieve NHS Numbers from batch {link_text}, {str(e)}"
-                )
             link.click()
             break
         elif (i + 1) == batch_description_cells.count():
             pytest.fail(f"No open '{batch_type} - {batch_description}' batch found")
 
-    prepare_and_print_batch(page, link_text)
+    if get_subjects_from_pdf:
+        logging.info(f"Getting NHS Numbers for batch {link_text} from the PDF File")
+        nhs_no_df = prepare_and_print_batch(page, link_text, get_subjects_from_pdf)
+    else:
+        logging.info(f"Getting NHS Numbers for batch {link_text} from the DB")
+        prepare_and_print_batch(page, link_text, get_subjects_from_pdf)
+        nhs_no_df = get_nhs_no_from_batch_id(link_text)
 
     check_batch_in_archived_batch_list(page, link_text)
 
     first_nhs_no = nhs_no_df["subject_nhs_number"].iloc[0]
-    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:
-        pytest.fail(f"Verification failed for NHS number {first_nhs_no}: {str(e)}")
+    verify_subject_event_status_by_nhs_no(page, first_nhs_no, latest_event_status)
 
     if run_timed_events:
         OracleDB().exec_bcss_timed_events(nhs_no_df)
 
 
-def prepare_and_print_batch(page: Page, link_text) -> None:
+def prepare_and_print_batch(
+    page: Page, link_text: str, get_subjects_from_pdf: bool = False
+) -> pd.DataFrame | None:
     """
     This method prepares the batch, retreives the files and confirms them as printed
     Once those buttons have been pressed it waits for the message 'Batch Successfully Archived'
@@ -114,6 +109,11 @@ def prepare_and_print_batch(page: Page, link_text) -> None:
             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)
+            nhs_no_df = (
+                extract_nhs_no_from_pdf(file)
+                if file.endswith(".pdf") and get_subjects_from_pdf
+                else None
+            )
             os.remove(file)  # Deletes the file after extracting the necessary data
     except Exception as e:
         pytest.fail(f"No retrieve button available to click: {str(e)}")
@@ -137,6 +137,8 @@ def prepare_and_print_batch(page: Page, link_text) -> None:
     except Exception as e:
         pytest.fail(f"Batch successfully archived message is not shown: {str(e)}")
 
+    return nhs_no_df
+
 
 def check_batch_in_archived_batch_list(page: Page, link_text) -> None:
     """

utils/pdf_reader.py

@@ -0,0 +1,19 @@
+from pypdf import PdfReader
+import pandas as pd
+
+
+def extract_nhs_no_from_pdf(file: str) -> pd.DataFrame:
+    reader = PdfReader(file)
+    nhs_no_df = pd.DataFrame(columns=["subject_nhs_number"])
+    # For loop looping through all pages of the file to find the NHS Number
+    for i, pages in enumerate(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()])
+                    nhs_no_df.loc[i] = [nhs_no]
+    return nhs_no_df