pxil-selenium-automation/tests/regression/exchange/idas/matching/bid_transfer_actions.py

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

from tests.regression.ui_helpers import (
    open_bootstrap_nav_submenu,
    select_select2_value_by_container_id,
)


BID_TRANSFER_URL_FRAGMENT = "../MCO/IDASBidTransfer.aspx"
SUPPORTED_MARKETS = ("GDAM", "DAM", "HPDAM")


def open_idas_bid_transfer_page(driver):

    open_bootstrap_nav_submenu(
        driver,
        "Trade Processing",
        "IDAS",
        BID_TRANSFER_URL_FRAGMENT,
    )
    WebDriverWait(driver, 20).until(
        EC.visibility_of_element_located((By.ID, "s2id_ddlproduct"))
    )


def select_bid_transfer_product(driver, product_code):

    select_select2_value_by_container_id(
        driver,
        "s2id_ddlproduct",
        product_code,
    )
    WebDriverWait(driver, 20).until(
        lambda current_driver: product_code.lower()
        in current_driver.find_element(By.ID, "s2id_ddlproduct").text.lower()
    )


def transfer_market_orders(driver, market, timeout_seconds=30):

    market = _validate_market(market)
    wait = WebDriverWait(driver, timeout_seconds)
    row = _market_row(driver, market)
    old_request_id = row.find_elements(By.TAG_NAME, "td")[1].text.strip()
    old_status = _status_element(driver, market).text.strip()

    transfer_button = wait.until(EC.element_to_be_clickable((By.ID, market)))
    driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", transfer_button)
    transfer_button.click()

    wait.until(
        lambda current_driver: _transfer_response_received(
            current_driver,
            market,
            old_request_id,
            old_status,
        )
    )

    error_message = _visible_message(driver, "divError")
    if error_message:
        raise AssertionError(f"{market} bid transfer failed: {error_message}")

    success_message = _visible_message(driver, "successMsg")
    status = _read_status(driver, market)
    if not success_message:
        raise AssertionError(
            f"{market} bid transfer did not show a success message. Current status: {status['text']}"
        )
    _assert_not_failed(market, "transfer", status)
    return {"message": success_message, **status}


def check_market_transfer_status(driver, market, timeout_seconds=30):

    market = _validate_market(market)
    check_button = WebDriverWait(driver, 20).until(
        EC.element_to_be_clickable(
            (
                By.XPATH,
                f"//button[contains(@onclick,\"GetMCOstatus('{market}')\")]",
            )
        )
    )
    driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", check_button)
    check_button.click()

    modal = WebDriverWait(driver, timeout_seconds).until(_visible_status_modal)

    error_message = _visible_message(driver, "divError")
    if error_message:
        raise AssertionError(f"{market} status check failed: {error_message}")

    modal_text = modal.text.strip()
    status_match = re.search(
        r"(?:^|\n)\s*Status\s*:\s*([^\r\n]+)",
        modal_text,
        flags=re.IGNORECASE,
    )
    if not status_match:
        raise AssertionError(
            f"Could not read {market} MCO status from the Send Order Details popup."
        )

    status_text = status_match.group(1).strip()
    return {
        "text": status_text,
        "normalized": _normalize_popup_status(status_text),
        "modal": modal,
    }


def close_market_transfer_status(driver, status_result):

    modal = status_result["modal"]
    close_button = modal.find_element(
        By.XPATH,
        ".//button[normalize-space()='Close' or @data-dismiss='modal']",
    )
    close_button.click()
    WebDriverWait(driver, 20).until(EC.invisibility_of_element(modal))


def _validate_market(market):

    normalized_market = market.strip().upper()
    if normalized_market not in SUPPORTED_MARKETS:
        raise ValueError(
            f"Unsupported IDAS bid-transfer market '{market}'. Use one of {SUPPORTED_MARKETS}."
        )
    return normalized_market


def _market_row(driver, market):

    return driver.find_element(
        By.XPATH,
        f"//div[@id='SendOrderId']//tr[td[1][normalize-space()='{market}']]",
    )


def _status_element(driver, market):

    return driver.find_element(By.ID, f"{market.lower()}sat")


def _transfer_response_received(driver, market, old_request_id, old_status):

    if _visible_message(driver, "divError"):
        return True
    row = _market_row(driver, market)
    request_id = row.find_elements(By.TAG_NAME, "td")[1].text.strip()
    status = _status_element(driver, market).text.strip()
    return bool(request_id and request_id != old_request_id) or status != old_status


def _visible_message(driver, element_id):

    elements = driver.find_elements(By.ID, element_id)
    if not elements or not elements[0].is_displayed():
        return ""
    return elements[0].text.strip()


def _visible_status_modal(driver):

    modals = driver.find_elements(
        By.XPATH,
        "//div[contains(@class,'modal')][.//*[contains(normalize-space(),'Send Order Details')]]",
    )
    for modal in modals:
        if modal.is_displayed():
            return modal
    return False


def _normalize_popup_status(status):

    return re.sub(r"[\s-]+", "_", status.strip()).upper()


def _read_status(driver, market):

    element = _status_element(driver, market)
    background_color = element.value_of_css_property("background-color")
    return {
        "text": element.text.strip(),
        "background_color": background_color,
        "is_green": _is_green(background_color),
    }


def _is_green(css_color):

    channels = [int(value) for value in re.findall(r"\d+", css_color)[:3]]
    return len(channels) == 3 and channels[1] > channels[0] and channels[1] > channels[2]


def _assert_not_failed(market, action, status):

    if "fail" in status["text"].lower():
        raise AssertionError(
            f"{market} {action} returned '{status['text']}'. Matching has been stopped."
        )