Feature/bcss 20230 compartment 4 utils

pages/base_page.py

@@ -1,4 +1,4 @@
-from playwright.sync_api import Page, expect, Locator
+from playwright.sync_api import Page, expect, Locator, Dialog
 import logging
 
 
@@ -167,3 +167,25 @@ def click(self, locator: Locator) -> None:
                 f"Failed to click element with error: {locatorClickError}, trying again..."
             )
             locator.click()
+
+    def _accept_dialog(self, dialog: Dialog) -> None:
+        """
+        This method is used to accept dialogs
+        If it has already been accepted then it is ignored
+        """
+        try:
+            dialog.accept()
+        except Exception:
+            logging.warning("Dialog already handled")
+
+    def safe_accept_dialog(self, locator: Locator) -> None:
+        """
+        Safely accepts a dialog triggered by a click, avoiding the error:
+        playwright._impl._errors.Error: Dialog.accept: Cannot accept dialog which is already handled!
+        If no dialog appears, continues without error.
+        """
+        self.page.once("dialog", self._accept_dialog)
+        try:
+            self.click(locator)
+        except Exception as e:
+            logging.error(f"Click failed: {e}")

pages/gfobt_test_kits/gfobt_test_kit_logging_page.py

@@ -2,7 +2,7 @@
 from pages.base_page import BasePage
 
 
-class TestKitLogging(BasePage):
+class GFOBTTestKitLogging(BasePage):
     def __init__(self, page: Page):
         super().__init__(page)
         self.page = page

pages/gfobt_test_kits/gfobt_test_kit_quality_control_reading_page.py

@@ -2,7 +2,7 @@
 from pages.base_page import BasePage
 
 
-class TestKitQualityControlReading(BasePage):
+class GFOBTTestKitQualityControlReading(BasePage):
     def __init__(self, page: Page):
         super().__init__(page)
         self.page = page

pages/screening_practitioner_appointments/book_appointment_page.py

@@ -1,5 +1,6 @@
 from playwright.sync_api import Page, expect
 from pages.base_page import BasePage
+from utils.table_util import TableUtils
 
 
 class BookAppointmentPage(BasePage):
@@ -9,24 +10,24 @@ def __init__(self, page: Page):
         # Book Appointment - page locators
         self.screening_center_dropdown = page.locator("#UI_NEW_SCREENING_CENTRE")
         self.site_dropdown = page.locator("#UI_NEW_SITE")
-        self.day_with_available_slots = page.locator(
-            'input.twoColumnCalendar[style*="background-color: rgb(102, 255, 153);"]'
-        ).last
         self.appointment_time_radio_button = page.locator(
             page.get_by_role("radio", name="UI_NEW_SLOT_SELECTION_ID")
         )
         self.save_button = page.get_by_role("button", name="Save")
+        self.appointments_table = TableUtils(self.page, "#displayRS")
+        self.current_month_displayed = self.page.locator("#MONTH_AND_YEAR")
+
+        self.appointment_cell_locators = self.page.locator("input.twoColumnCalendar")
+        self.available_background_colour = "rgb(102, 255, 153)"
+        self.some_available_background_colour = "rgb(255, 220, 144)"
 
     def select_screening_centre_dropdown_option(self, screening_centre: str) -> None:
         self.screening_center_dropdown.select_option(label=screening_centre)
 
-    def select_site_dropdown_option(self, screening_site: str) -> None:
+    def select_site_dropdown_option(self, screening_site: str | list) -> None:
         self.site_dropdown.select_option(label=screening_site)
         self.site_dropdown.press("Enter")
 
-    def choose_day_with_available_slots(self) -> None:
-        self.click(self.day_with_available_slots)
-
     def choose_appointment_time(self) -> None:
         self.appointment_time_radio_button.check()
 
@@ -35,3 +36,6 @@ def click_save_button(self) -> None:
 
     def appointment_booked_confirmation_is_displayed(self, message: str) -> None:
         expect(self.page.get_by_text(message)).to_be_visible()
+
+    def get_current_month_displayed(self) -> str:
+        return self.current_month_displayed.text_content()

pages/screening_practitioner_appointments/colonoscopy_assessment_appointments_page.py

@@ -8,14 +8,22 @@ def __init__(self, page: Page):
         self.page = page
         # Colonoscopy Assessment Appointments - page locators
         self.page_header = self.page.locator("#page-title")
