working on scenario 1 surveillance

Author: mepr1

classes/event/event_status_type.py

@@ -1032,6 +1032,7 @@ class EventStatusType(Enum):
     U81 = (11283, "U81", "Kit Returned and Logged (Technical Fail; Weak Positive")
     U97 = (11284, "U97", "Weak Positive, Waiting for Screening Centre Assistance")
     U98 = (11285, "U98", "Weak Positive, Waiting for Programme Hub Assistance")
+    X500=(20100, "X500", "Selected For Surveillance")
 
     def __init__(self, valid_value_id: int, allowed_value: str, description: str):
         """

classes/repositories/subject_repository.py

@@ -1,3 +1,4 @@
+from ast import Str
 import logging
 from typing import Optional
 from classes.subject.pi_subject import PISubject
@@ -280,3 +281,42 @@ def there_is_letter_batch_for_subject(
             logging.info(
                 f"[DB ASSERTION Passed] Subject {nhs_no} does not have a {letter_batch_code} - {letter_batch_title} batch"
             )
+
+
+    def get_early_subject_to_be_invited_for_surveillance(
+        self, screening_centre_id: int, s_due_count_date: str
+    ) -> Optional[str]:
+        """
+        Finds a subject in the DB who is due to be invited for surveillance early.
+        Args:
+            screening_centre_id (int): The screening centre ID.
+            s_due_count_date (str): The due count date in 'DD/MM/YYYY' format.
+        Returns:
+            str: The NHS number of a subject that matches the provided criteria, or None if not found.
+        """
+        sql_query = """SELECT ss.subject_nhs_number
+            FROM screening_subject_t ss
+            INNER JOIN sd_contact_t c ON ss.subject_nhs_number = c.nhs_number
+            WHERE c.responsible_sc_id = :screeningCentreId
+            AND c.deduction_reason IS NULL
+            AND c.date_of_death IS NULL
+            AND TRUNC (ss.surveillance_screen_due_date) <= TO_DATE(:sDueCountDate, 'DD/MM/YYYY')
+            AND ss.screening_status_id = 4006
+            AND c.date_of_death IS NULL
+            AND NOT EXISTS (
+                SELECT 1
+                FROM ep_subject_episode_t ep
+                WHERE ep.screening_subject_id = ss.screening_subject_id
+                AND ep.episode_status_id IN (11352, 11354)
+            )
+            ORDER BY TRUNC(ss.surveillance_screen_due_date)
+            FETCH FIRST 1 ROW ONLY
+        """
+        bind_vars = {
+            "screeningCentreId": screening_centre_id,
+            "sDueCountDate": s_due_count_date,
+        }
+        df = self.oracle_db.execute_query(sql_query, bind_vars)
+        if df.empty:
+            return None
+        return df["subject_nhs_number"].iloc[0]

pages/base_page.py

@@ -54,6 +54,7 @@ def __init__(self, page: Page):
             "#ntshPageTitle"
         )
         self.main_menu_header = self.page.locator("#ntshPageTitle")
+        self.surveillance = self.page.get_by_text("Surveillance", exact=True)
 
     def click_main_menu_link(self) -> None:
         """Click the Base Page 'Main Menu' link if it is visible."""
@@ -121,6 +122,10 @@ def click_help_link(self) -> None:
         """Click the Base Page 'Help' link."""
         self.click(self.help_link)
 
+    def click_surveillance_link(self) -> None:
+        """Click the Surveillance Page link."""
+        self.click(self.surveillance)    
+
     def bowel_cancer_screening_system_header_is_displayed(self) -> None:
         """Asserts that the Bowel Cancer Screening System header is displayed."""
         expect(self.bowel_cancer_screening_system_header).to_contain_text(

pages/surveillance/produce_healthcheck_forms.py

@@ -0,0 +1,42 @@
+from playwright.sync_api import Page
+from pages.base_page import BasePage
+
+
+class ProduceHealthCheckPage(BasePage):
+    """Page object for navigating to and interacting with the Produce Healthcheck Forms section."""
+
+    def __init__(self, page: Page):
+        super().__init__(page)
+
+        # Locators
+        self.surveillance_due_count_volume = self.page.locator(
+            'input[name="btnChangeVolume"]'
+        )
+        self.surveillance_due_count_volume_textbox = self.page.locator('[name="txtVolume"]')
+        self.recalculate_button = self.page.locator('[name="btnRecalc"]')
+        self.surveillance_due_count_date = self.page.locator('[name="txtDueDateTo"]')
+        self.health_check_forms=self.page.get_by_role('button', name='Generate HealthCheck Forms')
+
+    def click_on_surveillance_due_count_volume_button(self):
+        """Clicks on the Surveillance Due Count Volume Change button."""
+        self.click(self.surveillance_due_count_volume)
+
+    def fill_volume_in_surveillance_due_count_volume_textbox(self, volume: int):
+        """Enters a volume in the Surveillance Due Count Volume textbox.
+
+        Args:
+            volume (str): The volume to enter in the textbox.
+        """
+        self.surveillance_due_count_volume_textbox.fill(str(volume))
+    def click_on_recalculate_button(self):
+        """Clicks on the Recalculate button."""
+        self.click(self.recalculate_button)
+
+    def get_surveillance_due_count_date(self):
+        """Gets the value from the Surveillance Due Count Date textbox.
+        """
+        return self.surveillance_due_count_date.input_value()
+
+    def click_on_generate_healthcheck_forms_button(self):
+        """Clicks on the Generate HealthCheck Forms button."""
+        self.safe_accept_dialog(self.health_check_forms)

pages/surveillance/surveillance_page.py

@@ -0,0 +1,18 @@
+from playwright.sync_api import Page
+from pages.base_page import BasePage
+
+
+class SurveillancePage(BasePage):
+    """Page object for navigating to and interacting with the Surveillance Pagesection."""
+
+    def __init__(self, page: Page):
+        super().__init__(page)
+
+        # Locators
+        self.produce_healthcheck_forms = self.page.get_by_role('link', name='Produce Healthcheck Forms')
+        
+
+    def navigate_to_produce_healthcheck_forms(self):
+        """Navigates to the Produce Healthcheck Forms section."""
+        self.click(self.produce_healthcheck_forms)
+        

pytest.ini

@@ -55,4 +55,5 @@ markers =
     spine_retrieval_search_tests: tests that are part of subject spine retrieval demographics
     hub_user_tests: tests that are part of the hub user test suite
     fobt_regression_tests: tests that are part of the fobt regression test suite
+    survelliance_regression_tests: tests that are part of the survelliance regression test suite
     lynch_self_referral_tests: tests that are part of the lynch self referral test suite

tests/regression/regression_tests/survelliance_regression_tests/test_surveillance_scenario_1.py

@@ -0,0 +1,126 @@
+import pytest
+from playwright.sync_api import Page
+from classes.repositories.subject_repository import SubjectRepository
+from conftest import general_properties
+from pages.base_page import BasePage
+from pages.surveillance import produce_healthcheck_forms
+from pages.surveillance.produce_healthcheck_forms import ProduceHealthCheckPage
+from pages.surveillance.surveillance_page import SurveillancePage
+from utils import screening_subject_page_searcher
+from utils.generate_healthcheck_forms import generate_healthcheck_forms
+from utils.sspi_change_steps import SSPIChangeSteps
+from utils.user_tools import UserTools
+from utils.oracle.subject_creation_util import CreateSubjectSteps
+from utils.subject_assertion import subject_assertion
+from utils.call_and_recall_utils import CallAndRecallUtils
+import logging
+from utils.batch_processing import batch_processing
+from utils.fit_kit import FitKitGeneration, FitKitLogged
+from pages.logout.log_out_page import LogoutPage
+from datetime import datetime
+from pages.surveillance.produce_healthcheck_forms import ProduceHealthCheckPage
+from classes.repositories.subject_repository import SubjectRepository
+from utils.oracle.oracle_specific_functions.organisation_parameters import (
+    set_org_parameter_value,
+    check_parameter,
+)
+
+@pytest.mark.wip
+@pytest.mark.vpn_required
+@pytest.mark.regression
+@pytest.mark.survelliance_regression_tests
+def test_scenario_1(page: Page, general_properties: dict) -> None:
+    """
+        Scenario: 1: Discharge for clinical decision (GP letter required)
+
+    X500-X505-X600-X610-X615-X641-X600-X610-X615-X650-X390-X379-C203 [SSCL28b] X900-X600-X610-X2-X610-X615-X650-X382-X79-C203 [SSCL25a]
+
+    This scenario takes both an in-age and an over-age surveillance subject from invitation through to episode closure on X379 - discharge for clinical reason, GP letter required.  The scenario includes both DNA and reschedule of the SSP appointment.  It also includes a reopen, and checks that the episode could be postponed at most points during this pathway.
+
+    Because we cannot know if we will be inviting a subject who has had previous FOBT episodes, or only a bowel scope episode, it is impossible to check if they will be set to Call or Recall; we can only check that they are not longer in Surveillance status.
+
+    Note: parameter 82 controls whether or not a GP letter is required when a patient is discharged from Surveillance as a result of a clinical decision.  It actually defaults to Y, but it's set at SC level in the scenario to be sure it holds the correct value.  As a parameter can't be set with immediate effect through the screens, the scenario uses a direct database update to do this.
+
+
+        Scenario summary:
+        >Run surveillance invitations for 1 subject > X500 (3.1)
+    > SSPI update changes subject to in-age
+    > Process X500 letter batch > X505 (3.1)
+    > Record patient contact – contacted, SSP appointment > X600 (3.1)
+    > Book SSP appointment from report > X610 (3.3)
+    > Process X610 letter batch > X615 (3.3)
+    > Patient DNA SSP appointment > X641 (3.3)
+    > Choose to book SSP appointment > X600 (3.3)
+    > Book SSP appointment > X610 (3.3)
+    > Process X610 letter batch > X615 (3.3)
+    > Attend SSP appointment > X650 (3.3)
+    > Record discharge, clinical decision, GP letter required > X390 (3.4)
+    > Process X390 letter batch > X379 > C203 (3.4)
+    > Check recall [SSCL28b]
+    > Reopen episode for correction > X900 (3.1)
+    > Record patient contact – contacted, SSP appointment > X600 (3.1)
+    > Book SSP appointment from subject summary > X610 (3.3)
+    > Reschedule SSP appointment > X2 > X610 (3.3)
+    > Process X610 letter batch > X615 (3.3)
+    > Attend SSP appointment > X650 (3.3)
+    > SSPI update changes subject to over-age
+    > Record discharge, clinical decision, GP letter required > X382 (3.4)
+    > Process X382 letter batch > X79 > C203 (3.4)
+    > Check recall [SSCL25a]
+
+    """
+    # Given I log in to BCSS "England" as user role "Screening Centre Manager at BCS001"
+    user_role = UserTools.user_login(
+        page, "Screening Centre Manager at BCS001", return_role_type=True
+    )
+    if user_role is None:
+        raise ValueError("User cannot be assigned to a UserRoleType")
+
+    #    When I run surveillance invitations for 1 subject
+
+    nhs_no = generate_healthcheck_forms(page, general_properties)
+    # Then my subject has been updated as follows:
+
+    criteria = {
+        "latest episode status": "Open",
+        "latest episode type": "Surveillance",
+        "latest event status": "X500 Selected For Surveillance",
+        "responsible screening centre code": "User's screening centre",
+        "subject has unprocessed sspi updates": "No",
+        "subject has user dob updates": "No",
+    }
+
+    subject_assertion(nhs_no, criteria,user_role)
+    # And there is a "X500" letter batch for my subject with the exact title "Surveillance Selection"
+    SubjectRepository().there_is_letter_batch_for_subject(
+        nhs_no, "X500", "Surveillance Selection"
+    )
+    # Then Comment: NHS number    logging.info(f"Surveillance Scenario  NHS Number: {nhs_number}")
+    # When I set the value of parameter 82 to "Y" for my organisation with immediate effect
+    org_id = general_properties["eng_screening_centre_id"]
+    set_org_parameter_value(82, "Y", org_id)
+    # When I receive an SSPI update to change their date of birth to "72" years old
+    SSPIChangeSteps().sspi_update_to_change_dob_received(nhs_no, 72)
+    # Then my subject has been updated as follows:
+    subject_assertion(nhs_no, {"subject age": "72"})
+    # When I process the open "X500" letter batch for my subject
+    batch_processing(
+        page,
+        batch_type="X500",
+        batch_description="Surveillance Selection",
+        latest_event_status="X505 HealthCheck Form Printed",
+    )
+    # Then my subject has been updated as follows:
+    subject_assertion(
+        nhs_no,
+        {
+            "latest event status": "X505 HealthCheck Form Printed",
+        },
+    )
+    # When I view the subject
+    screening_subject_page_searcher.navigate_to_subject_summary_page(
+        page=page, nhs_no=nhs_no
+    )
+    # And I select the advance episode option for "Record Contact with Patient"
+    # SurveillancePage(page).select_advance_episode_option("Record Contact with Patient")
+

utils/generate_healthcheck_forms.py

@@ -0,0 +1,24 @@
+import logging
+from classes.repositories.subject_repository import SubjectRepository
+from pages.base_page import BasePage
+from pages.surveillance.surveillance_page import SurveillancePage
+from pages.surveillance.produce_healthcheck_forms import ProduceHealthCheckPage
+from logging import Logger
+
+
+def generate_healthcheck_forms(page, general_properties):
+    BasePage(page).click_surveillance_link()
+    SurveillancePage(page).navigate_to_produce_healthcheck_forms()
+    ProduceHealthCheckPage(page).click_on_surveillance_due_count_volume_button()
+    ProduceHealthCheckPage(page).fill_volume_in_surveillance_due_count_volume_textbox(1)
+    ProduceHealthCheckPage(page).click_on_recalculate_button()
+    subject_repo = SubjectRepository()
+    screening_centre_id = general_properties["eng_screening_centre_id"]
+    s_due_count_date = ProduceHealthCheckPage(page).get_surveillance_due_count_date()
+    nhs_number = subject_repo.get_early_subject_to_be_invited_for_surveillance(
+        screening_centre_id, s_due_count_date
+    )
+    assert nhs_number is not None
+    logging.info(f"Early subject NHS number: {nhs_number}")
+    ProduceHealthCheckPage(page).click_on_generate_healthcheck_forms_button()
+    return nhs_number

utils/oracle/subject_selection_query_builder.py

@@ -468,7 +468,7 @@ def _dispatch_criteria_key(self, user: "User", subject: "Subject") -> None:
             case SubjectSelectionCriteriaKey.LATEST_EPISODE_SUB_TYPE:
                 self._add_criteria_latest_episode_sub_type()
             case SubjectSelectionCriteriaKey.LATEST_EPISODE_STATUS:
-                self._add_criteria_latest_episode_status()
+                self._add_criteria_latest_episode_status()				   
 
             case SubjectSelectionCriteriaKey.LATEST_EPISODE_STATUS_REASON:
                 self._add_criteria_latest_episode_status_reason()
@@ -718,6 +718,7 @@ def _dispatch_criteria_key(self, user: "User", subject: "Subject") -> None:
                 self._add_extra_column_to_select_statement()
             case SubjectSelectionCriteriaKey.ADD_JOIN_TO_FROM_STATEMENT:
                 self._add_extra_join_to_from_statement()
+            
             # ------------------------------------------------------------------------
             # 🛑 Fallback: Unmatched Criteria Key
             # ------------------------------------------------------------------------
@@ -2966,7 +2967,7 @@ def _add_criteria_subject_screening_centre_code(self, user: "User"):
         sc_code = None
 
         try:
-            option = SubjectScreeningCentreCode.by_description(
+            option = SubjectScreeningCentreCode.by_description_case_insensitive(
                 self.criteria_value.lower()
             )
             match option:
@@ -2979,9 +2980,9 @@ def _add_criteria_subject_screening_centre_code(self, user: "User"):
                     | SubjectScreeningCentreCode.USER_SC
                     | SubjectScreeningCentreCode.USER_ORGANISATION
                 ):
-                    if user.organisation is None or user.organisation.id is None:
-                        raise ValueError("User organisation or organisation_id is None")
-                    sc_code = user.organisation.id
+                    if user.organisation is None or user.organisation.code is None:
+                        raise ValueError("User organisation or organisation_code is None")
+                    sc_code = user.organisation.code
                 case _:
                     raise SelectionBuilderException(
                         self.criteria_key_name, self.criteria_value