Skip to content

Commit eb3c515

Browse files
davethepunkyonedavethepunkyone
committed
Attempting push with new line ends
1 parent 542806a commit eb3c515

File tree

2 files changed

+179
-179
lines changed

2 files changed

+179
-179
lines changed

‎tests/smokescreen/test_compartment_3.py‎

Lines changed: 114 additions & 114 deletions
Original file line numberDiff line numberDiff line change
@@ -1,114 +1,114 @@
1-
import logging
2-
import pytest
3-
from playwright.sync_api import Page
4-
from pages.logout.log_out_page import Logout
5-
from utils.batch_processing import batch_processing
6-
from utils.fit_kit_logged import process_kit_data
7-
from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
8-
from utils.oracle.oracle_specific_functions import (
9-
update_kit_service_management_entity,
10-
execute_stored_procedures,
11-
)
12-
from utils.user_tools import UserTools
13-
from jproperties import Properties
14-
from sys import platform
15-
16-
17-
@pytest.fixture
18-
def smokescreen_properties() -> dict:
19-
"""
20-
Reads the 'bcss_smokescreen_tests.properties' file and populates a 'Properties' object.
21-
Returns a dictionary of properties for use in tests.
22-
23-
Returns:
24-
dict: A dictionary containing the values loaded from the 'bcss_smokescreen_tests.properties' file.
25-
"""
26-
configs = Properties()
27-
if platform == "win32": # File path from content root is required on Windows OS
28-
with open(
29-
"tests/smokescreen/bcss_smokescreen_tests.properties", "rb"
30-
) as read_prop:
31-
configs.load(read_prop)
32-
elif platform == "darwin": # Only the filename is required on macOS
33-
with open("bcss_smokescreen_tests.properties", "rb") as read_prop:
34-
configs.load(read_prop)
35-
return configs.properties
36-
37-
38-
@pytest.mark.smokescreen
39-
@pytest.mark.compartment3
40-
def test_compartment_3(page: Page, smokescreen_properties: dict) -> None:
41-
"""
42-
This is the main compartment 3 method
43-
First it finds any relevant test data from the DB and stores it in a pandas dataframe
44-
Then it separates it out into normal and abnormal results
45-
Once that is done it updates a table on the DB and runs two stored procedures so that these subjects will not have normal/abnormal results on BCSS
46-
It then checks that the status of the subject is as expected using the subject screening summary page
47-
Then it process two batches performing checks on the subjects to ensure they always have the correct event status
48-
"""
49-
UserTools.user_login(page, "Hub Manager State Registered")
50-
51-
# Find data , separate it into normal and abnormal, Add results to the test records in the KIT_QUEUE table (i.e. mimic receiving results from the middleware)
52-
# and get device IDs and their flags
53-
device_ids = process_kit_data(smokescreen_properties)
54-
# Retrieve NHS numbers for each device_id and determine normal/abnormal status
55-
nhs_numbers = []
56-
normal_flags = []
57-
58-
for device_id, is_normal in device_ids:
59-
nhs_number = update_kit_service_management_entity(
60-
device_id, is_normal, smokescreen_properties
61-
)
62-
nhs_numbers.append(nhs_number)
63-
normal_flags.append(
64-
is_normal
65-
) # Store the flag (True for normal, False for abnormal)
66-
67-
# Run two stored procedures to process any kit queue records at status BCSS_READY
68-
try:
69-
execute_stored_procedures()
70-
logging.info("Stored procedures executed successfully.")
71-
except Exception as e:
72-
logging.error(f"Error executing stored procedures: {str(e)}")
73-
raise
74-
75-
# Check the results of the processed FIT kits have correctly updated the status of the associated subjects
76-
# Verify subject event status based on normal or abnormal classification
77-
for nhs_number, is_normal in zip(nhs_numbers, normal_flags):
78-
expected_status = (
79-
"S2 - Normal" if is_normal else "A8 - Abnormal"
80-
) # S2 for normal, A8 for abnormal
81-
logging.info(
82-
f"Verifying NHS number: {nhs_number} with expected status: {expected_status}"
83-
)
84-
85-
try:
86-
verify_subject_event_status_by_nhs_no(page, nhs_number, expected_status)
87-
logging.info(
88-
f"Successfully verified NHS number {nhs_number} with status {expected_status}"
89-
)
90-
except Exception as e:
91-
logging.error(
92-
f"Verification failed for NHS number {nhs_number} with status {expected_status}: {str(e)}"
93-
)
94-
raise
95-
96-
# Process S2 batch
97-
batch_processing(
98-
page,
99-
"S2",
100-
"Subject Result (Normal)",
101-
"S158 - Subject Discharge Sent (Normal)",
102-
True,
103-
)
104-
105-
# Process S158 batch
106-
batch_processing(
107-
page,
108-
"S158",
109-
"GP Result (Normal)",
110-
"S159 - GP Discharge Sent (Normal)",
111-
)
112-
113-
# Log out
114-
Logout(page).log_out()
1+
import logging
2+
import pytest
3+
from playwright.sync_api import Page
4+
from pages.logout.log_out_page import Logout
5+
from utils.batch_processing import batch_processing
6+
from utils.fit_kit_logged import process_kit_data
7+
from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
8+
from utils.oracle.oracle_specific_functions import (
9+
update_kit_service_management_entity,
10+
execute_stored_procedures,
11+
)
12+
from utils.user_tools import UserTools
13+
from jproperties import Properties
14+
from sys import platform
15+
16+
17+
@pytest.fixture
18+
def smokescreen_properties() -> dict:
19+
"""
20+
Reads the 'bcss_smokescreen_tests.properties' file and populates a 'Properties' object.
21+
Returns a dictionary of properties for use in tests.
22+
23+
Returns:
24+
dict: A dictionary containing the values loaded from the 'bcss_smokescreen_tests.properties' file.
25+
"""
26+
configs = Properties()
27+
if platform == "win32": # File path from content root is required on Windows OS
28+
with open(
29+
"tests/smokescreen/bcss_smokescreen_tests.properties", "rb"
30+
) as read_prop:
31+
configs.load(read_prop)
32+
elif platform == "darwin": # Only the filename is required on macOS
33+
with open("bcss_smokescreen_tests.properties", "rb") as read_prop:
34+
configs.load(read_prop)
35+
return configs.properties
36+
37+
38+
@pytest.mark.smokescreen
39+
@pytest.mark.compartment3
40+
def test_compartment_3(page: Page, smokescreen_properties: dict) -> None:
41+
"""
42+
This is the main compartment 3 method
43+
First it finds any relevant test data from the DB and stores it in a pandas dataframe
44+
Then it separates it out into normal and abnormal results
45+
Once that is done it updates a table on the DB and runs two stored procedures so that these subjects will not have normal/abnormal results on BCSS
46+
It then checks that the status of the subject is as expected using the subject screening summary page
47+
Then it process two batches performing checks on the subjects to ensure they always have the correct event status
48+
"""
49+
UserTools.user_login(page, "Hub Manager State Registered")
50+
51+
# Find data , separate it into normal and abnormal, Add results to the test records in the KIT_QUEUE table (i.e. mimic receiving results from the middleware)
52+
# and get device IDs and their flags
53+
device_ids = process_kit_data(smokescreen_properties)
54+
# Retrieve NHS numbers for each device_id and determine normal/abnormal status
55+
nhs_numbers = []
56+
normal_flags = []
57+
58+
for device_id, is_normal in device_ids:
59+
nhs_number = update_kit_service_management_entity(
60+
device_id, is_normal, smokescreen_properties
61+
)
62+
nhs_numbers.append(nhs_number)
63+
normal_flags.append(
64+
is_normal
65+
) # Store the flag (True for normal, False for abnormal)
66+
67+
# Run two stored procedures to process any kit queue records at status BCSS_READY
68+
try:
69+
execute_stored_procedures()
70+
logging.info("Stored procedures executed successfully.")
71+
except Exception as e:
72+
logging.error(f"Error executing stored procedures: {str(e)}")
73+
raise
74+
75+
# Check the results of the processed FIT kits have correctly updated the status of the associated subjects
76+
# Verify subject event status based on normal or abnormal classification
77+
for nhs_number, is_normal in zip(nhs_numbers, normal_flags):
78+
expected_status = (
79+
"S2 - Normal" if is_normal else "A8 - Abnormal"
80+
) # S2 for normal, A8 for abnormal
81+
logging.info(
82+
f"Verifying NHS number: {nhs_number} with expected status: {expected_status}"
83+
)
84+
85+
try:
86+
verify_subject_event_status_by_nhs_no(page, nhs_number, expected_status)
87+
logging.info(
88+
f"Successfully verified NHS number {nhs_number} with status {expected_status}"
89+
)
90+
except Exception as e:
91+
logging.error(
92+
f"Verification failed for NHS number {nhs_number} with status {expected_status}: {str(e)}"
93+
)
94+
raise
95+
96+
# Process S2 batch
97+
batch_processing(
98+
page,
99+
"S2",
100+
"Subject Result (Normal)",
101+
"S158 - Subject Discharge Sent (Normal)",
102+
True,
103+
)
104+
105+
# Process S158 batch
106+
batch_processing(
107+
page,
108+
"S158",
109+
"GP Result (Normal)",
110+
"S159 - GP Discharge Sent (Normal)",
111+
)
112+
113+
# Log out
114+
Logout(page).log_out()