+        self.page_header_with_title = self.page.locator(
+            "#page-title",
+            has_text="Patients that Require Colonoscopy Assessment Appointments",
+        )
         self.nhs_number_filter_text_field = self.page.locator("#nhsNumberFilter")
 
     def verify_page_header(self) -> None:
         expect(self.page_header).to_contain_text(
             "Patients that Require Colonoscopy Assessment Appointments"
         )
 
+    def wait_for_page_header(self) -> None:
+        self.page_header_with_title.wait_for()
+
     def filter_by_nhs_number(self, nhs_number: str) -> None:
+        self.click(self.nhs_number_filter_text_field)
         self.nhs_number_filter_text_field.fill(nhs_number)
         self.nhs_number_filter_text_field.press("Enter")
 

pages/screening_practitioner_appointments/screening_practitioner_appointments.py

@@ -34,3 +34,6 @@ def go_to_view_appointments_page(self) -> None:
 
     def go_to_patients_that_require_page(self) -> None:
         self.click(self.patients_that_require_page)
+
+    def go_to_set_availability_page(self) -> None:
+        self.click(self.set_availability_link)

pytest.ini

@@ -32,4 +32,6 @@ markers =
     compartment1: only for compartment 1
     compartment2: only for compartment 2
     compartment3: only for compartment 3
+    compartment4: only for compartment 4
+    compartment1_plan_creation: to run the plan creation for compartment 1
     vpn_required: for tests that require a VPN connection

tests/smokescreen/bcss_smokescreen_tests.properties

@@ -26,14 +26,11 @@
 # ----------------------------------
 #   compartment 4
 # ----------------------------------
-#    c4_eng_org_id=23159
-#    c4_eng_centre_name=Wolverhampton
-#    c4_eng_site_name=Royal Hospital
-#    c4_eng_practitioner_id=243
-#    c4_eng_practitioner_name=Arvind, Vanaja
-#    c4_eng_practitioner_name=Macarena, Gave  #use this for anonymised
-#    c4_eng_practitioner_id_for_rebooking=870
-#    c4_eng_practitioner_name_for_rebookings=Burns, Harriet
+    c4_eng_weeks_to_make_available = 6
+    c4_eng_centre_name=BCS001 - Wolverhampton Bowel Cancer Screening Centre
+    c4_eng_site_name1=THE ROYAL HOSPITAL (WOLVERHAMPTON)
+    c4_eng_site_name2=The Royal Hospital (Wolverhampton)
+    c4_eng_practitioner_name=Astonish, Ethanol
 
 # ----------------------------------
 #   compartment 6
@@ -76,7 +73,7 @@
 #   compartment 4
 # ----------------------------------
 #    # note there is a max of 16 slots available each day
-#    c4_eng_number_of_appointments_to_book=6
+    c4_eng_number_of_appointments_to_book=6
 
 # ----------------------------------
 #   compartment 5

tests/smokescreen/test_compartment_4.py

@@ -28,6 +28,9 @@
 from utils.calendar_picker import CalendarPicker
 from utils.batch_processing import batch_processing
 from datetime import datetime
+from utils.oracle.oracle_specific_functions import get_subjects_for_appointments
+from utils.nhs_number_tools import NHSNumberTools
+import logging
 
 
 @pytest.fixture
