Merge branch 'main' into feature/BCSS-20586-selenium-to-playwright-organisation

Author: mepr1

docs/utility-guides/Oracle.md

@@ -1,13 +1,18 @@
 # Utility Guide: Oracle
 
-The Oracle Utility provides an easy way to interact with an Oracle database directly from your Playwright test suite.It can be used to run SQL queries or stored procedures on the Oracle database.
+The Oracle Utility provides an easy way to interact with an Oracle database directly from your Playwright test suite. It can be used to run SQL queries or stored procedures on the Oracle database.
 
 ## When and Why to Use the Oracle Utility
 
-You might need to use this utility in scenarios such as:
+You might use this utility for:
 
-- To run SQL queries or stored procedures on the Oracle database.
-- Verifying that data is correctly written to or updated in the database after a workflow is completed in your application.
+- Running SQL queries or stored procedures on the Oracle database
+
+- Validating application workflows against live database records
+
+- Creating test data dynamically via helper functions
+
+- Verifying data integrity after system events
 
 ## Table of Contents
 
@@ -26,6 +31,10 @@ You might need to use this utility in scenarios such as:
 
 To use the Oracle Utility, import the 'OracleDB' class into your test file and then call the OracleDB methods from within your tests, as required.
 
+```python
+from utils.oracle.oracle import OracleDB
+```
+
 ## Required arguments
 
 The functions in this class require different arguments.<br>
@@ -42,6 +51,7 @@ The docstrings also specify when arguments are optional, and what the default va
 - **execute_stored_procedure(self, `procedure_name`, params=None)**: Executes a named stored procedure with optional parameters.
 - **exec_bcss_timed_events(self, nhs_number_df)**: Runs the `bcss_timed_events` stored procedure for each NHS number provided in a DataFrame.
 - **get_subject_id_from_nhs_number(self, nhs_number)**: Retrieves the `subject_screening_id` for a given NHS number.
+- **create_subjects_via_sspi(...)**: Creates synthetic subjects using stored procedure `PKG_SSPI.p_process_pi_subject`
 
 For full implementation details, see utils/oracle/oracle.py.
 
@@ -68,6 +78,16 @@ def test_oracle_query() -> None:
 
 def run_stored_procedure() -> None:
     OracleDB().execute_stored_procedure("bcss_timed_events")
