346 lines
11 KiB
Python
346 lines
11 KiB
Python
import time
|
|
from pathlib import Path
|
|
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.webdriver.support.select import Select
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
|
|
from tests.regression.ui_helpers import open_bootstrap_nav_submenu
|
|
|
|
|
|
MATCHING_URL_FRAGMENT = "../Application/IDASInitiateMatching.aspx"
|
|
GENERATE_FLOW_REQUEST_BUTTON_ID = "btnGenerateProvisionalFlowRequest"
|
|
REVIEW_RESULT_BUTTON_ID = "btnReviewResult"
|
|
REVIEW_RESULT_MODAL_ID = "divReviewResult"
|
|
SUCCESS_CONTAINER_ID = "divSuccess"
|
|
SUCCESS_MESSAGE_ID = "successMsg"
|
|
|
|
|
|
def open_idas_matching_page(driver):
|
|
|
|
open_bootstrap_nav_submenu(
|
|
driver,
|
|
"Trade Processing",
|
|
"IDAS",
|
|
MATCHING_URL_FRAGMENT
|
|
)
|
|
|
|
wait = WebDriverWait(driver, 20)
|
|
wait.until(EC.visibility_of_element_located((By.ID, "ddlproduct")))
|
|
|
|
|
|
def configure_matching_request(driver, product_code, matching_type, source_type=None):
|
|
|
|
wait = WebDriverWait(driver, 20)
|
|
|
|
if source_type:
|
|
source_type_select = wait.until(
|
|
EC.visibility_of_element_located(
|
|
(
|
|
By.XPATH,
|
|
"//label[normalize-space()='SourceType']/following::select[1]"
|
|
)
|
|
)
|
|
)
|
|
Select(source_type_select).select_by_visible_text(source_type)
|
|
|
|
product_select = wait.until(
|
|
EC.visibility_of_element_located((By.ID, "ddlproduct"))
|
|
)
|
|
_select_product_code(product_select, product_code)
|
|
|
|
matching_type_select = wait.until(
|
|
EC.visibility_of_element_located((By.ID, "ddlMatchingTypeIM"))
|
|
)
|
|
Select(matching_type_select).select_by_visible_text(matching_type)
|
|
|
|
|
|
def _select_product_code(select_element, product_code):
|
|
|
|
normalized_code = product_code.strip().lower()
|
|
product_select = Select(select_element)
|
|
for option in product_select.options:
|
|
option_text = option.text.strip()
|
|
option_code = option_text.split("(", 1)[0].strip().lower()
|
|
if option_text.lower() == normalized_code or option_code == normalized_code:
|
|
product_select.select_by_visible_text(option_text)
|
|
return
|
|
|
|
available_options = [option.text.strip() for option in product_select.options]
|
|
raise ValueError(
|
|
f"Product code '{product_code}' is unavailable on the matching page. "
|
|
f"Available options: {available_options}"
|
|
)
|
|
|
|
|
|
def start_provisional_match(driver):
|
|
|
|
wait = WebDriverWait(driver, 20)
|
|
provisional_button = wait.until(
|
|
EC.element_to_be_clickable((By.ID, "btnProvisionalMatch"))
|
|
)
|
|
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", provisional_button)
|
|
try:
|
|
provisional_button.click()
|
|
except Exception:
|
|
driver.execute_script("arguments[0].click();", provisional_button)
|
|
|
|
confirm_provisional_match(driver)
|
|
|
|
|
|
def confirm_provisional_match(driver):
|
|
|
|
wait = WebDriverWait(driver, 20)
|
|
modal = wait.until(
|
|
EC.visibility_of_element_located((By.ID, "matchingConfirmation"))
|
|
)
|
|
|
|
ok_button = wait.until(
|
|
EC.element_to_be_clickable((By.ID, "btnProvisional"))
|
|
)
|
|
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", ok_button)
|
|
try:
|
|
ok_button.click()
|
|
except Exception:
|
|
driver.execute_script("arguments[0].click();", ok_button)
|
|
|
|
wait.until(EC.invisibility_of_element(modal))
|
|
time.sleep(2)
|
|
|
|
|
|
def poll_provisional_match_status(driver, interval_seconds=2, timeout_seconds=120):
|
|
|
|
wait = WebDriverWait(driver, 20)
|
|
end_time = time.monotonic() + timeout_seconds
|
|
last_message = ""
|
|
|
|
while time.monotonic() < end_time:
|
|
check_status_button = wait.until(
|
|
EC.element_to_be_clickable((By.ID, "lnkCheckStatusProvisional"))
|
|
)
|
|
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", check_status_button)
|
|
try:
|
|
check_status_button.click()
|
|
except Exception:
|
|
driver.execute_script("arguments[0].click();", check_status_button)
|
|
|
|
message, state = _read_matching_status(driver)
|
|
if message:
|
|
last_message = message
|
|
|
|
if state == "success":
|
|
return message
|
|
|
|
if state == "error":
|
|
raise AssertionError(f"Matching failed: {message}")
|
|
|
|
time.sleep(interval_seconds)
|
|
|
|
raise AssertionError(
|
|
"Matching status did not reach success within the timeout. "
|
|
f"Last visible message: {last_message or 'No status message found'}"
|
|
)
|
|
|
|
|
|
def prepare_matching_download_folder(test_context, folder_name="matching_files"):
|
|
|
|
download_folder = Path(test_context["folder"]) / folder_name
|
|
download_folder.mkdir(parents=True, exist_ok=True)
|
|
|
|
extra_folders = test_context.setdefault("extra_folders", [])
|
|
if not any(item.get("path") == str(download_folder) for item in extra_folders):
|
|
extra_folders.append(
|
|
{
|
|
"name": "Matching Files Folder",
|
|
"path": str(download_folder),
|
|
}
|
|
)
|
|
|
|
return download_folder
|
|
|
|
|
|
def configure_browser_downloads(driver, download_folder):
|
|
|
|
driver.execute_cdp_cmd(
|
|
"Page.setDownloadBehavior",
|
|
{
|
|
"behavior": "allow",
|
|
"downloadPath": str(download_folder),
|
|
}
|
|
)
|
|
|
|
|
|
def click_generate_flow_request(driver):
|
|
|
|
wait = WebDriverWait(driver, 20)
|
|
|
|
generate_button = wait.until(
|
|
EC.element_to_be_clickable((By.ID, GENERATE_FLOW_REQUEST_BUTTON_ID))
|
|
)
|
|
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", generate_button)
|
|
try:
|
|
generate_button.click()
|
|
except Exception:
|
|
driver.execute_script("arguments[0].click();", generate_button)
|
|
|
|
|
|
def wait_for_flow_request_success(driver, timeout_seconds=30):
|
|
|
|
wait = WebDriverWait(driver, timeout_seconds)
|
|
success_container = wait.until(
|
|
EC.visibility_of_element_located((By.ID, SUCCESS_CONTAINER_ID))
|
|
)
|
|
|
|
try:
|
|
message = success_container.find_element(By.ID, SUCCESS_MESSAGE_ID).text.strip()
|
|
except Exception:
|
|
message = success_container.text.strip()
|
|
|
|
if not message:
|
|
raise AssertionError("Flow request success banner appeared without a message.")
|
|
|
|
return message
|
|
|
|
|
|
def export_review_result_for_source(driver, source_type, download_folder, timeout_seconds=60):
|
|
|
|
existing_files = {path.name for path in Path(download_folder).iterdir()}
|
|
wait = WebDriverWait(driver, 20)
|
|
|
|
review_result_button = wait.until(
|
|
EC.element_to_be_clickable((By.ID, REVIEW_RESULT_BUTTON_ID))
|
|
)
|
|
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", review_result_button)
|
|
try:
|
|
review_result_button.click()
|
|
except Exception:
|
|
driver.execute_script("arguments[0].click();", review_result_button)
|
|
|
|
modal = wait.until(
|
|
EC.visibility_of_element_located((By.ID, REVIEW_RESULT_MODAL_ID))
|
|
)
|
|
|
|
export_button = wait.until(
|
|
EC.element_to_be_clickable(
|
|
(
|
|
By.CSS_SELECTOR,
|
|
f"#{REVIEW_RESULT_MODAL_ID} .pExcel.pButton"
|
|
)
|
|
)
|
|
)
|
|
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", export_button)
|
|
try:
|
|
export_button.click()
|
|
except Exception:
|
|
driver.execute_script("arguments[0].click();", export_button)
|
|
|
|
downloaded_file = wait_for_new_download(
|
|
download_folder,
|
|
existing_files=existing_files,
|
|
timeout_seconds=timeout_seconds
|
|
)
|
|
renamed_file = rename_download_with_source_type(downloaded_file, source_type)
|
|
|
|
close_button = wait.until(
|
|
EC.element_to_be_clickable(
|
|
(
|
|
By.CSS_SELECTOR,
|
|
f"#{REVIEW_RESULT_MODAL_ID} button.close"
|
|
)
|
|
)
|
|
)
|
|
try:
|
|
close_button.click()
|
|
except Exception:
|
|
driver.execute_script("arguments[0].click();", close_button)
|
|
|
|
wait.until(EC.invisibility_of_element(modal))
|
|
return renamed_file
|
|
|
|
|
|
def wait_for_new_download(download_folder, existing_files=None, timeout_seconds=60):
|
|
|
|
download_path = Path(download_folder)
|
|
existing_files = existing_files or set()
|
|
end_time = time.monotonic() + timeout_seconds
|
|
|
|
while time.monotonic() < end_time:
|
|
current_files = [path for path in download_path.iterdir() if path.is_file()]
|
|
new_files = [path for path in current_files if path.name not in existing_files]
|
|
partial_downloads = [
|
|
path for path in current_files
|
|
if path.suffix.lower() in {".crdownload", ".tmp", ".part"}
|
|
]
|
|
|
|
completed_files = [
|
|
path for path in new_files
|
|
if path.suffix.lower() not in {".crdownload", ".tmp", ".part"}
|
|
]
|
|
if completed_files and not partial_downloads:
|
|
completed_files.sort(key=lambda item: item.stat().st_mtime, reverse=True)
|
|
return str(completed_files[0])
|
|
|
|
time.sleep(1)
|
|
|
|
visible_names = sorted(path.name for path in download_path.iterdir() if path.is_file())
|
|
raise AssertionError(
|
|
"No completed matching download was detected within the timeout. "
|
|
f"Folder: {download_folder}. Current files: {visible_names}"
|
|
)
|
|
|
|
|
|
def rename_download_with_source_type(downloaded_file, source_type):
|
|
|
|
file_path = Path(downloaded_file)
|
|
normalized_source_type = source_type.strip().lower().replace(" ", "_")
|
|
target_path = file_path.with_name(f"{normalized_source_type}_{file_path.name}")
|
|
|
|
if target_path.exists():
|
|
stem = file_path.stem
|
|
suffix = file_path.suffix
|
|
counter = 2
|
|
while True:
|
|
candidate = file_path.with_name(
|
|
f"{normalized_source_type}_{stem}_{counter}{suffix}"
|
|
)
|
|
if not candidate.exists():
|
|
target_path = candidate
|
|
break
|
|
counter += 1
|
|
|
|
file_path.rename(target_path)
|
|
return str(target_path)
|
|
|
|
|
|
def _read_matching_status(driver):
|
|
|
|
success_divs = driver.find_elements(By.ID, "divSuccess")
|
|
for current in success_divs:
|
|
if current.is_displayed():
|
|
text = current.text.strip()
|
|
if text:
|
|
return text, "success"
|
|
|
|
error_divs = driver.find_elements(By.ID, "divError")
|
|
for current in error_divs:
|
|
if current.is_displayed():
|
|
text = current.text.strip()
|
|
if text:
|
|
return text, "error"
|
|
|
|
warning_divs = driver.find_elements(By.ID, "divMatchingStatus")
|
|
for current in warning_divs:
|
|
if current.is_displayed():
|
|
text = current.text.strip()
|
|
if text:
|
|
return text, "in_progress"
|
|
|
|
warning_messages = driver.find_elements(By.ID, "warningMsg")
|
|
for current in warning_messages:
|
|
text = current.text.strip()
|
|
if text:
|
|
return text, "in_progress"
|
|
|
|
return "", "unknown"
|