@@ -48,67 +51,92 @@ def test_compartment_4(page: Page, smokescreen_properties: dict) -> None:
     Finally it processes the necessary batches to send out the letters and checks the subjects satus has been updated to what is expected
     """
 
-    # Add method of getting test data using the query below. To remove once subject retrieval logic is created
-    """select tk.kitid, ss.subject_nhs_number, se.screening_subject_id
-    from tk_items_t tk
-    inner join ep_subject_episode_t se on se.screening_subject_id = tk.screening_subject_id
-    inner join screening_subject_t ss on ss.screening_subject_id = se.screening_subject_id
-    inner join sd_contact_t c on c.nhs_number = ss.subject_nhs_number
-    where se.latest_event_status_id = 11132
-    and tk.logged_in_flag = 'Y'
-    and se.episode_status_id = 11352
-    and ss.screening_status_id != 4008
-    and tk.logged_in_at = 23159
-    and c.hub_id = 23159
-    and tk.tk_type_id = 2
-    and tk.datestamp > add_months(sysdate,-24)
-    order by ss.subject_nhs_number desc"""
+    subjects_df = get_subjects_for_appointments(
+        smokescreen_properties["c4_eng_number_of_appointments_to_book"]
+    )
 
+    logging.info(
+        f"Compartment 4 - Setting up appointments for {smokescreen_properties["c4_eng_weeks_to_make_available"]} Weeks"
+    )
     UserTools.user_login(page, "Screening Centre Manager at BCS001")
     BasePage(page).go_to_screening_practitioner_appointments_page()
     ScreeningPractitionerAppointmentsPage(page).go_to_set_availability_page()
     SetAvailabilityPage(page).go_to_practitioner_availability_page()
     PractitionerAvailabilityPage(page).select_site_dropdown_option(
-        "THE ROYAL HOSPITAL (WOLVERHAMPTON)"
+        smokescreen_properties["c4_eng_site_name1"]
     )
     PractitionerAvailabilityPage(page).select_practitioner_dropdown_option(
-        "Astonish, Ethanol"
+        smokescreen_properties["c4_eng_practitioner_name"]
     )
     PractitionerAvailabilityPage(page).click_calendar_button()
-    CalendarPicker(page).select_day(
-        datetime.today()
-    )  # This will make it so that we can only run this test once a day, or we need to restore the DB back to the snapshot
+    CalendarPicker(page).select_day(datetime.today())
     PractitionerAvailabilityPage(page).click_show_button()
     PractitionerAvailabilityPage(page).enter_start_time("09:00")
     PractitionerAvailabilityPage(page).enter_end_time("17:15")
     PractitionerAvailabilityPage(page).click_calculate_slots_button()
-    PractitionerAvailabilityPage(page).enter_number_of_weeks("6")
+    PractitionerAvailabilityPage(page).enter_number_of_weeks(
+        smokescreen_properties["c4_eng_weeks_to_make_available"]
+    )
     PractitionerAvailabilityPage(page).click_save_button()
     PractitionerAvailabilityPage(page).slots_updated_message_is_displayed(
-        "Slots Updated for 6 Weeks"
+        f"Slots Updated for {smokescreen_properties["c4_eng_weeks_to_make_available"]} Weeks"
     )
     Logout(page).log_out(close_page=False)
 
+    logging.info(
+        f"Compartment 4 - Booking {smokescreen_properties["c4_eng_number_of_appointments_to_book"]} subjects to appointments"
+    )
     ScreeningPractitionerAppointmentsPage(page).go_to_log_in_page()
     UserTools.user_login(page, "Hub Manager State Registered at BCS01")
+
     BasePage(page).go_to_screening_practitioner_appointments_page()
     ScreeningPractitionerAppointmentsPage(page).go_to_patients_that_require_page()
-    # Add for loop to loop x times (depends on how many we want to run it for) 70 - 79
-    ColonoscopyAssessmentAppointments(page).filter_by_nhs_number("999 205 6339")
-    ColonoscopyAssessmentAppointments(page).click_nhs_number_link("999 205 6339")
-    BookAppointmentPage(page).select_screening_centre_dropdown_option(
-        "BCS001 - Wolverhampton Bowel Cancer Screening Centre"
-    )
-    BookAppointmentPage(page).select_site_dropdown_option("Holly Hall Clinic (? km)")
-    BookAppointmentPage(page).choose_day_with_available_slots()
-    # page.locator("#UI_NEW_SLOT_SELECTION_ID_359119").check()
-    # Will be revisited as part of Utilities update
-    BookAppointmentPage(page).choose_appointment_time()
-    BookAppointmentPage(page).click_save_button()
-    BookAppointmentPage(page).appointment_booked_confirmation_is_displayed(
-        "Appointment booked"
-    )
 
+    for subject_num in range(
+        int(smokescreen_properties["c4_eng_number_of_appointments_to_book"])
+    ):
+        nhs_number = subjects_df["subject_nhs_number"].iloc[subject_num]
+        logging.info(f"Booking appointment for: {nhs_number}")
+
+        nhs_number_spaced = NHSNumberTools().spaced_nhs_number(nhs_number)
+        ColonoscopyAssessmentAppointments(page).filter_by_nhs_number(nhs_number)
+        ColonoscopyAssessmentAppointments(page).click_nhs_number_link(nhs_number_spaced)
+        BookAppointmentPage(page).select_screening_centre_dropdown_option(
+            smokescreen_properties["c4_eng_centre_name"]
+        )
+        BookAppointmentPage(page).select_site_dropdown_option(
+            [
+                f"{smokescreen_properties["c4_eng_site_name2"]} (? km)",
+                f"{smokescreen_properties["c4_eng_site_name2"]} (? km) (attended)",
+            ]
+        )
+
+        current_month_displayed = BookAppointmentPage(
+            page
+        ).get_current_month_displayed()
+        CalendarPicker(page).book_first_eligble_appointment(
+            current_month_displayed,
+            BookAppointmentPage(page).appointment_cell_locators,
+            [
+                BookAppointmentPage(page).available_background_colour,
+                BookAppointmentPage(page).some_available_background_colour,
+            ],
+        )
+        BookAppointmentPage(page).appointments_table.click_first_input_in_column(
+            "Appt/Slot Time"
+        )
+        BasePage(page).safe_accept_dialog(BookAppointmentPage(page).save_button)
+        try:
+            BookAppointmentPage(page).appointment_booked_confirmation_is_displayed(
+                "Appointment booked"
+            )
+            logging.info(f"Appointment successfully booked for: {nhs_number}")
+        except Exception as e:
+            pytest.fail(f"Appointment not booked successfully: {e}")
+        BasePage(page).click_back_button()
+    ColonoscopyAssessmentAppointments(page).wait_for_page_header()
+
+    logging.info("Compartment 4 - Sending out appointment invitations")
     batch_processing(
         page,
         "A183",

tests/test_gfobt_test_kits_page.py

@@ -3,9 +3,9 @@
 
 from pages.base_page import BasePage
 from pages.gfobt_test_kits.gfobt_test_kits_page import GFOBTTestKits
-from pages.gfobt_test_kits.gfobt_test_kit_logging_page import TestKitLogging
+from pages.gfobt_test_kits.gfobt_test_kit_logging_page import GFOBTTestKitLogging
 from pages.gfobt_test_kits.gfobt_test_kit_quality_control_reading_page import (
-    TestKitQualityControlReading,
+    GFOBTTestKitQualityControlReading,
 )
 from pages.gfobt_test_kits.gfobt_view_test_kit_result import ViewTestKitResult
 from pages.gfobt_test_kits.gfobt_create_qc_kit_page import (
@@ -27,6 +27,7 @@ def before_each(page: Page):
     # Go to gFOBT test kits page
     BasePage(page).go_to_gfobt_test_kits_page()
 
+
 @pytest.mark.smoke
 def test_gfobt_test_kit_page_navigation(page: Page) -> None:
     """