+
+def create_subjects_dynamically() -> None:
+OracleDB().create_subjects_via_sspi(
+    count=5,
+    screening_centre="BCS01",
+    base_age=60,
+    start_offset=-2,
+    end_offset=4,
+    nhs_start=9200000000
+)
 ```
 
 ---

pages/call_and_recall/generate_invitations_page.py

@@ -2,6 +2,13 @@
 from pages.base_page import BasePage
 import pytest
 import logging
+from utils.oracle.oracle import OracleDB
+from utils.oracle.subject_selection_query_builder import SubjectSelectionQueryBuilder
+from classes.user import User
+from classes.subject import Subject
+from utils.table_util import TableUtils
+
+DISPLAY_RS_SELECTOR = "#displayRS"
 
 
 class GenerateInvitationsPage(BasePage):
@@ -14,10 +21,10 @@ def __init__(self, page: Page):
         self.generate_invitations_button = self.page.get_by_role(
             "button", name="Generate Invitations"
         )
-        self.display_rs = self.page.locator("#displayRS")
+        self.display_rs = self.page.locator(DISPLAY_RS_SELECTOR)
         self.refresh_button = self.page.get_by_role("button", name="Refresh")
         self.planned_invitations_total = self.page.locator("#col8_total")
-        self.self_referrals_total = self.page.locator("#col9_total")
+        self.self_referrals_total = self.page.locator("#col5_total")
 
     def click_generate_invitations_button(self) -> None:
         """This function is used to click the Generate Invitations button."""
@@ -45,7 +52,7 @@ def wait_for_invitation_generation_complete(
         Every 5 seconds it refreshes the table and checks to see if the invitations have been generated.
         It also checks that enough invitations were generated and checks to see if self referrals are present
         """
-        self.page.wait_for_selector("#displayRS", timeout=5000)
+        self.page.wait_for_selector(DISPLAY_RS_SELECTOR, timeout=5000)
 
         if self.planned_invitations_total == "0":
             pytest.fail("There are no planned invitations to generate")
@@ -103,3 +110,132 @@ def wait_for_invitation_generation_complete(
         else:
             logging.warning("No S1 Digital Leaflet batch will be generated")
             return False
+
+    def wait_for_self_referral_invitation_generation_complete(
+        self, expected_minimum: int = 1
+    ) -> bool:
+        """
+        Waits until the invitations have been generated and checks that 'Self Referrals Generated' meets the expected threshold.
+
+        Args:
+            expected_minimum (int): Minimum number of self-referrals expected to be generated (default is 1)
+
+        Returns:
+            bool: True if threshold is met, False otherwise.
+        """
+        # Reuse the existing table completion logic — consider extracting this into a shared method later
+        timeout = 120000
+        wait_interval = 5000
+        elapsed = 0
+        logging.info(
+            "[WAIT] Waiting for self-referral invitation generation to complete"
+        )
+
+        while elapsed < timeout:
+            table_text = self.display_rs.text_content()
+            if table_text is None:
+                pytest.fail("Failed to retrieve table text content")
+
+            if "Failed" in table_text:
+                pytest.fail("Invitation has failed to generate")
+            elif "Queued" in table_text or "In Progress" in table_text:
+                self.click_refresh_button()
+                self.page.wait_for_timeout(wait_interval)
+                elapsed += wait_interval
+            else:
+                break
+
+        try:
+            expect(self.display_rs).to_contain_text("Completed")
+            logging.info(
+                f"[STATUS] Generation finished after {elapsed / 1000:.1f} seconds"
+            )
+        except Exception as e:
+            pytest.fail(f"[ERROR] Invitations not generated successfully: {str(e)}")
+
+        # Dynamically check 'Self Referrals Generated'
+        table_utils = TableUtils(self.page, DISPLAY_RS_SELECTOR)
+
+        try:
+            value_text = table_utils.get_footer_value_by_header(
+                "Self Referrals Generated"
+            )
+            value = int(value_text.strip())
+            logging.info(f"[RESULT] 'Self Referrals Generated' = {value}")
+        except Exception as e:
+            pytest.fail(f"[ERROR] Unable to read 'Self Referrals Generated': {str(e)}")
+
+        if value >= expected_minimum:
+            return True
+        else:
+            logging.warning(
+                f"[ASSERTION] Expected at least {expected_minimum}, but got {value}"
+            )
+            return False
+
+    def check_self_referral_subjects_ready(
+        self, search_scope: str, volume: str
+    ) -> None:
+        """
+        Asserts whether self-referral subjects are ready to invite.
+
+        Args:
+            search_scope (str): Either "currently" or "now" to determine when to assert
+            volume (str): Either "some" or "no" — expected number of self-referrals
+
+        Raises:
+            AssertionError if the count doesn't match expectation and search_scope is "now"
+        """
+        assert search_scope in (
+            "currently",
+            "now",
+        ), f"Invalid search_scope: '{search_scope}'"
+        assert volume in ("some", "no"), f"Invalid volume: '{volume}'"
+
+        # Confirm we're on the Generate Invitations page
+        self.verify_generate_invitations_title()
+
+        # Extract and clean the count
+        self_referrals_text = self.self_referrals_total.text_content()
+        if self_referrals_text is None:
+            pytest.fail("Failed to read self-referrals total")
+        else:
+            self_referrals_text = self_referrals_text.strip()
+
+        self_referrals_count = int(self_referrals_text)
+
+        # Determine if condition is met
+        condition_met = (
+            self_referrals_count > 0 if volume == "some" else self_referrals_count == 0
+        )
+
+        logging.debug(f"[DEBUG] Self-referral subject count = {self_referrals_count}")
+
+        if search_scope == "now":
+            assert (
+                condition_met
+            ), f"Expected {volume.upper()} self-referral subjects, but got {self_referrals_count}"
+        logging.info(
+            f"[SELF-REFERRAL CHECK] scope='{search_scope}' | expected='{volume}' | actual={self_referrals_count}"
+        )
+
+    def get_self_referral_eligible_subject(self, user: User, subject: Subject) -> str:
+        criteria = {
+            "screening status": "Inactive",
+            "subject age": ">= 75",
+            "has GP practice": "Yes - active",
+            "subject hub code": "BCS02",
+        }
+
+        builder = SubjectSelectionQueryBuilder()
+        query, bind_vars = builder.build_subject_selection_query(
+            criteria, user, subject
+        )
+
+        oracle = OracleDB()
+        result = oracle.execute_query(query, bind_vars)
+
+        if result.empty:
+            raise RuntimeError("No eligible subject found")
+
+        return result.iloc[0]["subject_nhs_number"]