pxil-selenium-automation/tests/regression/exchange/idas/bidding/upload_actions.py

from pathlib import Path
import time

from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait


def upload_bid_file(driver, upload_path):

    file_path = Path(upload_path)
    if not file_path.exists():
        raise FileNotFoundError(f"Upload file not found: {file_path}")

    wait = WebDriverWait(driver, 30)
    existing_inputs = _file_inputs(driver)

    file_input = _preferred_file_input(existing_inputs)
    if file_input is None:
        upload_button = wait.until(
            EC.element_to_be_clickable(
                (
                    By.XPATH,
                    "//*[self::button or self::a or self::input]"
                    "[normalize-space()='Upload' or contains(normalize-space(),'Upload') or @value='Upload']"
                )
            )
        )

        try:
            upload_button.click()
        except Exception:
            driver.execute_script("arguments[0].click();", upload_button)

        file_input = wait.until(
            lambda _driver: _find_new_or_existing_file_input(_driver, existing_inputs)
        )

    if file_input is None:
        raise RuntimeError(
            "No HTML file input was found for the Upload action. "
            "If this page only exposes the native Windows file picker, we will need a Windows GUI automation fallback."
        )

    driver.execute_script("arguments[0].style.display='block';", file_input)
    file_input.send_keys(str(file_path.resolve()))

    wait.until(
        lambda _driver: _upload_acknowledged(_driver, file_path.name)
    )


def submit_uploaded_bid(driver, post_upload_delay_seconds=0, submit_to_fetch_delay_seconds=0):

    wait = WebDriverWait(driver, 30)
    previous_status = _current_status_text(driver)

    if post_upload_delay_seconds > 0:
        time.sleep(post_upload_delay_seconds)

    submit_button = wait.until(
        EC.element_to_be_clickable((By.ID, "btnSubmitOrder"))
    )
    driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", submit_button)
    try:
        submit_button.click()
    except Exception:
        driver.execute_script("arguments[0].click();", submit_button)

    if submit_to_fetch_delay_seconds > 0:
        time.sleep(submit_to_fetch_delay_seconds)

    fetch_button = wait.until(
        EC.element_to_be_clickable((By.ID, "btnFetch"))
    )
    driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", fetch_button)
    try:
        fetch_button.click()
    except Exception:
        driver.execute_script("arguments[0].click();", fetch_button)

    status_text = wait.until(
        lambda _driver: _refreshed_status_text(_driver, previous_status) or False
    )

    lowered_status = status_text.lower()
    if "reject" in lowered_status:
        raise AssertionError(f"Uploaded order was rejected. Status: {status_text}")

    if "submit" not in lowered_status and "open" not in lowered_status and "accept" not in lowered_status:
        raise AssertionError(f"Unexpected uploaded order status after fetch: {status_text}")

    return status_text


def _find_new_or_existing_file_input(driver, existing_inputs):

    current_inputs = _file_inputs(driver)

    for current in current_inputs:
        if current not in existing_inputs:
            return current

    preferred = _preferred_file_input(current_inputs)
    if preferred is not None:
        return preferred

    for current in current_inputs:
        if current.is_displayed():
            return current

    if current_inputs:
        return current_inputs[0]

    return False


def _file_inputs(driver):

    return driver.find_elements(By.XPATH, "//input[@type='file' and not(@webkitdirectory)]")


def _preferred_file_input(inputs):

    if not inputs:
        return None

    for current in inputs:
        element_id = (current.get_attribute("id") or "").lower()
        element_name = (current.get_attribute("name") or "").lower()
        outer_html = (current.get_attribute("outerHTML") or "").lower()

        if any(token in text for token in ("upload", "file") for text in (element_id, element_name, outer_html)):
            return current

    for current in inputs:
        if current.is_displayed():
            return current

    return inputs[0]


def _upload_acknowledged(driver, file_name):

    page_text = driver.find_element(By.TAG_NAME, "body").text
    if file_name in page_text:
        return True

    if "Upload Logs" in page_text or "View Upload Logs" in page_text:
        return True

    return False


def _current_status_text(driver):

    by_id = driver.find_elements(By.ID, "lblStatus")
    for candidate in by_id:
        text = candidate.text.strip()
        if text:
            return text

    candidates = driver.find_elements(
        By.XPATH,
        "//*[contains(normalize-space(),'Status :') or contains(normalize-space(),'Status:')]"
    )
    for candidate in candidates:
        text = candidate.text.strip()
        if text:
            return text
    return ""


def _refreshed_status_text(driver, previous_status):

    current_status = _current_status_text(driver).strip()
    if not current_status:
        return ""

    normalized_current = " ".join(current_status.split()).lower()
    normalized_previous = " ".join((previous_status or "").split()).lower()

    terminal_tokens = ("reject", "submitted", "open", "accept")
    if any(token in normalized_current for token in terminal_tokens):
        if normalized_current != normalized_previous:
            return current_status

        if "order id" in normalized_current:
            return current_status

    return ""