‎utils/fit_kit_logged.py‎

Lines changed: 65 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -1,65 +1,65 @@
1-
from oracle.oracle import OracleDB
2-
from oracle.oracle_specific_functions import get_kit_id_logged_from_db
3-
import pandas as pd
4-
import logging
5-
import pytest
6-
7-
8-
def process_kit_data(smokescreen_properties) -> list:
9-
"""
10-
This method retrieved the test data needed for compartment 3 and then splits it into two data frames:
11-
- 1 normal
12-
- 1 abnormal
13-
Once the dataframe is split in two it then creates two lists, one for normal and one for abnormal
14-
Each list will either have true or false appended depending on if it is normal or abnormal
15-
"""
16-
# Get test data for compartment 3
17-
kit_id_df = get_kit_id_logged_from_db()
18-
19-
# Split dataframe into two different dataframes, normal and abnormal
20-
normal_fit_kit_df, abnormal_fit_kit_df = split_fit_kits(
21-
kit_id_df, smokescreen_properties
22-
)
23-
24-
# Prepare a list to store device IDs and their respective flags
25-
device_ids = []
26-
27-
# Process normal kits (only 1)
28-
if not normal_fit_kit_df.empty:
29-
device_id = normal_fit_kit_df["device_id"].iloc[0]
30-
logging.info(
31-
f"Processing normal kit with Device ID: {device_id}"
32-
) # Logging normal device_id
33-
device_ids.append((device_id, True)) # Add to the list with normal flag
34-
else:
35-
pytest.fail("No normal kits found for processing.")
36-
37-
# Process abnormal kits (multiple, loop through)
38-
if not abnormal_fit_kit_df.empty:
39-
for index, row in abnormal_fit_kit_df.iterrows():
40-
device_id = row["device_id"]
41-
logging.info(
42-
f"Processing abnormal kit with Device ID: {device_id}"
43-
) # Logging abnormal device_id
44-
device_ids.append((device_id, False)) # Add to the list with abnormal flag
45-
else:
46-
pytest.fail("No abnormal kits found for processing.")
47-
48-
return device_ids
49-
50-
51-
def split_fit_kits(kit_id_df, smokescreen_properties: dict) -> pd.DataFrame:
52-
"""
53-
This method splits the dataframe into two, 1 normal and 1 abnormal
54-
"""
55-
number_of_normal = int(smokescreen_properties["c3_eng_number_of_normal_fit_kits"])
56-
number_of_abnormal = int(
57-
smokescreen_properties["c3_eng_number_of_abnormal_fit_kits"]
58-
)
59-
60-
# Split dataframe into two dataframes
61-
normal_fit_kit_df = kit_id_df.iloc[:number_of_normal]
62-
abnormal_fit_kit_df = kit_id_df.iloc[
63-
number_of_normal : number_of_normal + number_of_abnormal
64-
]
65-
return normal_fit_kit_df, abnormal_fit_kit_df
1+
from oracle.oracle import OracleDB
2+
from oracle.oracle_specific_functions import get_kit_id_logged_from_db
3+
import pandas as pd
4+
import logging
5+
import pytest
6+
7+
8+
def process_kit_data(smokescreen_properties) -> list:
9+
"""
10+
This method retrieved the test data needed for compartment 3 and then splits it into two data frames:
11+
- 1 normal
12+
- 1 abnormal
13+
Once the dataframe is split in two it then creates two lists, one for normal and one for abnormal
14+
Each list will either have true or false appended depending on if it is normal or abnormal
15+
"""
16+
# Get test data for compartment 3
17+
kit_id_df = get_kit_id_logged_from_db()
18+
19+
# Split dataframe into two different dataframes, normal and abnormal
20+
normal_fit_kit_df, abnormal_fit_kit_df = split_fit_kits(
21+
kit_id_df, smokescreen_properties
22+
)
23+
24+
# Prepare a list to store device IDs and their respective flags
25+
device_ids = []
26+
27+
# Process normal kits (only 1)
28+
if not normal_fit_kit_df.empty:
29+
device_id = normal_fit_kit_df["device_id"].iloc[0]
30+
logging.info(
31+
f"Processing normal kit with Device ID: {device_id}"
32+
) # Logging normal device_id
33+
device_ids.append((device_id, True)) # Add to the list with normal flag
34+
else:
35+
pytest.fail("No normal kits found for processing.")
36+
37+
# Process abnormal kits (multiple, loop through)
38+
if not abnormal_fit_kit_df.empty:
39+
for index, row in abnormal_fit_kit_df.iterrows():
40+
device_id = row["device_id"]
41+
logging.info(
42+
f"Processing abnormal kit with Device ID: {device_id}"
43+
) # Logging abnormal device_id
44+
device_ids.append((device_id, False)) # Add to the list with abnormal flag
45+
else:
46+
pytest.fail("No abnormal kits found for processing.")
47+
48+
return device_ids
49+
50+
51+
def split_fit_kits(kit_id_df, smokescreen_properties: dict) -> pd.DataFrame:
52+
"""
53+
This method splits the dataframe into two, 1 normal and 1 abnormal
54+
"""
55+
number_of_normal = int(smokescreen_properties["c3_eng_number_of_normal_fit_kits"])
56+
number_of_abnormal = int(
57+
smokescreen_properties["c3_eng_number_of_abnormal_fit_kits"]
58+
)
59+
60+
# Split dataframe into two dataframes
61+
normal_fit_kit_df = kit_id_df.iloc[:number_of_normal]
62+
abnormal_fit_kit_df = kit_id_df.iloc[
63+
number_of_normal : number_of_normal + number_of_abnormal
64+
]
65+
return normal_fit_kit_df, abnormal_fit_kit_df

0 commit comments

Comments
 (0)