diff --git a/classes/subject_selection_criteria_key.py b/classes/subject_selection_criteria_key.py index 270a73b5..6f1c1954 100644 --- a/classes/subject_selection_criteria_key.py +++ b/classes/subject_selection_criteria_key.py @@ -276,6 +276,7 @@ class SubjectSelectionCriteriaKey(Enum): SUBJECT_HAS_DIAGNOSTIC_TESTS = ("subject has diagnostic tests", False, False) SUBJECT_HAS_EPISODES = ("subject has episodes", False, False) SUBJECT_HAS_EVENT_STATUS = ("subject has event status", False, False) + SUBJECT_IS_DUE_FOR_INVITE = ("subject is due for invite", False, False) SUBJECT_DOES_NOT_HAVE_EVENT_STATUS = ( "subject does not have event status", False, diff --git a/classes/temporary_address_type.py b/classes/temporary_address_type.py new file mode 100644 index 00000000..b645db87 --- /dev/null +++ b/classes/temporary_address_type.py @@ -0,0 +1,43 @@ +class TemporaryAddressType: + """ + Utility class for handling temporary address type values. + + Members: + YES: Represents a 'yes' value. + NO: Represents a 'no' value. + EXPIRED: Represents an 'expired' value. + CURRENT: Represents a 'current' value. + FUTURE: Represents a 'future' value. + + Methods: + from_description(description: str) -> str: + Returns the normalized value for a given description. + Raises ValueError if the description is not recognized. + """ + + YES = "yes" + NO = "no" + EXPIRED = "expired" + CURRENT = "current" + FUTURE = "future" + + _valid = {YES, NO, EXPIRED, CURRENT, FUTURE} + + @classmethod + def from_description(cls, description: str) -> str: + """ + Returns the normalized value for a given description. + + Args: + description (str): The input description to normalize. + + Returns: + str: 'yes', 'no', 'expired', 'current', 'future'. + + Raises: + ValueError: If the description is not recognized as 'yes', 'no', 'expired', 'current', 'future'. + """ + key = description.strip().lower() + if key not in cls._valid: + raise ValueError(f"Expected 'yes', 'no', 'expired', 'current', 'future', got: '{description}'") + return key diff --git a/pages/base_page.py b/pages/base_page.py index 2599d977..97c16d67 100644 --- a/pages/base_page.py +++ b/pages/base_page.py @@ -134,6 +134,9 @@ def main_menu_header_is_displayed(self) -> None: """ expect(self.main_menu__header).to_contain_text(self.main_menu_string) + def page_title_contains_text(self, text: str) -> None: + self.bowel_cancer_screening_page_title_contains_text(text) + def bowel_cancer_screening_page_title_contains_text(self, text: str) -> None: """Asserts that the page title contains the specified text. diff --git a/pages/reports/operational/subjects_to_be_invited_with_temporary_address_page.py b/pages/reports/operational/subjects_to_be_invited_with_temporary_address_page.py new file mode 100644 index 00000000..b570cae1 --- /dev/null +++ b/pages/reports/operational/subjects_to_be_invited_with_temporary_address_page.py @@ -0,0 +1,74 @@ +from enum import Enum +from playwright.sync_api import Page, expect +from pages.base_page import BasePage +from datetime import datetime +from utils.calendar_picker import CalendarPicker +from utils.nhs_number_tools import NHSNumberTools +from utils.table_util import TableUtils +import logging + + +class SubjectsToBeInvitedWithTemporaryAddressPage(BasePage): + """Subjects To Be Invited with Temporary Address Page locators, and methods for interacting with the page.""" + + def __init__(self, page: Page): + super().__init__(page) + self.page = page + + self.title = "Subjects To Be Invited with Temporary Address"; + self.nhs_filter = self.page.locator("#nhsNumberFilter") + + self.subjects_to_be_invited_with_temporary_address_report_page = self.page.get_by_role("link", name="Subjects To Be Invited with Temporary Address") + self.report_table = TableUtils(page, "#tobeinvitedtemporaryaddress") + + def go_to_page(self) -> None: + logging.info(f"Go to '{self.title}'") + ""f"Click the '{self.title}' link.""" + self.click(self.subjects_to_be_invited_with_temporary_address_report_page) + + def verify_page_title(self) -> None: + logging.info(f"Verify title as '{self.title}'") + ""f"Verifies that the {self.title} page title is displayed correctly.""" + self.page_title_contains_text(self.title) + + def filter_by_nhs_number(self, search_text: str) -> None: + logging.info(f"filter_by_nhs_number : {search_text}") + """Enter text in the NHS Number filter and press Enter""" + self.nhs_filter.fill(search_text) + self.nhs_filter.press("Enter") + + def assertRecordsVisible(self, nhs_no: str, expected_visible: bool) -> None: + logging.info(f"Attempting to check if records for {nhs_no} are visible : {expected_visible}") + + subject_summary_link = self.page.get_by_role( + "link", name = NHSNumberTools().spaced_nhs_number(nhs_no) + ) + + logging.info(f"subject_summary_link.is_visible() : {subject_summary_link.is_visible()}") + if ( + expected_visible != subject_summary_link.is_visible() + ): + raise ValueError( + f"Record for {NHSNumberTools().spaced_nhs_number(nhs_no)} does not have expected visibility : {expected_visible}. Was in fact {subject_summary_link.is_visible()}." + ) + + def filterByReviewed(self, filter_value: str) -> None: + logging.info(f"Filter reviewed column by : {filter_value}") + + filter_box = self.page.locator("#reviewedFilter") + filter_box.click() + filter_box.select_option(filter_value) + + def reviewSubject(self, nhs_no: str, reviewed: bool) -> None: + logging.info(f"Mark subject {nhs_no} as reviewed: {reviewed}") + + row = self.page.locator(f'tr:has-text("{NHSNumberTools().spaced_nhs_number(nhs_no)}")') + last_cell = row.get_by_role("cell").nth(-1) + checkbox = last_cell.get_by_role("checkbox") + checkbox.set_checked(reviewed) + + def click_subject_link(self, nhs_no: str) -> None: + subject_summary_link = self.page.get_by_role( + "link", name = NHSNumberTools().spaced_nhs_number(nhs_no) + ) + subject_summary_link.click() diff --git a/pages/screening_subject_search/subject_demographic_page.py b/pages/screening_subject_search/subject_demographic_page.py index 8fb26036..ebccdb7d 100644 --- a/pages/screening_subject_search/subject_demographic_page.py +++ b/pages/screening_subject_search/subject_demographic_page.py @@ -2,7 +2,7 @@ from pages.base_page import BasePage from datetime import datetime from utils.calendar_picker import CalendarPicker - +import logging class SubjectDemographicPage(BasePage): """Subject Demographic Page locators, and methods for interacting with the page.""" @@ -10,6 +10,9 @@ class SubjectDemographicPage(BasePage): def __init__(self, page: Page): super().__init__(page) self.page = page + + self.title = "Subject Demographic"; + # Subject Demographic - page filters self.forename_field = self.page.get_by_role("textbox", name="Forename") self.surname_field = self.page.get_by_role("textbox", name="Surname") @@ -54,6 +57,11 @@ def __init__(self, page: Page): "#UI_SUBJECT_ALT_POSTCODE_0" ) + def verify_page_title(self) -> None: + logging.info(f"Verify title as '{self.title}'") + ""f"Verifies that the {self.title} page title is displayed correctly.""" + self.page_title_contains_text(self.title) + def is_forename_filled(self) -> bool: """ Checks if the forename textbox contains a value. diff --git a/tests/regression/reports/operational/test_subjects_to_be_invited_with_temporary_address.py b/tests/regression/reports/operational/test_subjects_to_be_invited_with_temporary_address.py new file mode 100644 index 00000000..80c534ef --- /dev/null +++ b/tests/regression/reports/operational/test_subjects_to_be_invited_with_temporary_address.py @@ -0,0 +1,278 @@ +import logging +import pytest +from playwright.sync_api import Page, expect +from utils.user_tools import UserTools +from classes.user import User +from classes.subject import Subject +from pages.base_page import BasePage +from pages.reports.operational.subjects_to_be_invited_with_temporary_address_page import ( + SubjectsToBeInvitedWithTemporaryAddressPage +) +from pages.reports.reports_page import ReportsPage +from pages.screening_subject_search.subject_demographic_page import ( + SubjectDemographicPage, +) +from pages.screening_subject_search.subject_screening_summary_page import ( + SubjectScreeningSummaryPage, +) +from utils.oracle.oracle import OracleDB +from utils.oracle.subject_selection_query_builder import SubjectSelectionQueryBuilder +from utils.screening_subject_page_searcher import search_subject_episode_by_nhs_number +from datetime import datetime, timedelta + + +@pytest.mark.regression +@pytest.mark.reports_operational +def test_subject_not_on_report_if_not_due(page: Page) -> None: + """ + Test to check that a subject who is not due for invite is not in the report. + """ + logging.info( + "Starting test: Test to check that a subject who is not due for invite is not in the report." + ) + + criteria1 = { + "subject has temporary address": "current", + "subject hub code": "BCS01", + "subject is due for invite": "no" + } + + nhs_no = _db_search(criteria1) + + UserTools.user_login(page, "Hub Manager at BCS01") + _go_to_report_page(page) + + report = SubjectsToBeInvitedWithTemporaryAddressPage(page) + report.filter_by_nhs_number(nhs_no) + report.assertRecordsVisible(nhs_no, False) + +@pytest.mark.wip +@pytest.mark.regression +@pytest.mark.reports_operational +def test_subject_to_be_invited_without_temporary_address_at_this_hub_not_on_report(page: Page) -> None: + """ + Test to check that a subject who is due to be invited, but does not have a temporary address, and is at this hub, does not appear in the report. + """ + logging.info( + "Starting test: Check that a subject who is due to be invited, but does not have a temporary address, and is at this hub, does not appear in the report." + ) + + criteria = { + "subject has temporary address": "No", + "subject hub code": "BCS01", + "subject is due for invite": "yes" + } + + nhs_no = _db_search(criteria) + + UserTools.user_login(page, "Hub Manager State Registered at BCS01") + + BasePage(page).click_main_menu_link() + _go_to_report_page(page) + + report = SubjectsToBeInvitedWithTemporaryAddressPage(page) + report.filter_by_nhs_number(nhs_no) + report.assertRecordsVisible(nhs_no, False) + report.filter_by_nhs_number("") + + BasePage(page).click_main_menu_link() + _go_to_demographics_page_for_subject_from_menu(page, nhs_no) + _add_a_temp_address(page) + + BasePage(page).click_main_menu_link() + _go_to_report_page(page) + + report.filter_by_nhs_number(nhs_no) + report.assertRecordsVisible(nhs_no, True) + + +@pytest.mark.regression +@pytest.mark.reports_operational +def test_subject_to_be_invited_with_temporary_address_at_another_hub_not_on_report(page: Page) -> None: + """ + Test to check that a subject who is due to be invited, and does have a temporary address but is at another hub, does not appear in the report. + """ + logging.info( + "Starting test: Check that a subject who is due to be invited, and does have a temporary address but is at another hub, does not appear in the report." + ) + + criteria = { + "subject has temporary address": "No", + "subject hub code": "BCS01", + "subject is due for invite": "yes" + } + + nhs_no = _db_search(criteria) + + UserTools.user_login(page, "Hub Manager at BCS02") + + BasePage(page).click_main_menu_link() + _go_to_report_page(page) + + report = SubjectsToBeInvitedWithTemporaryAddressPage(page) + report.filter_by_nhs_number(nhs_no) + report.assertRecordsVisible(nhs_no, False) + + +@pytest.mark.regression +@pytest.mark.reports_operational +def test_subject_on_report_is_hidden_after_review(page: Page) -> None: + """ + Test to check that a subject who becomes reviewed is then hidden from the default view in the report. + """ + logging.info( + "Starting test: Check that a subject who becomes reviewed is then hidden from the default view in the report." + ) + + criteria = { + "subject has temporary address": "Current", + "subject hub code": "BCS01", + "subject is due for invite": "yes" + } + + nhs_no = _db_search(criteria) + + UserTools.user_login(page, "Hub Manager State Registered at BCS01") + _go_to_report_page(page) + + report = SubjectsToBeInvitedWithTemporaryAddressPage(page) + + logging.info("Limit report to show just our subject, make sure starting state is not-reviewed") + report.filterByReviewed("All") + report.filter_by_nhs_number(nhs_no) + report.assertRecordsVisible(nhs_no, True) + report.reviewSubject(nhs_no, False) + + logging.info("Only show Non-reviewed, review the subject, watch subject disappear") + report.filterByReviewed("No") + report.reviewSubject(nhs_no, True) + report.assertRecordsVisible(nhs_no, False) + + logging.info("Only show Reviewed, watch subject appear") + report.filterByReviewed("Yes") + report.assertRecordsVisible(nhs_no, True) + + report.filterByReviewed("All") + report.assertRecordsVisible(nhs_no, True) + + logging.info("Only show Reviewed, un-review the subject, watch subject disappear") + report.filterByReviewed("Yes") + report.reviewSubject(nhs_no, False) + report.assertRecordsVisible(nhs_no, False) + + logging.info("Only show Non-reviewed, watch subject appear") + report.filterByReviewed("No") + report.assertRecordsVisible(nhs_no, True) + + +@pytest.mark.regression +@pytest.mark.reports_operational +def test_subject_to_be_invited_with_temporary_address_at_this_hub_is_on_report(page: Page) -> None: + """ + Test to check that a subject who is due to be invited, and does have a temporary address, and is at this hub, does appear in the report. + """ + logging.info( + "Starting test: Check that a subject who is due to be invited, and does have a temporary address, and is at this hub, does appear in the report." + ) + + criteria = { + "subject has temporary address": "current", + "subject hub code": "BCS01", + "subject is due for invite": "yes" + } + + nhs_no = _db_search(criteria) + + UserTools.user_login(page, "Hub Manager State Registered at BCS01") + + BasePage(page).click_main_menu_link() + _go_to_report_page(page) + + report = SubjectsToBeInvitedWithTemporaryAddressPage(page) + report.filter_by_nhs_number(nhs_no) + report.assertRecordsVisible(nhs_no, True) + _go_to_demographics_page_for_subject_from_report(page, nhs_no) + _remove_a_temp_address(page) + + BasePage(page).click_main_menu_link() + _go_to_report_page(page) + + report.filter_by_nhs_number(nhs_no) + report.assertRecordsVisible(nhs_no, False) + + +def _db_search(criteria) -> None: + logging.info( + f"Starting _db_search: {criteria}" + ) + user = User() + subject = Subject() + + builder = SubjectSelectionQueryBuilder() + + query, bind_vars = builder.build_subject_selection_query( + criteria=criteria, user=user, subject=subject, subjects_to_retrieve=1 + ) + + df = OracleDB().execute_query(query, bind_vars) + nhs_no = df.iloc[0]["subject_nhs_number"] + + logging.info( + f"Identified {nhs_no}." + ) + return nhs_no + +def _go_to_report_page(page: Page) -> None: + BasePage(page).go_to_reports_page() + ReportsPage(page).go_to_operational_reports_page() + SubjectsToBeInvitedWithTemporaryAddressPage(page).go_to_page() + + SubjectsToBeInvitedWithTemporaryAddressPage(page).verify_page_title() + +def _go_to_demographics_page_for_subject_from_menu(page: Page, nhs_no: str) -> None: + BasePage(page).go_to_screening_subject_search_page() + search_subject_episode_by_nhs_number(page, nhs_no) + _go_to_demographics_page_for_subject(page) + +def _go_to_demographics_page_for_subject_from_report(page: Page, nhs_no: str) -> None: + SubjectsToBeInvitedWithTemporaryAddressPage(page).click_subject_link(nhs_no) + _go_to_demographics_page_for_subject(page) + +def _go_to_demographics_page_for_subject(page: Page) -> None: + SubjectScreeningSummaryPage(page).click_subject_demographics() + + SubjectDemographicPage(page).verify_page_title() + +def _add_a_temp_address(page: Page) -> None: + temp_address = { + "valid_from": datetime.today(), + "valid_to": datetime.today() + timedelta(days=100), + "address_line_1": "add line 1", + "address_line_2": "add line 2", + "address_line_3": "add line 3", + "address_line_4": "add line 4", + "address_line_5": "add line 5", + "postcode": "EX2 5SE", + } + _update_temp_address(page, temp_address) + +def _remove_a_temp_address(page: Page) -> None: + temp_address = { + "valid_from": None, + "valid_to": None, + "address_line_1": "", + "address_line_2": "", + "address_line_3": "", + "address_line_4": "", + "address_line_5": "", + "postcode": "", + } + _update_temp_address(page, temp_address) + +def _update_temp_address(page: Page, temp_address) -> None: + logging.info(f"Updating Temporary address : {temp_address}.") + demographic_page = SubjectDemographicPage(page) + + logging.info("Updating postcode as sometimes the existing address has a bad or blank value which might break the save") + demographic_page.fill_postcode_input("EX11AA") + demographic_page.update_temporary_address(temp_address) diff --git a/utils/oracle/subject_selection_query_builder.py b/utils/oracle/subject_selection_query_builder.py index c7b4efe1..eb5ff50f 100644 --- a/utils/oracle/subject_selection_query_builder.py +++ b/utils/oracle/subject_selection_query_builder.py @@ -57,6 +57,7 @@ from classes.prevalent_incident_status_type import PrevalentIncidentStatusType from classes.notify_event_status import NotifyEventStatus from classes.yes_no_type import YesNoType +from classes.temporary_address_type import TemporaryAddressType from classes.episode_sub_type import EpisodeSubType from classes.episode_status_type import EpisodeStatusType from classes.episode_status_reason_type import EpisodeStatusReasonType @@ -392,6 +393,8 @@ def _dispatch_criteria_key(self, user: "User", subject: "Subject") -> None: ) case SubjectSelectionCriteriaKey.BOWEL_SCOPE_DUE_DATE_REASON: self._add_criteria_bowel_scope_due_date_reason() + case SubjectSelectionCriteriaKey.SUBJECT_IS_DUE_FOR_INVITE: + self._add_criteria_subject_is_due_for_invite() # ------------------------------------------------------------------------ # ⛔ Cease & Manual Override Criteria # ------------------------------------------------------------------------ @@ -2382,27 +2385,39 @@ def _add_criteria_has_temporary_address(self) -> None: """ Filters subjects based on whether they have a temporary address on record. """ + try: - answer = YesNoType.by_description_case_insensitive(self.criteria_value) + answer = TemporaryAddressType.from_description(self.criteria_value) + + NEVER_HAS_HAD_TEMP_ADDRESS = (" LEFT OUTER JOIN sd_address_t adds " + " ON adds.contact_id = c.contact_id " + " AND adds.address_type = 13043 " + " AND adds.effective_from IS NOT NULL ") + + HAS_HAD_TEMP_ADDRESS = (" INNER JOIN sd_address_t adds " + " ON adds.contact_id = c.contact_id " + " AND adds.address_type = 13043 " + " AND adds.effective_from IS NOT NULL ") + + if answer == TemporaryAddressType.YES: + self.sql_from.append(HAS_HAD_TEMP_ADDRESS) + + elif answer == TemporaryAddressType.NO: + self.sql_from.append(NEVER_HAS_HAD_TEMP_ADDRESS) + self.sql_where.append(" AND adds.address_id IS NULL ") + + elif answer == TemporaryAddressType.EXPIRED: + self.sql_from.append(HAS_HAD_TEMP_ADDRESS) + self.sql_where.append(" AND adds.effective_to < TRUNC(SYSDATE) ") + + elif answer == TemporaryAddressType.CURRENT: + self.sql_from.append(HAS_HAD_TEMP_ADDRESS) + self.sql_where.append(" AND TRUNC(SYSDATE) BETWEEN adds.effective_from AND adds.effective_to ") + + elif answer == TemporaryAddressType.FUTURE: + self.sql_from.append(HAS_HAD_TEMP_ADDRESS) + self.sql_where.append(" AND adds.effective_from > TRUNC(SYSDATE) ") - if answer == YesNoType.YES: - self.sql_from.append( - " INNER JOIN sd_address_t adds ON adds.contact_id = c.contact_id " - " AND adds.ADDRESS_TYPE = 13043 " - " AND adds.EFFECTIVE_FROM IS NOT NULL " - ) - elif answer == YesNoType.NO: - self.sql_from.append( - " LEFT JOIN sd_address_t adds ON adds.contact_id = c.contact_id " - ) - self.sql_where.append( - " AND NOT EXISTS (" - " SELECT 1 " - " FROM sd_address_t x " - " WHERE x.contact_id = c.contact_id " - " AND x.address_type = 13043" - " AND x.effective_from is not null) " - ) else: raise ValueError() except Exception: @@ -3187,6 +3202,23 @@ def _add_criteria_bowel_scope_due_date_reason(self): except Exception: raise SelectionBuilderException(self.criteria_key_name, self.criteria_value) + def _add_criteria_subject_is_due_for_invite(self) -> None: + try: + answer = YesNoType.by_description_case_insensitive(self.criteria_value) + + if answer == YesNoType.YES: + self.sql_from.append(" INNER JOIN next_invitation_subject nis ") + self.sql_from.append(" ON nis.subject_id = ss.screening_subject_id ") + + if answer == YesNoType.NO: + self.sql_from.append(" LEFT OUTER JOIN next_invitation_subject nis ") + self.sql_from.append(" ON nis.subject_id = ss.screening_subject_id ") + + self.sql_where.append(" AND nis.subject_id IS NULL ") + + except Exception: + raise SelectionBuilderException(self.criteria_key_name, self.criteria_value) + def _add_criteria_manual_cease_requested(self) -> None: """ Adds criteria for filtering subjects based on the manual cease requested status. diff --git a/utils/subject_notes.py b/utils/subject_notes.py index fc7b1db8..9ad9a01a 100644 --- a/utils/subject_notes.py +++ b/utils/subject_notes.py @@ -182,3 +182,4 @@ def verify_note_removal_and_obsolete_transition( found ), f"❌ Removed note NOT found in obsolete list. Title: '{removed_title}', Note: '{removed_note}'" logging.info("✅ Removed note confirmed in obsolete notes.") + diff --git a/utils/table_util.py b/utils/table_util.py index cc09634e..9b49cd20 100644 --- a/utils/table_util.py +++ b/utils/table_util.py @@ -286,25 +286,28 @@ def get_cell_value(self, column_name: str, row_index: int) -> str: ) def assert_surname_in_table(self, surname_pattern: str) -> None: + assert self.assert_value_in_table(surname_pattern, 3), f"No surname matching '{surname_pattern}' found in table." + + def assert_value_in_table(self, value_pattern: str, col_id: int) -> bool: """ - Asserts that a surname matching the given pattern exists in the table. + Asserts that a value matching the given pattern exists in the table. Args: - surname_pattern (str): The surname or pattern to search for (supports '*' as a wildcard at the end). + value_pattern (str): The value or pattern to search for (supports '*' as a wildcard at the end). """ - # Locate all surname cells (adjust selector as needed) - surname_criteria = self.page.locator( - "//table//tr[position()>1]/td[3]" + # Locate all cells for given column + value_criteria = self.page.locator( + f"//table//tr[position()>1]/td[{col_id}]" ) # Use the correct column index - if surname_pattern.endswith("*"): - prefix = surname_pattern[:-1] + if value_pattern.endswith("*"): + prefix = value_pattern[:-1] found = any( cell.inner_text().startswith(prefix) - for cell in surname_criteria.element_handles() + for cell in value_criteria.element_handles() ) else: found = any( - surname_pattern == cell.inner_text() - for cell in surname_criteria.element_handles() + value_pattern == cell.inner_text() + for cell in value_criteria.element_handles() ) assert found, f"No surname matching '{surname_pattern}' found in table."