@@ -35,12 +36,12 @@ def test_gfobt_test_kit_page_navigation(page: Page) -> None:
     """
     # Test kit logging page opens as expected
     GFOBTTestKits(page).go_to_test_kit_logging_page()
-    TestKitLogging(page).verify_test_kit_logging_title()
+    GFOBTTestKitLogging(page).verify_test_kit_logging_title()
     BasePage(page).click_back_button()
 
     # Test kit reading page opens as expected
     GFOBTTestKits(page).go_to_test_kit_reading_page()
-    TestKitQualityControlReading(page).verify_test_kit_logging_tile()
+    GFOBTTestKitQualityControlReading(page).verify_test_kit_logging_tile()
     BasePage(page).click_back_button()
 
     # View test kit result page opens as expected

utils/batch_processing.py

@@ -1,7 +1,12 @@
 from pages.base_page import BasePage
-from pages.communication_production.communications_production_page import CommunicationsProduction
+from pages.communication_production.communications_production_page import (
+    CommunicationsProduction,
+)
 from pages.communication_production.manage_active_batch_page import ManageActiveBatch
-from pages.communication_production.batch_list_page import ActiveBatchList, ArchivedBatchList
+from pages.communication_production.batch_list_page import (
+    ActiveBatchList,
+    ArchivedBatchList,
+)
 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
@@ -93,7 +98,7 @@ def prepare_and_print_batch(page: Page, link_text) -> None:
     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()
+    ManageActiveBatch(page).reprepare_batch_text.wait_for(timeout=60000)
 
     # This loops through each Retrieve button and clicks each one
     retrieve_button_count = 0
@@ -119,8 +124,9 @@ def prepare_and_print_batch(page: Page, link_text) -> None:
             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()
+            ManageActiveBatch(page).safe_accept_dialog(
+                ManageActiveBatch(page).confirm_button.nth(0)
+            )
     except Exception as e:
         pytest.fail(f"No confirm printed button available to click: {str(e)}")