Initial commit - Selenium automation framework

This commit is contained in:
2026-09-08 09:24:02 +05:30
commit e84b7e9950
76 changed files with 7484 additions and 0 deletions
View File
+14
View File
@@ -0,0 +1,14 @@
from selenium import webdriver
def create_chrome_driver():
options = webdriver.ChromeOptions()
options.add_argument("--disable-background-timer-throttling")
options.add_argument("--disable-backgrounding-occluded-windows")
options.add_argument("--disable-renderer-backgrounding")
options.add_argument("--disable-features=CalculateNativeWinOcclusion")
options.add_argument("--disable-hang-monitor")
options.add_argument("--disable-popup-blocking")
return webdriver.Chrome(options=options)
+62
View File
@@ -0,0 +1,62 @@
import json
from pathlib import Path
CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"
def load_config():
with CONFIG_PATH.open("r", encoding="utf-8") as f:
return json.load(f)
def get_step_delay_seconds(default=5):
config = load_config()
return config.get("step_delay_seconds", default)
def get_login_captcha_wait_seconds(default=20):
config = load_config()
return config.get("login_captcha_wait_seconds", default)
def get_login_post_submit_wait_seconds(default=10):
config = load_config()
return config.get("login_post_submit_wait_seconds", default)
def is_setup_session_enabled(default=True):
config = load_config()
return config.get("setup", {}).get("run_session", default)
def is_setup_token_enabled(default=True):
config = load_config()
return config.get("setup", {}).get("run_token", default)
def is_combined_setup_enabled():
config = load_config()
setup = config.get("setup", {})
return setup.get("run_session", True) and setup.get("run_token", True)
def parse_product_codes(value):
if value is None:
return []
if isinstance(value, str):
return [code.strip() for code in value.split(",") if code.strip()]
if isinstance(value, list):
return [str(code).strip() for code in value if str(code).strip()]
return [str(value).strip()]
+242
View File
@@ -0,0 +1,242 @@
import pytest
import os
import json
from tests.browser import create_chrome_driver
def get_run_id(config):
run_id = getattr(config, "_pxil_run_id", None)
if run_id:
return run_id
base_folder = os.path.join("reports", "screenshots")
os.makedirs(base_folder, exist_ok=True)
run_prefix = _get_run_prefix(config)
existing = [
name for name in os.listdir(base_folder)
if name.startswith(f"{run_prefix}_")
]
if not existing:
next_number = 1
else:
numbers = [
int(name.split("_")[-1])
for name in existing
if name.split("_")[-1].isdigit()
]
next_number = max(numbers) + 1
run_id = f"{run_prefix}_{next_number:03d}"
config._pxil_run_id = run_id
return run_id
def pytest_configure(config):
run_id = get_run_id(config)
if config.pluginmanager.hasplugin("html"):
reports_dir = os.path.abspath(os.path.join("reports", "html"))
os.makedirs(reports_dir, exist_ok=True)
config.option.htmlpath = os.path.join(reports_dir, f"{run_id}.html")
config.option.self_contained_html = True
# -----------------------------
# WebDriver Fixture
# -----------------------------
@pytest.fixture
def driver():
driver = create_chrome_driver()
driver.maximize_window()
yield driver
driver.quit()
# -----------------------------
# Generate Incremental Test ID
# -----------------------------
@pytest.fixture
def test_context(request):
test_id = get_run_id(request.config)
test_name = request.node.name
folder = os.path.abspath(
os.path.join("reports", "screenshots", test_id, test_name)
)
log_folder = os.path.abspath(
os.path.join("reports", "logs", test_id, test_name)
)
os.makedirs(folder, exist_ok=True)
os.makedirs(log_folder, exist_ok=True)
print(f"\nTEST ID: {test_id}")
return {
"test_id": test_id,
"folder": folder,
"log_folder": log_folder,
"steps": [],
"extra_folders": [],
}
# -----------------------------
# Attach Screenshot Folder Link
# -----------------------------
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
outcome = yield
report = outcome.get_result()
if report.when == "call":
context = item.funcargs.get("test_context", None)
if context:
folder = context["folder"]
steps = context.get("steps", [])
if report.failed:
driver = item.funcargs.get("driver", None)
failure_index = len(steps) + 1
failure_name = "failure_state"
failure_filename = f"{failure_index:02d}_{failure_name}.png"
failure_path = os.path.join(folder, failure_filename)
if driver:
try:
driver.save_screenshot(failure_path)
steps.append(
{
"index": failure_index,
"name": failure_name,
"path": failure_path,
"status": "failed",
}
)
except Exception:
pass
if item.config.pluginmanager.hasplugin("html"):
from pytest_html import extras
extras_list = getattr(report, "extras", [])
for step in steps:
extras_list.append(
extras.url(
_as_file_url(step["path"]),
name=f"Step {step['index']:02d} [{step['status']}] {step['name']}"
)
)
if steps:
extras_list.append(
extras.html(_build_steps_html(steps))
)
extras_list.append(
extras.url(_as_file_url(folder), name="Screenshots Folder")
)
log_folder = context.get("log_folder")
if log_folder:
extras_list.append(
extras.url(_as_file_url(log_folder), name="Logs Folder")
)
for extra_folder in context.get("extra_folders", []):
extras_list.append(
extras.url(
_as_file_url(extra_folder["path"]),
name=extra_folder["name"]
)
)
report.extras = extras_list
def _as_file_url(path):
absolute_path = os.path.abspath(path).replace("\\", "/")
return f"file:///{absolute_path}"
def _build_steps_html(steps):
rows = []
for step in steps:
step_url = _as_file_url(step["path"])
rows.append(
"<tr>"
f"<td>{step['index']:02d}</td>"
f"<td>{step['status'].capitalize()}</td>"
f"<td>{step['name']}</td>"
f"<td><a href=\"{step_url}\">Open Screenshot</a></td>"
"</tr>"
)
return (
"<div>"
"<h4>Step Details</h4>"
"<table border='1' cellpadding='4' cellspacing='0' "
"style='border-collapse: collapse; margin-top: 8px;'>"
"<thead><tr><th>#</th><th>Status</th><th>Step</th><th>Screenshot</th></tr></thead>"
f"<tbody>{''.join(rows)}</tbody>"
"</table>"
"</div>"
)
def _get_run_prefix(config):
invocation_args = " ".join(str(arg) for arg in config.invocation_params.args).lower()
if "tests/setup/sessions" in invocation_args or "test_session.py" in invocation_args:
return "PXIL_SESSION"
if "tests/setup/tokens" in invocation_args or "test_trading_token.py" in invocation_args:
return "PXIL_TOKEN"
runtime_path = os.path.abspath(os.path.join("config", "runtime.json"))
if os.path.exists(runtime_path):
try:
with open(runtime_path, "r", encoding="utf-8") as runtime_file:
runtime_config = json.load(runtime_file)
execution = runtime_config.get("execution", {})
product_family = execution.get("product_family", "regression").upper()
flow_scope = execution.get("flow_scope", "run").upper()
return f"{product_family}_{flow_scope}"
except Exception:
pass
if "matching" in invocation_args:
return "REGRESSION_MATCHING"
if "bidding" in invocation_args:
return "REGRESSION_BIDDING"
if "reports" in invocation_args:
return "REGRESSION_REPORTS"
if "order_book" in invocation_args:
return "REGRESSION_ORDER_BOOK"
if "trade_book" in invocation_args:
return "REGRESSION_TRADE_BOOK"
return "PXIL_RUN"
+393
View File
@@ -0,0 +1,393 @@
"""
generate_bid_json.py
====================
Bulk JSON generator for bid submission payloads.
FOLDER STRUCTURE (input_root):
input_root/
<segment>/ e.g. GDAM | DAM | HPDAM
<order_type>/ e.g. Single | Block
<order_side>/ e.g. Buy | Sell
<TM_folder>/ e.g. C1001 → becomes userId AND tmId
<tacId>.xlsx ← one file per TAC / bid sheet
SEGMENT → JSON "segment" MAPPING:
GDAM → "GREEN"
DAM → "CONVENTIONAL"
HPDAM → "HPDAM"
SEGMENT → JSON order "source" MAPPING:
GDAM → "SL"
DAM → "CO"
HPDAM → "HP"
ORDER SIDE → JSON "orderSide" MAPPING:
Buy → "B"
Sell → "S"
ORDER TYPE → JSON "orderType" MAPPING:
Single → "N"
Block → "B"
EXCEL LAYOUT (every .xlsx):
Row 1 : "Select Source" | "Order 1" | "Order 2" ... "Order N"
Row 2 : "<label>" | <price1> | <price2> ... <priceN>
Row 3+: "HH:MM-HH:MM [slot]" | <qty1> | <qty2> ...
(rows where ALL qty cells are blank are skipped)
OUTPUT:
output_root/
<segment>/
<order_type>/
<order_side>/
<TM_folder>/
<tacId>_bid.json
CONFIGURABLE via CLI (or edit DEFAULTS below):
--input_root path to input folder tree
--output_root path where JSON tree is written
--product_code e.g. "IDAS(01)"
--password password applied to every payload
--ocf_opted Y or N
--ocf_premium_disc integer
--ocf_qty integer
FIXED / AUTO-DERIVED:
transcode = "3040" (always)
requestType = "N" (always)
productType = "I" (always)
userId = TM folder name
tmId = TM folder name
segment = derived from segment folder name (see mapping above)
orderSide = derived from order_side folder name
orderType = derived from order_type folder name
deliveryDate = tomorrow's date (auto-calculated at run time)
USAGE:
python generate_bid_json.py \\
--input_root ./input \\
--output_root ./output \\
--product_code "IDAS(01)" \\
--password "your_password"
Or just edit DEFAULTS below and run:
python generate_bid_json.py
"""
import argparse
import json
import re
import sys
from datetime import date, timedelta
from pathlib import Path
from openpyxl import load_workbook
# ─────────────────────────────────────────────────────────────────────────────
# DEFAULTS – edit here or override via CLI
# ─────────────────────────────────────────────────────────────────────────────
DEFAULTS = {
"input_root": "D:/Load testing input json/Input scenario upload files",
"output_root": "D:/Load testing input json/UAT_Team_10%_Load_File",
"product_code": "IDASTEST25",
"password": "CHANGE_ME",
"ocf_opted": "N",
"ocf_premium_disc": 0,
"ocf_qty": 0,
}
# ─────────────────────────────────────────────────────────────────────────────
# FIXED CONSTANTS
# ─────────────────────────────────────────────────────────────────────────────
TRANSCODE = "3040"
REQUEST_TYPE = "N"
PRODUCT_TYPE = "I"
# Segment folder name → JSON segment value
SEGMENT_MAP = {
"GDAM": "GREEN",
"DAM": "Conventional",
"HPDAM": "HPDAM",
}
# Order-type folder name → JSON orderType value
ORDER_TYPE_MAP = {
"Single": "N",
"Block": "B",
}
# Order-side folder name → JSON orderSide value
ORDER_SIDE_MAP = {
"Buy": "B",
"Sell": "S",
}
# Segment folder name → JSON order "source" value
SOURCE_MAP = {
"GDAM": "SL",
"DAM": "CO",
"HPDAM": "HP",
}
# ─────────────────────────────────────────────────────────────────────────────
# HELPERS
# ─────────────────────────────────────────────────────────────────────────────
def tomorrow_str() -> str:
"""Return tomorrow's date as 'DD-Mon-YYYY', e.g. '18-Jun-2026'."""
d = date.today() + timedelta(days=1)
# %-d is Linux-only; %#d is Windows-only — use manual strip for cross-platform
return d.strftime("%d-%b-%Y").lstrip("0")
def parse_excel(xlsx_path: Path, order_type_val: str, order_side_val: str, source_val: str, seg_folder: str, cfg: dict) -> list[dict]:
"""
Parse one bid Excel file and return a list of order dicts.
tacId is the filename stem (e.g. 'MCOC100500002').
"""
wb = load_workbook(str(xlsx_path), read_only=True, data_only=True)
ws = wb.active
rows = list(ws.iter_rows(values_only=True))
if len(rows) < 3:
raise ValueError(f"File has fewer than 3 rows: {xlsx_path}")
# Row 2 (index 1): price row
price_row = rows[1]
prices = []
for cell in price_row[1:]:
if cell is None or cell == "":
break
prices.append(int(round(float(cell))))
if not prices:
raise ValueError(f"No price columns found in row 2 of: {xlsx_path}")
tac_id = xlsx_path.stem
orders = []
for row in rows[2:]:
label = row[0]
if not label:
continue
match = re.search(r'\[(\d+)\]', str(label))
if not match:
continue
slot_num = int(match.group(1))
qtys = []
for i in range(len(prices)):
raw = row[i + 1] if (i + 1) < len(row) else None
qtys.append(round(abs(float(raw)), 4) if (raw is not None and raw != "") else None)
# Skip slots where every qty cell is blank
if all(q is None for q in qtys):
continue
pq_data = [
{"price": price, "qty": qty}
for price, qty in zip(prices, qtys)
if qty is not None
]
order = {
"tacId": tac_id,
"fromSlot": slot_num,
"toSlot": slot_num,
"orderType": order_type_val,
"orderSide": order_side_val,
"source": source_val,
"sourceType": "",
}
# OCF fields are excluded for HPDAM segment
if seg_folder != "HPDAM":
order["ocfOpted"] = cfg["ocf_opted"]
order["ocfPremiumDisc"] = cfg["ocf_premium_disc"]
order["ocfQty"] = cfg["ocf_qty"]
order["pqData"] = pq_data
orders.append(order)
return orders
def build_payload(tm_id: str, segment_val: str, orders: list[dict], cfg: dict) -> dict:
return {
"transcode": TRANSCODE,
"userId": tm_id,
"password": cfg["password"],
"tmId": tm_id,
"requestType": REQUEST_TYPE,
"productCode": cfg["product_code"],
"productType": PRODUCT_TYPE,
"segment": segment_val,
"deliveryDate": tomorrow_str(),
"orders": orders,
}
# ─────────────────────────────────────────────────────────────────────────────
# MAIN WALK
# ─────────────────────────────────────────────────────────────────────────────
def process_all(cfg: dict):
input_root = Path(cfg["input_root"])
output_root = Path(cfg["output_root"])
if not input_root.exists():
print(f"[ERROR] input_root not found: {input_root}", file=sys.stderr)
sys.exit(1)
total_files = 0
total_errors = 0
print(f"Delivery date (tomorrow): {tomorrow_str()}")
print(f"Input : {input_root.resolve()}")
print(f"Output : {output_root.resolve()}")
print("-" * 72)
# ── Level 1: segment folder (GDAM / DAM / HPDAM) ─────────────────────────
for seg_dir in sorted(input_root.iterdir()):
if not seg_dir.is_dir() or seg_dir.name.startswith("."):
continue
seg_folder = seg_dir.name
segment_val = SEGMENT_MAP.get(seg_folder)
if segment_val is None:
print(f"[WARN] Unknown segment folder '{seg_folder}' — skipping.")
continue
source_val = SOURCE_MAP[seg_folder] # guaranteed present since SEGMENT_MAP check passed
# ── Level 2: order type folder (Single / Block) ───────────────────────
for ot_dir in sorted(seg_dir.iterdir()):
if not ot_dir.is_dir():
continue
ot_folder = ot_dir.name
if ot_folder == "Block":
print(f" [SKIP] Block order-type folder ignored: {ot_dir.relative_to(input_root)}")
continue
order_type_val = ORDER_TYPE_MAP.get(ot_folder)
if order_type_val is None:
print(f"[WARN] Unknown order-type folder '{ot_folder}' — skipping.")
continue
# ── Level 3: order side folder (Buy / Sell) ───────────────────────
for os_dir in sorted(ot_dir.iterdir()):
if not os_dir.is_dir():
continue
os_folder = os_dir.name
order_side_val = ORDER_SIDE_MAP.get(os_folder)
if order_side_val is None:
print(f"[WARN] Unknown order-side folder '{os_folder}' — skipping.")
continue
# Fix 2: source is blank for Buy orders, segment-mapped only for Sell
effective_source = "" if os_folder == "Buy" else source_val
# ── Level 4: TM folder (e.g. C1001) ──────────────────────────
for tm_dir in sorted(os_dir.iterdir()):
if not tm_dir.is_dir():
continue
tm_id = tm_dir.name # → userId AND tmId
# ── Level 5: xlsx files ───────────────────────────────────
for xlsx_file in sorted(tm_dir.glob("*.xlsx")):
total_files += 1
rel_path = xlsx_file.relative_to(input_root)
print(f" {rel_path}", end=" ... ")
try:
orders = parse_excel(
xlsx_file,
order_type_val=order_type_val,
order_side_val=order_side_val,
source_val=effective_source,
seg_folder=seg_folder,
cfg=cfg,
)
payload = build_payload(
tm_id=tm_id,
segment_val=segment_val,
orders=orders,
cfg=cfg,
)
# Output lands directly in output_root/<segment>/
out_dir = output_root / seg_folder
out_dir.mkdir(parents=True, exist_ok=True)
out_file = out_dir / f"{xlsx_file.stem}_bid.json"
with open(out_file, "w") as f:
json.dump(payload, f, indent=4)
print(f"OK → {out_file.relative_to(output_root)} ({len(orders)} orders)")
except Exception as exc:
print(f"FAILED — {exc}", file=sys.stderr)
total_errors += 1
print("-" * 72)
print(f"Done. {total_files} file(s) processed, {total_errors} error(s).")
# ─────────────────────────────────────────────────────────────────────────────
# CLI
# ─────────────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="Bulk-generate bid JSON files from Excel bid sheets.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Folder structure expected under --input_root:
<segment>/ GDAM | DAM | HPDAM
<order_type>/ Single | Block
<order_side>/ Buy | Sell
<TM>/ e.g. C1001 (becomes userId + tmId)
*.xlsx bid data files
""",
)
parser.add_argument(
"--input_root",
default=DEFAULTS["input_root"],
help="Root folder containing segment/order_type/order_side/TM/file.xlsx tree (default: %(default)s)",
)
parser.add_argument(
"--output_root",
default=DEFAULTS["output_root"],
help="Root folder where JSON files are written (same sub-tree) (default: %(default)s)",
)
parser.add_argument(
"--product_code",
default=DEFAULTS["product_code"],
help='Product code string, e.g. "IDAS(01)" (default: %(default)s)',
)
parser.add_argument(
"--password",
default=DEFAULTS["password"],
help="Password applied to every payload (default: %(default)s)",
)
parser.add_argument(
"--ocf_opted",
default=DEFAULTS["ocf_opted"],
help="OCF opted flag Y/N (default: %(default)s)",
)
parser.add_argument(
"--ocf_premium_disc",
type=int,
default=DEFAULTS["ocf_premium_disc"],
help="OCF premium discount integer (default: %(default)s)",
)
parser.add_argument(
"--ocf_qty",
type=int,
default=DEFAULTS["ocf_qty"],
help="OCF quantity integer (default: %(default)s)",
)
args = parser.parse_args()
process_all(vars(args))
if __name__ == "__main__":
main()
+560
View File
@@ -0,0 +1,560 @@
import json
import os
from datetime import datetime
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys
from selenium.common.exceptions import NoAlertPresentException, TimeoutException, StaleElementReferenceException
from .locators import Locator
from .config_reader import (
get_login_captcha_wait_seconds,
get_login_post_submit_wait_seconds,
get_step_delay_seconds,
)
import time
import msvcrt
STEP_DELAY_SECONDS = get_step_delay_seconds()
LOGIN_CAPTCHA_WAIT_SECONDS = get_login_captcha_wait_seconds()
LOGIN_POST_SUBMIT_WAIT_SECONDS = get_login_post_submit_wait_seconds()
def pause_after_step(seconds=STEP_DELAY_SECONDS):
time.sleep(seconds)
def wait_for_manual_input(prompt="Press Enter in the terminal when you want to finish this run..."):
print(prompt)
try:
input()
except EOFError as exc:
raise RuntimeError(
"Manual input is required for this test. Run pytest with -s so Enter can be captured."
) from exc
def clear_pending_console_input():
while msvcrt.kbhit():
msvcrt.getwch()
def enter_pressed():
if not msvcrt.kbhit():
return False
pressed = False
while msvcrt.kbhit():
key = msvcrt.getwch()
if key in ("\r", "\n"):
pressed = True
return pressed
def login(driver, username, password):
wait = WebDriverWait(driver, 20)
# username
userid_field = wait.until(
EC.visibility_of_element_located((By.XPATH, Locator.userid))
)
userid_field.send_keys(username)
# password
password_field = driver.find_element(By.XPATH, Locator.password)
password_field.send_keys(password)
print(f"Solve captcha within {LOGIN_CAPTCHA_WAIT_SECONDS} seconds...")
# wait for manual captcha solve
time.sleep(LOGIN_CAPTCHA_WAIT_SECONDS)
# click submit
submit_btn = wait.until(
EC.element_to_be_clickable((By.XPATH, Locator.submitbtn))
)
submit_btn.click()
time.sleep(LOGIN_POST_SUBMIT_WAIT_SECONDS)
accept_browser_alert_if_present(driver)
# wait for dashboard
wait.until(
EC.visibility_of_element_located((By.XPATH, Locator.dashboard_marker))
)
close_tac_expiry_modal_if_present(driver)
print("Login successful")
def accept_browser_alert_if_present(driver, timeout=5):
try:
alert = WebDriverWait(driver, timeout).until(EC.alert_is_present())
print(f"Browser alert accepted: {alert.text}")
alert.accept()
return True
except (TimeoutException, NoAlertPresentException):
return False
def close_tac_expiry_modal_if_present(driver, timeout=5):
try:
close_button = WebDriverWait(driver, timeout).until(
EC.element_to_be_clickable((By.XPATH, Locator.tacexpiryclosebtn))
)
except TimeoutException:
return False
try:
close_button.click()
except Exception:
driver.execute_script("arguments[0].click();", close_button)
print("Closed TAC Expiry modal")
return True
def handle_transient_popups(driver):
accepted_alert = accept_browser_alert_if_present(driver, timeout=2)
closed_modal = close_tac_expiry_modal_if_present(driver, timeout=2)
return accepted_alert or closed_modal
def backup_current_session_token(driver):
session_backup = {
"current_url": driver.current_url,
"cookies": driver.get_cookies(),
}
print(f"Backed up current browser session with {len(session_backup['cookies'])} cookies")
return session_backup
def save_session_backup(driver, test_context, name="session_backup"):
backup = backup_current_session_token(driver)
backup["created_at"] = datetime.now().isoformat(timespec="seconds")
path = os.path.join(test_context["folder"], f"{name}.json")
with open(path, "w", encoding="utf-8") as backup_file:
json.dump(backup, backup_file, indent=2)
test_context.setdefault("extra_folders", [])
print(f"Session backup saved: {path}")
return backup
def logout(driver):
wait = WebDriverWait(driver, 20)
handle_transient_popups(driver)
try:
user_menu = wait.until(
EC.element_to_be_clickable((By.XPATH, Locator.usermenutoggle))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", user_menu)
try:
user_menu.click()
except Exception:
driver.execute_script("arguments[0].click();", user_menu)
wait.until(
EC.visibility_of_element_located(
(
By.XPATH,
"//li[contains(@class,'dropdown') and contains(@class,'navbar-user') and contains(@class,'open')]"
)
)
)
logout_link = wait.until(
EC.element_to_be_clickable((By.XPATH, Locator.logoutlink))
)
try:
logout_link.click()
except Exception:
driver.execute_script("arguments[0].click();", logout_link)
accept_browser_alert_if_present(driver, timeout=3)
wait.until(
EC.visibility_of_element_located((By.XPATH, Locator.userid))
)
print("Logout successful")
return True
except TimeoutException:
print("Logout link was not found; browser will still close from fixture teardown")
return False
def select_ui_dropdown(driver, dropdown_label, value):
wait = WebDriverWait(driver, 20)
# 1️⃣ Open dropdown
dropdown = wait.until(
EC.element_to_be_clickable(
(By.XPATH, f"//span[@aria-label='{dropdown_label} activate']")
)
)
dropdown.click()
# 2️⃣ Find search box
search_box = wait.until(
EC.visibility_of_element_located((By.CSS_SELECTOR, "input.ui-select-search"))
)
# 3️⃣ Type product name
search_box.clear()
search_box.send_keys(value)
# 4️⃣ Wait for dropdown options to load
option_xpath = f"//div[contains(@class,'ui-select-choices-row')]//span[contains(text(),'{value}')]"
option = wait.until(
EC.element_to_be_clickable((By.XPATH, option_xpath))
)
# 5️⃣ Click option
option.click()
print(f"{value} selected successfully")
def select_labeled_ui_dropdown(driver, field_label, value):
wait = WebDriverWait(driver, 20)
dropdown_xpath = (
f"//label[normalize-space()='{field_label}']"
"/following::span[contains(@class,'ui-select-toggle')][1]"
)
dropdown = wait.until(
EC.element_to_be_clickable(
(
By.XPATH,
dropdown_xpath
)
)
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", dropdown)
try:
dropdown.click()
except Exception:
driver.execute_script("arguments[0].click();", dropdown)
search_xpath = (
f"//label[normalize-space()='{field_label}']"
"/following::input[contains(@class,'ui-select-search')][1]"
)
search_box = wait.until(
EC.visibility_of_element_located(
(By.XPATH, search_xpath)
)
)
search_box.click()
search_box.send_keys(Keys.CONTROL, "a")
search_box.clear()
search_box.send_keys(value)
wait.until(
EC.presence_of_element_located(
(By.XPATH, "//div[contains(@class,'ui-select-choices-row')]")
)
)
def _visible_choice_rows():
rows = driver.find_elements(By.XPATH, "//div[contains(@class,'ui-select-choices-row')]")
return [row for row in rows if row.is_displayed()]
option = None
visible_rows = _visible_choice_rows()
for row in visible_rows:
if row.text.strip() == value:
option = row
break
if option is None:
available_options = [row.text.strip() for row in visible_rows if row.text.strip()]
raise AssertionError(
f"Exact option '{value}' was not available in {field_label}. "
f"Visible options: {available_options}"
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", option)
try:
option.click()
except Exception:
driver.execute_script("arguments[0].click();", option)
print(f"{value} selected successfully from {field_label}")
def open_trading_maintenance_page(driver, menu_xpath):
wait = WebDriverWait(driver, 20)
trading_maintenance = wait.until(
EC.presence_of_element_located((By.XPATH, Locator.tradingmaintenance))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", trading_maintenance)
submenu = driver.find_elements(By.XPATH, menu_xpath)
submenu_visible = any(item.is_displayed() for item in submenu)
if not submenu_visible:
try:
wait.until(EC.element_to_be_clickable((By.XPATH, Locator.tradingmaintenance)))
trading_maintenance.click()
except Exception:
driver.execute_script("arguments[0].click();", trading_maintenance)
menu_item = wait.until(
EC.element_to_be_clickable((By.XPATH, menu_xpath))
)
try:
menu_item.click()
except Exception:
driver.execute_script("arguments[0].click();", menu_item)
def click_visible_menu_item(driver, menu_xpath):
wait = WebDriverWait(driver, 20)
menu_items = driver.find_elements(By.XPATH, menu_xpath)
for menu_item in menu_items:
if menu_item.is_displayed():
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", menu_item)
try:
menu_item.click()
except Exception:
driver.execute_script("arguments[0].click();", menu_item)
return
menu_item = wait.until(
EC.element_to_be_clickable((By.XPATH, menu_xpath))
)
try:
menu_item.click()
except Exception:
driver.execute_script("arguments[0].click();", menu_item)
def type_into_field(driver, locator, value):
wait = WebDriverWait(driver, 20)
field = wait.until(
EC.visibility_of_element_located(locator)
)
field.clear()
field.send_keys(value)
def select_bootstrap_dropdown_by_id(driver, element_id, value):
wait = WebDriverWait(driver, 20)
container = wait.until(
EC.presence_of_element_located((By.ID, element_id))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", container)
toggle_xpath = (
f"//div[@id='{element_id}']//span[contains(@class,'ui-select-toggle')]"
f"|//div[@id='{element_id}']//span[contains(@class,'btn-default')]"
f"|//div[@id='{element_id}']//span[contains(@class,'ui-select-placeholder')]"
f"|//div[@id='{element_id}']//input[contains(@class,'ui-select-search')]"
)
toggle = wait.until(
EC.presence_of_element_located((By.XPATH, toggle_xpath))
)
try:
wait.until(EC.element_to_be_clickable((By.XPATH, toggle_xpath)))
toggle.click()
except Exception:
driver.execute_script("arguments[0].click();", toggle)
try:
search_box = wait.until(
EC.visibility_of_element_located(
(By.XPATH, f"//div[@id='{element_id}']//input[contains(@class,'ui-select-search')]")
)
)
except TimeoutException:
search_box = wait.until(
EC.visibility_of_element_located(
(By.XPATH, "//input[contains(@class,'ui-select-search') and not(@aria-hidden='true')]")
)
)
search_box.click()
search_box.send_keys(Keys.CONTROL, "a")
search_box.clear()
search_box.send_keys(value)
option = wait.until(
EC.element_to_be_clickable(
(
By.XPATH,
"//div[contains(@class,'ui-select-choices-row')]"
f"//*[normalize-space()='{value}' or starts-with(normalize-space(),'{value}')]"
)
)
)
option.click()
print(f"{value} selected successfully from {element_id}")
def click_grid_button_for_row_text(driver, row_text, button_text):
wait = WebDriverWait(driver, 20)
button_xpath = (
"//div[contains(@class,'ui-grid-row')]"
f"[contains(.,'{row_text}')]"
f"//button[normalize-space()='{button_text}']"
)
last_error = None
for _ in range(3):
try:
button = wait.until(
EC.presence_of_element_located((By.XPATH, button_xpath))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", button)
wait.until(EC.element_to_be_clickable((By.XPATH, button_xpath)))
refreshed_button = driver.find_element(By.XPATH, button_xpath)
try:
refreshed_button.click()
except Exception:
driver.execute_script("arguments[0].click();", refreshed_button)
print(f"Clicked {button_text} for row containing {row_text}")
return
except StaleElementReferenceException as exc:
last_error = exc
time.sleep(1)
if last_error:
raise last_error
def click_first_grid_button(driver, button_text):
wait = WebDriverWait(driver, 20)
button_xpath = f"(//button[normalize-space()='{button_text}' and not(contains(@class,'ng-hide'))])[1]"
last_error = None
for _ in range(3):
try:
button = wait.until(
EC.presence_of_element_located((By.XPATH, button_xpath))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'start'});", button)
refreshed_button = driver.find_element(By.XPATH, button_xpath)
try:
refreshed_button.click()
except Exception:
driver.execute_script("arguments[0].click();", refreshed_button)
print(f"Clicked first visible {button_text}")
return
except StaleElementReferenceException as exc:
last_error = exc
time.sleep(1)
if last_error:
raise last_error
def type_first_visible_input(driver, xpath, value):
wait = WebDriverWait(driver, 20)
def _find_visible_input(_driver):
elements = _driver.find_elements(By.XPATH, xpath)
for element in elements:
if element.is_displayed() and element.is_enabled():
return element
return False
element = wait.until(_find_visible_input)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", element)
element.click()
element.clear()
element.send_keys(value)
def toggle_bootstrap_dropdown_by_id(driver, element_id):
wait = WebDriverWait(driver, 20)
toggle_xpath = (
f"//div[@id='{element_id}']//span[contains(@class,'ui-select-toggle')]"
f"|//div[@id='{element_id}']//span[contains(@class,'btn-default')]"
f"|//div[@id='{element_id}']//span[contains(@class,'ui-select-placeholder')]"
f"|//div[@id='{element_id}']//input[contains(@class,'ui-select-search')]"
)
toggle = wait.until(
EC.presence_of_element_located((By.XPATH, toggle_xpath))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", toggle)
try:
wait.until(EC.element_to_be_clickable((By.XPATH, toggle_xpath)))
toggle.click()
except Exception:
driver.execute_script("arguments[0].click();", toggle)
def click_page_body(driver):
body = WebDriverWait(driver, 20).until(
EC.presence_of_element_located((By.TAG_NAME, "body"))
)
body.click()
def toggle_labeled_ui_dropdown(driver, field_label):
wait = WebDriverWait(driver, 20)
dropdown_xpath = (
f"//label[normalize-space()='{field_label}']"
"/following::span[contains(@class,'ui-select-toggle')][1]"
)
dropdown = wait.until(
EC.element_to_be_clickable((By.XPATH, dropdown_xpath))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", dropdown)
try:
dropdown.click()
except Exception:
driver.execute_script("arguments[0].click();", dropdown)
+49
View File
@@ -0,0 +1,49 @@
class Locator(object):
userid="//input[@placeholder='User ID']"
password="//input[@placeholder='Password']"
submitbtn="//button[@value='Login']"
dashboard_marker="//span[normalize-space()='TRADING MAINTENANCE']"
tacexpiryclosebtn="//div[contains(@class,'modal') and contains(@style,'display: block')]//button[contains(@class,'close')]"
usermenutoggle="//li[contains(@class,'dropdown') and contains(@class,'navbar-user')]//a[contains(@class,'dropdown-toggle') and @data-toggle='dropdown']"
logoutlink="//li[contains(@class,'dropdown') and contains(@class,'navbar-user')]//ul[contains(@class,'dropdown-menu')]//a[@ng-click='logout()' and normalize-space()='Log Out']"
tradingmaintenance="//span[normalize-space()='TRADING MAINTENANCE']"
tradingsession="//a[normalize-space()='Trading Session Master']"
tokengeneration="//a[@href='#/tradingMaintenance/generateToken' or normalize-space()='Token Generation']"
tokenandtimesessionview="//a[@href='#/tradingMaintenance/tokenTimeSessionMap' or normalize-space()='Token and Time Session View']"
addsessionbtn="//button[contains(text(),'Add')]"
addnewtokenbtn="//button[@data-ng-click='addToken()' and normalize-space()='Add New Token']"
setmeinstance="//span[@aria-label='meInstanceId activate']"
#inputmeid="meInstanceId activate"
#"//li[@id='ui-select-choices-178']"
#"//i[@class='caret pull-right']"
#idasme="//li[text()='IDAS']"
inputstartdate="//input[@placeholder='Start Date']"
inputenddate="//input[@placeholder='End Date']"
sessionsavebtn = "//button[@data-ng-click='saveTrade()']"
tokenmarketcode = "mmMarketCode"
tokenproductcode = "mmPrdCode"
tokendeliverystartdate = "//input[@placeholder='DELIVERY_START_DATE']"
tokendeliveryenddate = "//input[@placeholder='DELIVERY_END_DATE']"
tokenexpirydate = "//input[@placeholder='Token Expiry Date']"
savetokenbtn = "//button[@data-ng-click='saveToken()' and normalize-space()='save Token']"
tokentimesessionbtn = "//button[@data-ng-click='redirectToTimeSession()' and normalize-space()='Time Session']"
viewproducttype = "pmProductType"
viewproductcode = "pmProductCode"
viewdeliveryfromdate = "//input[@placeholder='Delivery From Date']"
viewdeliverytodate = "//input[@placeholder='Delivery To Date']"
viewsearchbtn = "//button[@data-ng-click='getSearchList()' and normalize-space()='Search']"
savetokenandactivatebtn = (
"//button[@data-ng-click=\"saveTokenDetails('T')\" "
"and contains(normalize-space(),'Save') "
"and contains(normalize-space(),'Activate')]"
)
edittimesessionbtn = "//button[@data-ng-click='editSession()' and normalize-space()='Edit']"
savetimesessionbtn = "//button[@data-ng-click='updateSession()' and normalize-space()='Save']"
timesessionstartdate = "//input[@name='tsmTokenStartDateTime' or @placeholder='TokenStartDateTime']"
timesessionenddate = "//input[@name='TokenEndDateTime']"
timesessionstarttime = "//input[@name='startTime']"
timesessionendtime = "//input[@name='endTime']"
+36
View File
@@ -0,0 +1,36 @@
# Regression Suite Structure
This suite is split first by login role and then by product and business flow.
## Current layout
- `tests/setup/sessions/`: session creation and other session-level setup flows.
- `tests/setup/tokens/`: token generation and token/time-session setup flows.
- `tests/regression/exchange/`: end-to-end regression scenarios executed with exchange login.
- `tests/regression/trading_member/`: end-to-end regression scenarios executed with trading member login.
## Product-wise regression pattern
Create product folders under both login roles as needed:
- `tests/regression/exchange/<product>/bidding/`
- `tests/regression/exchange/<product>/matching/`
- `tests/regression/exchange/<product>/reports/`
- `tests/regression/exchange/<product>/order_book/`
- `tests/regression/exchange/<product>/trade_book/`
- `tests/regression/trading_member/<product>/bidding/`
- `tests/regression/trading_member/<product>/matching/`
- `tests/regression/trading_member/<product>/reports/`
- `tests/regression/trading_member/<product>/order_book/`
- `tests/regression/trading_member/<product>/trade_book/`
## Suggested naming
- Buy-side bidding flow: `test_buy_flow.py`
- Sell-side bidding flow: `test_sell_flow.py`
- Matching flow: `test_matching_flow.py`
- Report download flow: `test_report_downloads.py`
- Order book flow: `test_order_book.py`
- Trade book flow: `test_trade_book.py`
For now, `idas` is scaffolded under both roles as the first product. Additional products can be added by repeating the same folder pattern.
+1
View File
@@ -0,0 +1 @@
+152
View File
@@ -0,0 +1,152 @@
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
class TradingLoginLocator:
username = "txtUserID"
password = "txtPwd"
captcha_input = (
"//input["
"contains(@placeholder,'Capt') "
"or contains(@placeholder,'capt') "
"or contains(@id,'Captcha') "
"or contains(@name,'Captcha')"
"]"
)
login_button = "loginButtonLink"
class TradingMemberLoginLocator:
username = "uname"
username_continue_button = "submit_user"
password = "password"
captcha_input = "InputCaptcha"
password_continue_button = "submit_password"
class McoLoginLocator:
form = "loginform"
username = "#loginform input:not([type='hidden']):not([type='password'])"
password = "#loginform input[type='password']"
sign_in_button = "#loginform button[type='submit']"
def login_to_trading_application(driver, username, password, timing, login_role="exchange"):
if login_role == "trading_member":
return login_to_trading_member_application(driver, username, password, timing)
return login_to_exchange_application(driver, username, password, timing)
def login_to_exchange_application(driver, username, password, timing):
wait = WebDriverWait(driver, 20)
username_field = wait.until(
EC.visibility_of_element_located((By.ID, TradingLoginLocator.username))
)
username_field.clear()
username_field.send_keys(username)
password_field = wait.until(
EC.visibility_of_element_located((By.ID, TradingLoginLocator.password))
)
password_field.clear()
password_field.send_keys(password)
wait.until(
EC.visibility_of_element_located((By.XPATH, TradingLoginLocator.captcha_input))
)
captcha_wait_seconds = timing.get("login_captcha_wait_seconds", 10)
print(f"Enter trading captcha manually within {captcha_wait_seconds} seconds...")
time.sleep(captcha_wait_seconds)
login_button = wait.until(
EC.element_to_be_clickable((By.ID, TradingLoginLocator.login_button))
)
login_button.click()
post_submit_wait_seconds = timing.get("login_post_submit_wait_seconds", 10)
time.sleep(post_submit_wait_seconds)
def login_to_trading_member_application(driver, username, password, timing):
wait = WebDriverWait(driver, 20)
username_field = wait.until(
EC.visibility_of_element_located((By.ID, TradingMemberLoginLocator.username))
)
username_field.clear()
username_field.send_keys(username)
username_continue_button = wait.until(
EC.element_to_be_clickable((By.ID, TradingMemberLoginLocator.username_continue_button))
)
username_continue_button.click()
password_field = wait.until(
EC.visibility_of_element_located((By.ID, TradingMemberLoginLocator.password))
)
password_field.clear()
password_field.send_keys(password)
wait.until(
EC.visibility_of_element_located((By.ID, TradingMemberLoginLocator.captcha_input))
)
captcha_wait_seconds = timing.get("login_captcha_wait_seconds", 10)
print(f"Enter trading member captcha manually within {captcha_wait_seconds} seconds...")
time.sleep(captcha_wait_seconds)
password_continue_button = wait.until(
EC.element_to_be_clickable((By.ID, TradingMemberLoginLocator.password_continue_button))
)
password_continue_button.click()
post_submit_wait_seconds = timing.get("login_post_submit_wait_seconds", 10)
time.sleep(post_submit_wait_seconds)
def login_to_mco_application(driver, username, password, timing):
if not username or not password:
raise ValueError("Configure users.mco.username and users.mco.password in config/runtime.json.")
wait = WebDriverWait(driver, 20)
wait.until(EC.presence_of_element_located((By.ID, McoLoginLocator.form)))
username_field = wait.until(
EC.visibility_of_element_located((By.CSS_SELECTOR, McoLoginLocator.username))
)
username_field.clear()
username_field.send_keys(username)
password_field = wait.until(
EC.visibility_of_element_located((By.CSS_SELECTOR, McoLoginLocator.password))
)
password_field.clear()
password_field.send_keys(password)
sign_in_button = wait.until(
EC.element_to_be_clickable((By.CSS_SELECTOR, McoLoginLocator.sign_in_button))
)
sign_in_button.click()
wait.until(EC.url_contains("/Dashboard/"))
wait.until(
EC.visibility_of_element_located(
(By.XPATH, "//*[self::a or self::h1 or self::h2 or self::div][normalize-space()='Dashboard']")
)
)
post_submit_wait_seconds = timing.get("login_post_submit_wait_seconds", 10)
time.sleep(post_submit_wait_seconds)
+272
View File
@@ -0,0 +1,272 @@
import json
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
REGRESSION_CONFIG_DIR = PROJECT_ROOT / "config"
RUNTIME_CONFIG_PATH = REGRESSION_CONFIG_DIR / "runtime.json"
def _load_json(path):
with path.open("r", encoding="utf-8") as file:
return json.load(file)
def load_runtime_config():
return _load_json(RUNTIME_CONFIG_PATH)
def load_users_config():
runtime_config = load_runtime_config()
users_config = runtime_config.get("users")
if not users_config:
raise KeyError(f"Users config not found in {RUNTIME_CONFIG_PATH}.")
return users_config
def load_product_config(product_family=None):
runtime_config = load_runtime_config()
selected_product = product_family or runtime_config["execution"]["product_family"]
product_config = runtime_config.get(selected_product)
if not product_config:
product_config = runtime_config.get(selected_product.upper())
if not product_config:
product_config = runtime_config.get(selected_product.lower())
if not product_config:
raise KeyError(
f"Product config not found for '{selected_product}' in {RUNTIME_CONFIG_PATH}."
)
return product_config
def _resolve_login(users_config, execution_config):
login_role = execution_config["login_role"]
trading_member_key = execution_config.get("trading_member_key")
if login_role == "exchange":
return users_config["exchange"]
if login_role == "trading_member":
trading_member = users_config["trading_members"].get(trading_member_key)
if not trading_member:
raise KeyError(
f"Trading member '{trading_member_key}' is not defined in {RUNTIME_CONFIG_PATH}"
)
return {
"username": trading_member["username"],
"password": trading_member["password"],
"label": trading_member.get("label", trading_member_key),
"member_id": trading_member.get("member_id", ""),
"folder_upload_tm_ids": trading_member.get("folder_upload_tm_ids", []),
}
raise ValueError(
f"Unsupported login_role '{login_role}'. Use 'exchange' or 'trading_member'."
)
def _resolve_exchange_parallel_logins(users_config):
return [
{
"username": exchange_user["username"],
"password": exchange_user["password"],
"label": exchange_user.get("label", exchange_user["username"]),
}
for exchange_user in users_config.get("exchange_parallel", [])
if exchange_user.get("username") and exchange_user.get("password")
]
def _resolve_trading_member_parallel_logins(users_config):
return [
{
"username": trading_member["username"],
"password": trading_member["password"],
"label": trading_member.get("label", trading_member_key),
"key": trading_member_key,
"member_id": trading_member.get("member_id", ""),
"folder_upload_tm_ids": trading_member.get("folder_upload_tm_ids", []),
}
for trading_member_key, trading_member in users_config.get("trading_members", {}).items()
if trading_member.get("username") and trading_member.get("password")
]
def _resolve_trading_member(users_config, execution_config):
trading_member_key = execution_config.get("trading_member_key")
trading_member = users_config["trading_members"].get(trading_member_key)
if not trading_member:
raise KeyError(
f"Trading member '{trading_member_key}' is not defined in {RUNTIME_CONFIG_PATH}"
)
return trading_member
def _resolve_portal_urls(environment_config, execution_config):
active_environment = environment_config.get("active_environment")
profiles = environment_config.get("portal_environments", {})
profile = profiles.get(active_environment, {})
pratyay_urls = profile.get("pratyay", {})
login_role = execution_config["login_role"]
if profile:
if not pratyay_urls.get(login_role):
raise KeyError(
f"Pratyay URL for role '{login_role}' is not configured in "
f"environment.portal_environments.{active_environment}."
)
if not profile.get("mco"):
raise KeyError(
f"MCO URL is not configured in "
f"environment.portal_environments.{active_environment}."
)
return {
"environment": active_environment,
"pratyay": pratyay_urls[login_role],
"mco": profile["mco"],
}
return {
"environment": active_environment or "legacy",
"pratyay": _resolve_application_url(environment_config, execution_config),
"mco": "",
}
def _resolve_application_url(environment_config, execution_config):
login_role = execution_config["login_role"]
role_urls = environment_config.get("login_urls", {})
if login_role in role_urls and role_urls[login_role]:
return role_urls[login_role]
return environment_config["base_url"]
def load_regression_config():
runtime_config = load_runtime_config()
users_config = load_users_config()
execution_config = runtime_config["execution"]
product_config = load_product_config(execution_config["product_family"])
active_selection = product_config.get("active_selection", {})
defaults = product_config.get("defaults", {})
selected_product_code = execution_config.get(
"product_code",
active_selection.get("product_code", defaults["product_code"])
)
selected_trading_market = active_selection.get(
"trading_market",
defaults["trading_market"]
)
selected_trading_markets = active_selection.get("trading_markets", [])
selected_bidding_screen = active_selection.get(
"bidding_screen",
defaults["bidding_screen"]
)
trading_member = _resolve_trading_member(users_config, execution_config)
login = _resolve_login(users_config, execution_config)
portal_urls = _resolve_portal_urls(runtime_config["environment"], execution_config)
return {
"url": portal_urls["pratyay"],
"portal_urls": portal_urls,
"login": login,
"mco_login": users_config.get("mco", {}),
"exchange_parallel_logins": _resolve_exchange_parallel_logins(users_config),
"trading_member_parallel_logins": _resolve_trading_member_parallel_logins(users_config),
"product": {
"family": execution_config["product_family"],
"product_type": defaults["product_type"],
"product_code": selected_product_code,
"trading_market": selected_trading_market,
"trading_markets": selected_trading_markets,
"bidding_screen": selected_bidding_screen
},
"execution": execution_config,
"selected_role": execution_config["login_role"],
"selected_flow": execution_config["flow_scope"],
"selected_trading_member": trading_member,
"timing": runtime_config["timing"],
"available_product_codes": product_config.get("available_product_codes", []),
"available_trading_markets": product_config.get("available_trading_markets", []),
"available_bidding_screens": product_config.get("available_bidding_screens", []),
"flows": product_config.get("flows", {}),
"setup": product_config.get("setup", {}),
"exchange_bidding": _normalize_bidding_config(product_config.get("bidding", {})),
"exchange_matching": _normalize_matching_config(product_config.get("matching", {}))
}
def should_run_flow(flow_name):
selected_flow = load_regression_config()["selected_flow"]
return selected_flow == "e2e" or selected_flow == flow_name
def _normalize_bidding_config(raw_bidding_config):
if not raw_bidding_config:
return {}
entry_method = raw_bidding_config.get("entry_method", {})
data_paths = raw_bidding_config.get("data_paths", {})
selection = raw_bidding_config.get("selection", {})
timing = raw_bidding_config.get("timing", {})
return {
**entry_method,
**data_paths,
**selection,
**timing,
"block_bid": entry_method.get("block_bid", raw_bidding_config.get("block_bid", {})),
"excel_layout": raw_bidding_config.get("excel_layout", {}),
"entries": raw_bidding_config.get("entries", [])
}
def _normalize_matching_config(raw_matching_config):
if not raw_matching_config:
return {}
return {
"matching_product_code": raw_matching_config.get("matching_product_code", ""),
"matching_type": raw_matching_config.get("matching_type", ""),
"source_types": raw_matching_config.get("source_types", []),
"bid_transfer_markets": raw_matching_config.get(
"bid_transfer_markets",
["GDAM", "DAM", "HPDAM"]
),
"accepted_bid_transfer_statuses": raw_matching_config.get(
"accepted_bid_transfer_statuses",
["ACCEPTED", "PARTIALLY_ACCEPTED", "NO_BIDS"]
),
"initiation_mode": raw_matching_config.get("initiation_mode", ""),
"portals": raw_matching_config.get("portals", {}),
"download_folder_name": raw_matching_config.get("downloads", {}).get(
"download_folder_name",
"matching_files"
),
**raw_matching_config.get("timing", {})
}
+1
View File
@@ -0,0 +1 @@
+32
View File
@@ -0,0 +1,32 @@
# IDAS Exchange Regression
This area is for exchange-side IDAS regression flows.
## Current scaffold
- `bidding/test_idas_exchange_bidding_flow.py`: starting point for the exchange bidding flow.
## Config source
This regression suite reads from the new regression config layer:
- `config/runtime.json`
- `config/runtime.json > users`
- `config/runtime.json > IDAS`
## Current runtime expectations
- `execution.login_role` should be `exchange`
- `execution.product_family` should be `IDAS`
- `execution.flow_scope` can be `e2e` or `bidding`
## Planned expansion
As steps are finalized, we can add:
- exchange bidding buy flow
- exchange bidding sell flow
- matching verification
- report downloads
- order book filters
- trade book filters
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1,400 @@
import json
from pathlib import Path
import time
from datetime import datetime
from selenium.common.exceptions import NoAlertPresentException, TimeoutException
from selenium.webdriver.common.action_chains import ActionChains
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 select_select2_value_by_container_id
def prepare_block_bulk_upload_page(driver, config, entry, upload_method):
product = config["product"]
bidding_config = config["exchange_bidding"]
block_config = bidding_config.get("block_bid", {})
_click_tab(driver, upload_method)
_click_block_bid_tab(driver)
_select_product_type(driver, product["product_type"])
_select_first_available_select2_value(
driver,
block_config.get("product_container_id", "s2id_ddlproduct3"),
[
block_config.get("product_code"),
product["product_code"],
],
)
_select_first_available_select2_value(
driver,
block_config.get("segment_container_id", "s2id_ddlSegment3"),
[product["trading_market"]],
)
if config["selected_role"] != "trading_member":
_select_first_available_select2_value(
driver,
block_config.get("tm_container_id", "s2id_ddlTMID3"),
[entry["tm_id"]],
)
def upload_block_bulk_file(
driver,
file_path,
success_timeout_seconds=1800,
poll_seconds=2,
progress_log_interval_seconds=60,
progress_log_label="block_file_upload",
):
upload_path = Path(file_path)
if not upload_path.exists() or not upload_path.is_file():
raise FileNotFoundError(f"Block bid upload file not found: {upload_path}")
file_input = _find_block_upload_input(driver)
_prepare_file_input(driver, file_input)
selected_at = datetime.now()
selected_monotonic = time.monotonic()
file_input.send_keys(str(upload_path.resolve()))
_accept_browser_upload_confirmation_if_present(driver)
success = wait_for_block_upload_success(
driver,
timeout_seconds=success_timeout_seconds,
poll_seconds=poll_seconds,
progress_log_interval_seconds=progress_log_interval_seconds,
progress_log_label=progress_log_label,
)
return _build_upload_timing(selected_at, selected_monotonic, success, [upload_path])
def upload_block_bulk_folder(
driver,
folder_path,
success_timeout_seconds=1800,
poll_seconds=2,
progress_log_interval_seconds=60,
progress_log_label="block_folder_upload",
):
upload_path = Path(folder_path)
if not upload_path.exists() or not upload_path.is_dir():
raise FileNotFoundError(f"Block bid upload folder not found: {upload_path}")
upload_files = [path for path in upload_path.rglob("*") if path.is_file()]
file_input = _find_block_upload_input(driver)
_prepare_file_input(driver, file_input)
selected_at = datetime.now()
selected_monotonic = time.monotonic()
file_input.send_keys(str(upload_path.resolve()))
_accept_browser_upload_confirmation_if_present(driver)
success = wait_for_block_upload_success(
driver,
timeout_seconds=success_timeout_seconds,
poll_seconds=poll_seconds,
progress_log_interval_seconds=progress_log_interval_seconds,
progress_log_label=progress_log_label,
)
return _build_upload_timing(selected_at, selected_monotonic, success, upload_files)
def wait_for_block_upload_success(
driver,
timeout_seconds=1800,
poll_seconds=2,
progress_log_interval_seconds=60,
progress_log_label="block_upload",
):
end_time = time.monotonic() + timeout_seconds
next_progress_log = time.monotonic() + progress_log_interval_seconds
last_message = ""
while time.monotonic() < end_time:
message = _current_upload_message(driver)
if message:
last_message = message
if _is_success_message(message):
return {
"message": message,
"displayed_at": datetime.now().isoformat(timespec="seconds"),
}
if progress_log_interval_seconds and time.monotonic() >= next_progress_log:
print(
f"Block upload progress [{progress_log_label}]: "
f"waiting for success message. Last message: {last_message or 'none'}"
)
next_progress_log = time.monotonic() + progress_log_interval_seconds
time.sleep(poll_seconds)
raise TimeoutException(
"Block upload success message did not appear within the timeout. "
f"Last visible message: {last_message or 'No status message found'}"
)
def save_block_bulk_upload_log(test_context, market, entry, upload_method, timing):
folder = Path(test_context.get("log_folder", test_context["folder"]))
folder.mkdir(parents=True, exist_ok=True)
side_name = entry.get("side", "")
log_name_parts = ["block_bulk_upload", upload_method, market]
if side_name:
log_name_parts.append(side_name)
log_name_parts.append(entry["tm_id"])
path = folder / f"{'_'.join(log_name_parts)}.json"
payload = {
"market": market,
"side": side_name,
"tm_id": entry["tm_id"],
"folder_name": entry.get("folder_name", ""),
"folder_path": entry.get("folder_path", ""),
"file_name": entry.get("file_name", ""),
"file_path": entry.get("upload_path", ""),
"upload_method": upload_method,
"timing": timing,
}
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
return str(path)
def _click_tab(driver, upload_method):
tab_href = "#Folder" if upload_method == "folder_upload" else "#File"
_click_by_xpath(
driver,
f"//a[@href='{tab_href}' or normalize-space()='{tab_href.lstrip('#')}']"
)
def _click_block_bid_tab(driver):
if _block_bid_fields_are_visible(driver):
return
_click_visible_by_xpath(
driver,
"//a[contains(normalize-space(),'Block Bid')]"
"|//span[contains(normalize-space(),'Block Bid')]/ancestor::a[1]"
"|//*[self::button][contains(normalize-space(),'Block Bid')]"
)
WebDriverWait(driver, 20).until(lambda _driver: _block_bid_fields_are_visible(_driver))
def _select_product_type(driver, product_type):
wait = WebDriverWait(driver, 20)
candidates = driver.find_elements(
By.XPATH,
"//select[contains(@id,'ProductType') or contains(@name,'ProductType')]"
"|//select[option[normalize-space()='IDAS']]"
)
for candidate in candidates:
if not candidate.is_displayed():
continue
try:
from selenium.webdriver.support.select import Select
Select(candidate).select_by_visible_text(product_type)
return
except Exception:
continue
wait.until(lambda _driver: candidates or True)
def _select_first_available_select2_value(driver, container_id, values):
last_error = None
for value in values:
if not value:
continue
try:
select_select2_value_by_container_id(
driver,
container_id,
value,
exact=True
)
return value
except TimeoutException as exc:
last_error = exc
if last_error:
raise last_error
raise ValueError(f"No values configured for select2 container '{container_id}'.")
def _find_block_upload_input(driver):
wait = WebDriverWait(driver, 30)
return wait.until(
lambda _driver: _first_displayed_or_existing(
_driver.find_elements(
By.XPATH,
"//input[@type='file' and ("
"@id='filepicker3' or contains(@id,'filepicker') or "
"contains(@id,'file') or contains(@name,'file') or @webkitdirectory"
")]"
)
) or False,
"No block bid upload input was found."
)
def _first_displayed_or_existing(elements):
for element in elements:
if element.is_displayed() and element.is_enabled():
return element
if elements:
return elements[0]
return None
def _prepare_file_input(driver, file_input):
driver.execute_script(
"""
arguments[0].style.display = 'block';
arguments[0].style.visibility = 'visible';
arguments[0].style.opacity = 1;
arguments[0].style.height = '20px';
arguments[0].style.width = '120px';
arguments[0].removeAttribute('disabled');
arguments[0].value = '';
""",
file_input
)
def _click_by_xpath(driver, xpath):
wait = WebDriverWait(driver, 20)
element = wait.until(EC.element_to_be_clickable((By.XPATH, xpath)))
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", element)
try:
element.click()
except Exception:
driver.execute_script("arguments[0].click();", element)
def _click_visible_by_xpath(driver, xpath):
wait = WebDriverWait(driver, 20)
element = wait.until(lambda _driver: _first_clickable(_driver, xpath) or False)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", element)
try:
element.click()
except Exception:
driver.execute_script("arguments[0].click();", element)
def _first_clickable(driver, xpath):
for element in driver.find_elements(By.XPATH, xpath):
try:
if element.is_displayed() and element.is_enabled():
return element
except Exception:
continue
return None
def _block_bid_fields_are_visible(driver):
product_fields = driver.find_elements(By.ID, "s2id_ddlproduct3")
segment_fields = driver.find_elements(By.ID, "s2id_ddlSegment3")
upload_inputs = driver.find_elements(By.ID, "filepicker3")
return any(_is_visible(element) for element in product_fields) and (
any(_is_visible(element) for element in segment_fields)
or any(_is_visible(element) for element in upload_inputs)
)
def _is_visible(element):
try:
return element.is_displayed()
except Exception:
return False
def _accept_browser_upload_confirmation_if_present(driver):
try:
alert = driver.switch_to.alert
alert.accept()
return
except NoAlertPresentException:
pass
try:
ActionChains(driver).send_keys("\ue007").perform()
except Exception:
pass
def _current_upload_message(driver):
candidates = []
for locator in (
(By.ID, "successMsg"),
(By.ID, "divSuccess"),
(By.ID, "divInfo"),
(By.ID, "divError"),
):
candidates.extend(driver.find_elements(*locator))
candidates.extend(
driver.find_elements(
By.XPATH,
"//*[contains(@class,'alert') and not(contains(@style,'display: none'))]"
)
)
for candidate in candidates:
try:
if candidate.is_displayed():
text = candidate.text.strip()
if text:
return text
except Exception:
continue
return ""
def _is_success_message(message):
normalized_message = " ".join(str(message or "").split()).lower()
return (
"folder_uploaded_successfully" in normalized_message
or "folder uploaded successfully" in normalized_message
or "file_uploaded_successfully" in normalized_message
or "file uploaded successfully" in normalized_message
)
def _build_upload_timing(selected_at, selected_monotonic, success, upload_files):
success_monotonic = time.monotonic()
return {
"files_selected_at": selected_at.isoformat(timespec="seconds"),
"success_displayed_at": success["displayed_at"],
"duration_seconds": round(success_monotonic - selected_monotonic, 3),
"success_message": success["message"],
"file_count": len(upload_files),
"files": [str(path) for path in upload_files],
}
@@ -0,0 +1,167 @@
from pathlib import Path
import re
import zipfile
from xml.etree import ElementTree as ET
MAIN_NS = {"a": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
REL_NS = {"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships"}
PACKAGE_REL_NS = {"pr": "http://schemas.openxmlformats.org/package/2006/relationships"}
def load_single_bid_market_matrix(excel_path, trading_market, sheet_name="Sheet1", excel_layout=None):
excel_file = Path(excel_path)
if not excel_file.exists():
raise FileNotFoundError(f"Excel input file not found: {excel_file}")
layout = excel_layout or {}
with zipfile.ZipFile(excel_file) as workbook_zip:
shared_strings = _load_shared_strings(workbook_zip)
sheet_xml_path = _resolve_sheet_xml_path(workbook_zip, sheet_name)
sheet_root = ET.fromstring(workbook_zip.read(sheet_xml_path))
cell_map = _load_cell_map(sheet_root, shared_strings)
start_column = layout.get("column_start", "B")
market_price_rows = layout.get("market_price_rows", {})
common_data_start_row = int(layout.get("common_data_start_row", 6))
price_row = int(market_price_rows.get(trading_market, 2))
_, start_col = _split_cell_reference(f"{start_column}1")
max_row, max_col = _max_used_extent(cell_map, price_row, start_col)
matrix = []
for row_index in [price_row, *range(common_data_start_row, max_row + 1)]:
row_values = []
for col_index in range(start_col, max_col + 1):
row_values.append(cell_map.get((row_index, col_index), ""))
matrix.append(row_values)
matrix = _trim_trailing_empty_rows_and_columns(matrix)
if not matrix or not matrix[0]:
raise ValueError(f"No bidding data found in {excel_file} for market {trading_market}.")
tsv = "\n".join("\t".join(value for value in row) for row in matrix)
return {
"matrix": matrix,
"tsv": tsv,
"row_count": len(matrix),
"column_count": len(matrix[0]),
"first_value": matrix[0][0],
}
def _load_shared_strings(workbook_zip):
try:
root = ET.fromstring(workbook_zip.read("xl/sharedStrings.xml"))
except KeyError:
return []
values = []
for item in root.findall("a:si", MAIN_NS):
text_parts = [node.text or "" for node in item.findall(".//a:t", MAIN_NS)]
values.append("".join(text_parts))
return values
def _resolve_sheet_xml_path(workbook_zip, sheet_name):
workbook_root = ET.fromstring(workbook_zip.read("xl/workbook.xml"))
relationships_root = ET.fromstring(workbook_zip.read("xl/_rels/workbook.xml.rels"))
relationship_map = {
rel.attrib["Id"]: rel.attrib["Target"]
for rel in relationships_root.findall("pr:Relationship", PACKAGE_REL_NS)
}
for sheet in workbook_root.findall("a:sheets/a:sheet", {**MAIN_NS, **REL_NS}):
if sheet.attrib.get("name") != sheet_name:
continue
relationship_id = sheet.attrib.get(f"{{{REL_NS['r']}}}id")
target = relationship_map[relationship_id]
target = target.lstrip("/")
if not target.startswith("xl/"):
target = f"xl/{target}"
return target
raise KeyError(f"Sheet '{sheet_name}' not found in workbook.")
def _load_cell_map(sheet_root, shared_strings):
cell_map = {}
for row in sheet_root.findall(".//a:sheetData/a:row", MAIN_NS):
for cell in row.findall("a:c", MAIN_NS):
reference = cell.attrib.get("r", "")
row_index, col_index = _split_cell_reference(reference)
cell_map[(row_index, col_index)] = _cell_value(cell, shared_strings)
return cell_map
def _cell_value(cell, shared_strings):
cell_type = cell.attrib.get("t")
if cell_type == "inlineStr":
text_nodes = cell.findall(".//a:t", MAIN_NS)
return "".join(node.text or "" for node in text_nodes)
value_node = cell.find("a:v", MAIN_NS)
if value_node is None or value_node.text is None:
return ""
raw_value = value_node.text
if cell_type == "s":
return shared_strings[int(raw_value)]
return raw_value
def _split_cell_reference(reference):
match = re.match(r"([A-Z]+)(\d+)", reference)
if not match:
raise ValueError(f"Unsupported cell reference: {reference}")
column_letters, row_number = match.groups()
return int(row_number), _column_letters_to_number(column_letters)
def _column_letters_to_number(column_letters):
result = 0
for character in column_letters:
result = result * 26 + (ord(character) - ord("A") + 1)
return result
def _max_used_extent(cell_map, start_row, start_col):
used_rows = [row for row, col in cell_map if row >= start_row and col >= start_col and cell_map[(row, col)] != ""]
used_cols = [col for row, col in cell_map if row >= start_row and col >= start_col and cell_map[(row, col)] != ""]
if not used_rows or not used_cols:
raise ValueError("No populated cells were found in the requested data area.")
return max(used_rows), max(used_cols)
def _trim_trailing_empty_rows_and_columns(matrix):
while matrix and all(value == "" for value in matrix[-1]):
matrix.pop()
if not matrix:
return matrix
while matrix and matrix[0]:
if any(row[-1] != "" for row in matrix):
break
for row in matrix:
row.pop()
return matrix
@@ -0,0 +1,72 @@
import re
from pathlib import Path
TM_ID_PATTERN = re.compile(r"([CS]\d{4})", re.IGNORECASE)
def get_batch_upload_entries(config):
bidding_config = config["exchange_bidding"]
upload_root = Path(bidding_config["upload_input_root"])
manifest_path = upload_root / "upload_manifest.json"
if manifest_path.exists():
return _load_manifest_entries(manifest_path)
return _load_entries_from_filenames(upload_root)
def _load_manifest_entries(manifest_path):
import json
with manifest_path.open("r", encoding="utf-8") as file:
payload = json.load(file)
entries = []
for item in payload.get("entries", []):
entries.append(
{
"tm_id": item["tm_id"],
"tac_id": item["tac_id"],
"upload_path": str((manifest_path.parent / item["file_name"]).resolve()),
"file_name": item["file_name"],
}
)
return entries
def _load_entries_from_filenames(upload_root):
entries = []
for path in sorted(upload_root.iterdir()):
if not path.is_file() or path.name.startswith("."):
continue
tac_id = path.stem.upper()
tm_id = _derive_tm_id_from_tac(tac_id)
entries.append(
{
"tm_id": tm_id,
"tac_id": tac_id,
"upload_path": str(path.resolve()),
"file_name": path.name,
}
)
if not entries:
raise FileNotFoundError(f"No upload files were found in {upload_root}")
return entries
def _derive_tm_id_from_tac(tac_id):
match = TM_ID_PATTERN.search(tac_id)
if not match:
raise ValueError(
f"Could not derive TM ID from TAC/file name '{tac_id}'. "
"Use a manifest file for custom mapping."
)
return match.group(1).upper()
@@ -0,0 +1,622 @@
import json
import re
import time
from datetime import datetime
from pathlib import Path
from selenium.common.exceptions import NoAlertPresentException, TimeoutException, WebDriverException
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
TM_ID_PATTERN = re.compile(r"([CS]\d{4})", re.IGNORECASE)
SCREEN_FOLDER_NAMES = {
"single_bid": "Single",
"block_bid": "Block",
"single_and_block_bid": "Single",
}
BUYSELL_SIDE = "BUYSELL"
BUYSELL_COMPONENT_SIDES = ("Buy", "Sell")
def get_folder_upload_entries(config):
product = config["product"]
bidding_config = config["exchange_bidding"]
upload_root = Path(
bidding_config.get("folder_upload_root")
or bidding_config.get("upload_input_root", "")
)
base_paths = _resolve_folder_upload_base_paths(upload_root, product, bidding_config)
entries = []
for base_path, side_name in base_paths:
for folder_path in sorted(path for path in base_path.iterdir() if path.is_dir()):
if folder_path.name.startswith("."):
continue
entries.append(
{
"folder_path": str(folder_path.resolve()),
"folder_name": folder_path.name,
"side": side_name,
"tm_id": _derive_tm_id_from_folder_name(folder_path.name),
}
)
if not entries:
searched_paths = [str(base_path) for base_path, _side_name in base_paths]
raise FileNotFoundError(f"No TM folders were found in {searched_paths}")
return entries
def upload_folder(driver, folder_path):
folder = Path(folder_path)
if not folder.exists() or not folder.is_dir():
raise FileNotFoundError(f"Upload folder not found: {folder}")
wait = WebDriverWait(driver, 30)
folder_input = wait.until(
lambda _driver: _folder_file_input(_driver) or False,
"No folder upload input was found. Expected an input[type='file'][webkitdirectory]."
)
driver.execute_script(
"""
arguments[0].style.display = 'block';
arguments[0].style.visibility = 'visible';
arguments[0].style.opacity = 1;
arguments[0].style.height = '1px';
arguments[0].style.width = '1px';
arguments[0].removeAttribute('disabled');
arguments[0].value = '';
""",
folder_input
)
folder_input.send_keys(str(folder.resolve()))
_accept_upload_confirmation_if_present(driver)
wait.until(lambda _driver: _folder_file_rows(_driver) or False)
def submit_all_and_wait_for_progress(
driver,
timeout_seconds=300,
poll_seconds=2,
keep_alive_interval_seconds=30,
progress_log_interval_seconds=60,
progress_log_label="folder_upload",
continue_when_last_file_complete=False,
):
wait = WebDriverWait(driver, 30)
submit_all_button = wait.until(
EC.element_to_be_clickable(
(
By.XPATH,
"//table[@id='tblFileList']//button[normalize-space()='Submit All']"
"|//button[normalize-space()='Submit All' or contains(normalize-space(),'Submit All')]"
)
)
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", submit_all_button)
_click(driver, submit_all_button)
submit_all_clicked_at = datetime.now()
submit_all_monotonic = time.monotonic()
progress_rows = _wait_for_all_progress_100(
driver,
timeout_seconds,
poll_seconds,
keep_alive_interval_seconds,
progress_log_interval_seconds,
progress_log_label,
continue_when_last_file_complete,
)
all_progress_100_at = datetime.now()
incomplete_rows = [
row for row in progress_rows
if row.get("progress", 0) < 100
]
return {
"submit_all_clicked_at": submit_all_clicked_at.isoformat(timespec="seconds"),
"all_progress_100_at": all_progress_100_at.isoformat(timespec="seconds"),
"completion_condition": _completion_condition(progress_rows),
"duration_seconds": round(time.monotonic() - submit_all_monotonic, 3),
"file_count": len(progress_rows),
"incomplete_count": len(incomplete_rows),
"incomplete_files": incomplete_rows,
"files": progress_rows,
}
def fetch_folder_upload_statuses(driver, timeout_seconds=30):
results = []
rows = _folder_file_rows(driver)
for index, row in enumerate(rows, start=1):
file_name = _row_cell_text(row, 1)
remarks = _row_cell_text(row, 2)
progress_text = _row_cell_text(row, 4)
progress = _row_progress_value(row, progress_text)
if progress < 100:
results.append(
{
"index": index,
"file_name": file_name,
"status": "upload_incomplete",
"raw_status": remarks,
"remarks": remarks,
"progress": progress,
"progress_text": progress_text,
"fetch_attempted": False,
}
)
print(
f"Skipping fetch for incomplete upload row: "
f"file={file_name}, progress={progress}%, remarks={remarks}"
)
continue
previous_status = _current_status_text(driver)
fetch_button = _row_fetch_button(row)
if fetch_button is None:
results.append(
{
"index": index,
"file_name": file_name,
"status": "missing_fetch_button",
"raw_status": "",
"remarks": remarks,
"progress": progress,
"progress_text": progress_text,
"fetch_attempted": False,
}
)
continue
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", fetch_button)
_click(driver, fetch_button)
try:
status_text = _wait_for_refreshed_status(driver, previous_status, timeout_seconds)
except TimeoutException:
status_text = _current_status_text(driver)
results.append(
{
"index": index,
"file_name": file_name,
"status": "fetch_timeout",
"raw_status": status_text,
"previous_status": previous_status,
"remarks": remarks,
"progress": progress,
"progress_text": progress_text,
"fetch_attempted": True,
}
)
print(
f"Fetch status timeout for uploaded row: "
f"file={file_name}, progress={progress}%, "
f"previous_status={previous_status}, current_status={status_text}"
)
continue
lowered_status = status_text.lower()
if "reject" in lowered_status:
status = "rejected"
elif "open" in lowered_status or "accept" in lowered_status or "submitted" in lowered_status:
status = "open"
else:
status = "unknown"
results.append(
{
"index": index,
"file_name": file_name,
"status": status,
"raw_status": status_text,
"remarks": remarks,
"progress": progress,
"progress_text": progress_text,
"fetch_attempted": True,
}
)
return results
def save_folder_upload_log(test_context, market, entry, timing, fetch_results):
folder = Path(test_context.get("log_folder", test_context["folder"]))
folder.mkdir(parents=True, exist_ok=True)
side_name = entry.get("side", "")
log_name_parts = ["folder_upload", market]
if side_name:
log_name_parts.append(side_name)
log_name_parts.append(entry["tm_id"])
path = folder / f"{'_'.join(log_name_parts)}.json"
rejected_files = [item for item in fetch_results if item["status"] == "rejected"]
unresolved_files = [
item for item in fetch_results
if item["status"] in {"upload_incomplete", "fetch_timeout", "missing_fetch_button", "unknown"}
]
payload = {
"market": market,
"side": side_name,
"tm_id": entry["tm_id"],
"folder_name": entry["folder_name"],
"folder_path": entry["folder_path"],
"timing": timing,
"fetch_results": fetch_results,
"rejected_files": rejected_files,
"unresolved_files": unresolved_files,
}
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
return str(path)
def _resolve_folder_upload_base_paths(upload_root, product, bidding_config):
if not upload_root:
raise ValueError("Set exchange_bidding.folder_upload_root or upload_input_root in config/runtime.json.")
base_path = upload_root.resolve()
if not base_path.exists():
raise FileNotFoundError(f"Folder upload root does not exist: {base_path}")
market_path = base_path / product["trading_market"]
if not market_path.exists():
available_markets = [
path.name for path in base_path.iterdir()
if path.is_dir() and not path.name.startswith(".")
]
raise FileNotFoundError(
f"Folder upload market path does not exist: {market_path}. "
f"Available market folders under {base_path}: {available_markets}. "
"Set active_selection.trading_market to an available market, "
"or create folders for every market when using ALL."
)
base_path = market_path
screen_folder = SCREEN_FOLDER_NAMES.get(product.get("bidding_screen", "single_bid"))
if screen_folder:
screen_path = base_path / screen_folder
if not screen_path.exists():
raise FileNotFoundError(f"Folder upload screen path does not exist: {screen_path}")
base_path = screen_path
sides = bidding_config.get("folder_upload_sides") or [bidding_config.get("folder_upload_side", "Buy")]
if isinstance(sides, str):
sides = [sides]
expanded_sides = _expand_folder_upload_sides(base_path, sides)
existing_side_paths = [
(base_path / side, side)
for side in expanded_sides
if (base_path / side).exists()
]
if existing_side_paths:
return existing_side_paths
raise FileNotFoundError(
f"Folder upload side path does not exist under {base_path}. "
f"Configured folder_upload_sides: {sides}. Expanded sides: {expanded_sides}"
)
def _expand_folder_upload_sides(base_path, sides):
expanded_sides = []
for side in sides:
side_name = str(side).strip()
if not side_name:
continue
if side_name.upper() != BUYSELL_SIDE:
expanded_sides.append(side_name)
continue
if (base_path / BUYSELL_SIDE).exists():
expanded_sides.append(BUYSELL_SIDE)
continue
expanded_sides.extend(BUYSELL_COMPONENT_SIDES)
return list(dict.fromkeys(expanded_sides))
def _derive_tm_id_from_folder_name(folder_name):
match = TM_ID_PATTERN.search(folder_name)
if not match:
raise ValueError(f"Could not derive TM ID from folder name '{folder_name}'.")
return match.group(1).upper()
def _folder_file_input(driver):
inputs = driver.find_elements(
By.XPATH,
"//input[@type='file' and (@webkitdirectory or @id='fileInput' or @name='fileInput')]"
)
if inputs:
return inputs[0]
return None
def _accept_upload_confirmation_if_present(driver):
try:
alert = driver.switch_to.alert
alert.accept()
except NoAlertPresentException:
pass
def _folder_file_rows(driver):
rows = driver.find_elements(By.XPATH, "//table[@id='tblFileList']//tbody/tr[td]")
return [row for row in rows if row.is_displayed()]
def _wait_for_all_progress_100(
driver,
timeout_seconds,
poll_seconds,
keep_alive_interval_seconds,
progress_log_interval_seconds,
progress_log_label,
continue_when_last_file_complete,
):
end_time = time.monotonic() + timeout_seconds
next_keep_alive = time.monotonic() + keep_alive_interval_seconds
next_progress_log = time.monotonic() + progress_log_interval_seconds
last_rows = []
while time.monotonic() < end_time:
try:
rows = _read_progress_rows(driver)
except WebDriverException as exc:
raise RuntimeError(
"Browser session disconnected while waiting for folder upload progress. "
"This usually means Chrome was closed, crashed, or Windows/Chrome "
"suspended the renderer during the long upload. Re-run with the browser "
"windows kept open and the PC sleep/lock settings disabled for the upload window."
) from exc
if rows:
last_rows = rows
if all(row["progress"] >= 100 for row in rows):
_print_progress_summary(progress_log_label, rows)
return rows
if continue_when_last_file_complete and _last_file_is_complete(rows):
_print_progress_summary(progress_log_label, rows)
_print_incomplete_rows(progress_log_label, rows)
return rows
if keep_alive_interval_seconds and time.monotonic() >= next_keep_alive:
_keep_session_active(driver)
next_keep_alive = time.monotonic() + keep_alive_interval_seconds
if progress_log_interval_seconds and time.monotonic() >= next_progress_log:
_print_progress_summary(progress_log_label, last_rows)
next_progress_log = time.monotonic() + progress_log_interval_seconds
time.sleep(poll_seconds)
raise TimeoutException(
"Folder upload progress did not reach 100% for all files. "
f"Last progress rows: {last_rows}"
)
def _print_progress_summary(label, rows):
if not rows:
print(f"Folder upload progress [{label}]: waiting for progress rows...")
return
completed = sum(1 for row in rows if row["progress"] >= 100)
total = len(rows)
max_progress = max(row["progress"] for row in rows)
min_progress = min(row["progress"] for row in rows)
print(
f"Folder upload progress [{label}]: "
f"{completed}/{total} files at 100%, "
f"min={min_progress}%, max={max_progress}%"
)
def _print_incomplete_rows(label, rows):
incomplete_rows = [
row for row in rows
if row.get("progress", 0) < 100
]
if not incomplete_rows:
return
print(
f"Folder upload progress [{label}]: continuing because last file is 100%. "
"Incomplete/error rows: "
+ "; ".join(
f"{row.get('file_name', '')} progress={row.get('progress', 0)}% "
f"remarks={row.get('remarks', '')}"
for row in incomplete_rows
)
)
def _last_file_is_complete(rows):
return bool(rows) and rows[-1].get("progress", 0) >= 100
def _completion_condition(rows):
if all(row.get("progress", 0) >= 100 for row in rows):
return "all_files_100"
if _last_file_is_complete(rows):
return "last_file_100_with_incomplete_rows"
return "incomplete"
def _read_progress_rows(driver):
rows = []
for row in _folder_file_rows(driver):
file_name = _row_cell_text(row, 1)
remarks = _row_cell_text(row, 2)
progress_text = _row_cell_text(row, 4)
progress = _row_progress_value(row, progress_text)
rows.append(
{
"file_name": file_name,
"remarks": remarks,
"progress": progress,
"progress_text": progress_text,
}
)
return rows
def _keep_session_active(driver):
driver.execute_script(
"""
const scrollY = window.scrollY || document.documentElement.scrollTop || 0;
window.scrollBy(0, 1);
window.scrollTo(0, scrollY);
document.dispatchEvent(new MouseEvent('mousemove', {
bubbles: true,
clientX: 1,
clientY: 1
}));
document.dispatchEvent(new KeyboardEvent('keydown', {
bubbles: true,
key: 'Shift'
}));
"""
)
def _row_progress_value(row, fallback_text):
progress_bars = row.find_elements(By.XPATH, ".//*[contains(@class,'progress-bar')]")
for progress_bar in progress_bars:
aria_value = progress_bar.get_attribute("aria-valuenow")
progress = _parse_progress(aria_value)
if progress > 0:
return progress
style = progress_bar.get_attribute("style") or ""
progress = _parse_progress(style)
if progress > 0:
return progress
text = progress_bar.text.strip()
progress = _parse_progress(text)
if progress > 0:
return progress
return _parse_progress(fallback_text)
def _row_cell_text(row, zero_based_index):
cells = row.find_elements(By.XPATH, "./td")
if len(cells) <= zero_based_index:
return ""
return cells[zero_based_index].text.strip()
def _parse_progress(progress_text):
progress_text = str(progress_text or "")
match = re.search(r"(\d+)\s*%", progress_text)
if match:
return int(match.group(1))
match = re.search(r"\b(\d{1,3})\b", progress_text)
if match:
return int(match.group(1))
if "100" in progress_text:
return 100
return 0
def _row_fetch_button(row):
buttons = row.find_elements(By.XPATH, ".//*[self::button or self::a][normalize-space()='Fetch']")
for button in buttons:
if button.is_displayed() and button.is_enabled():
return button
return None
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 _wait_for_refreshed_status(driver, previous_status, timeout_seconds):
wait = WebDriverWait(driver, timeout_seconds)
return wait.until(
lambda _driver: _refreshed_status_text(_driver, previous_status) or False
)
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 ""
def _click(driver, element):
try:
element.click()
except Exception:
driver.execute_script("arguments[0].click();", element)
@@ -0,0 +1,97 @@
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
def click_fetch_button(driver):
wait = WebDriverWait(driver, 20)
button = wait.until(
EC.element_to_be_clickable(
(
By.XPATH,
"//*[self::button or self::a or self::input]"
"[normalize-space()='Fetch' or contains(normalize-space(),'Fetch') or @value='Fetch']"
"|//*[contains(normalize-space(),'Fetch')]/ancestor::*[self::button or self::a][1]"
)
)
)
try:
button.click()
except Exception:
driver.execute_script("arguments[0].click();", button)
def paste_matrix_into_bidding_grid(driver, tsv_text, expected_first_value):
matrix = [row.split("\t") for row in tsv_text.splitlines()]
wait = WebDriverWait(driver, 30)
wait.until(
lambda _driver: _populate_handsontable(_driver, matrix) is not False
)
wait.until(
lambda _driver: _top_left_grid_value(_driver).strip() == str(expected_first_value).strip()
)
def click_submit_button(driver):
wait = WebDriverWait(driver, 20)
button = wait.until(
EC.element_to_be_clickable(
(By.XPATH, "//button[normalize-space()='Submit']")
)
)
try:
button.click()
except Exception:
driver.execute_script("arguments[0].click();", button)
def _top_left_grid_value(driver):
cell = driver.find_element(
By.XPATH,
"(//table[contains(@class,'htCore')]//tbody/tr[1]/td[@role='gridcell'])[1]"
)
return cell.text
def _populate_handsontable(driver, matrix):
return driver.execute_script(
"""
const matrix = arguments[0];
const instances = (window.Handsontable && window.Handsontable.instances) || [];
for (const instance of instances) {
if (!instance || !instance.rootElement) {
continue;
}
const root = instance.rootElement;
if (!root.isConnected || root.offsetParent === null) {
continue;
}
const hasGridTable = !!root.querySelector("table.htCore tbody td[role='gridcell']");
const isInsideSelect2 = !!root.closest(".select2-drop, .select2-container, .select2-results");
const isInsidePopup = !!root.closest(".modal, .popover, .tooltip");
if (!hasGridTable || isInsideSelect2 || isInsidePopup) {
continue;
}
try {
instance.selectCell(0, 0);
instance.populateFromArray(0, 0, matrix, null, null, "paste");
return instance.getDataAtCell(0, 0);
} catch (error) {
// Try the next visible instance.
}
}
return false;
""",
matrix,
)
@@ -0,0 +1,74 @@
from pathlib import Path
def get_enabled_bidding_entries(config):
bidding_config = config["exchange_bidding"]
input_root = Path(bidding_config["input_root"])
json_input_root = Path(bidding_config.get("json_input_root", ""))
upload_input_root = Path(bidding_config.get("upload_input_root", ""))
excel_layout = bidding_config.get("excel_layout", {})
entries = bidding_config.get("entries", [])
resolved_entries = []
for entry in entries:
if not entry.get("enabled", True):
continue
resolved_entries.append(
_resolve_entry_paths(
**entry,
input_root=input_root,
json_input_root=json_input_root,
upload_input_root=upload_input_root,
excel_layout=excel_layout
)
)
return resolved_entries
def get_active_bidding_entry(config):
bidding_config = config["exchange_bidding"]
active_entry_key = bidding_config.get("active_entry_key")
for entry in get_enabled_bidding_entries(config):
if entry["key"] == active_entry_key:
return entry
raise KeyError(
f"Active bidding entry '{active_entry_key}' was not found among enabled entries."
)
def _resolve_entry_paths(**kwargs):
entry = dict(kwargs)
input_root = entry.pop("input_root")
json_input_root = entry.pop("json_input_root")
upload_input_root = entry.pop("upload_input_root")
excel_layout = entry.pop("excel_layout")
excel_path = (input_root / entry["excel_file"]).resolve()
json_path = None
if entry.get("json_file"):
json_path = (json_input_root / entry["json_file"]).resolve()
upload_path = None
upload_file_name = entry.get("upload_file") or entry.get("excel_file")
if upload_file_name:
preferred_upload_path = (upload_input_root / upload_file_name).resolve()
upload_path = preferred_upload_path if preferred_upload_path.exists() else excel_path
return {
**entry,
"excel_path": str(excel_path),
"json_path": str(json_path) if json_path else None,
"upload_path": str(upload_path) if upload_path else None,
"excel_layout": {
**excel_layout,
**entry.get("excel_layout", {})
}
}
@@ -0,0 +1,80 @@
import json
from pathlib import Path
def load_single_bid_json_matrix(json_path):
data_file = Path(json_path)
if not data_file.exists():
raise FileNotFoundError(f"JSON input file not found: {data_file}")
with data_file.open("r", encoding="utf-8") as file:
payload = json.load(file)
if "matrix" in payload:
matrix = [[_to_text(value) for value in row] for row in payload["matrix"]]
elif "slot_range" in payload:
matrix = _build_range_matrix(payload)
else:
price_row = [_to_text(value) for value in payload["price_row"]]
slot_rows = [
[_to_text(value) for value in slot["orders"]]
for slot in payload.get("slots", [])
]
matrix = [price_row, *slot_rows]
if not matrix or not matrix[0]:
raise ValueError(f"No bidding data found in {data_file}.")
tsv = "\n".join("\t".join(row) for row in matrix)
return {
"matrix": matrix,
"tsv": tsv,
"row_count": len(matrix),
"column_count": len(matrix[0]),
"first_value": matrix[0][0],
}
def _build_range_matrix(payload):
price_row = [_to_text(value) for value in payload["price_row"]]
slot_range = payload["slot_range"]
from_slot = int(slot_range["from_slot"])
to_slot = int(slot_range["to_slot"])
orders = payload.get("orders", [])
if from_slot < 1 or to_slot < from_slot:
raise ValueError(f"Invalid slot_range: from_slot={from_slot}, to_slot={to_slot}")
expected_row_count = to_slot - from_slot + 1
if len(orders) != expected_row_count:
raise ValueError(
"Order row count does not match slot range: "
f"expected {expected_row_count}, found {len(orders)}"
)
column_count = len(price_row)
blank_row = [""] * column_count
slot_rows = []
for _ in range(from_slot - 1):
slot_rows.append(blank_row.copy())
for order_row in orders:
normalized_row = [_to_text(value) for value in order_row]
if len(normalized_row) != column_count:
raise ValueError(
"Each order row must have the same number of values as price_row: "
f"expected {column_count}, found {len(normalized_row)}"
)
slot_rows.append(normalized_row)
return [price_row, *slot_rows]
def _to_text(value):
if value is None:
return ""
return str(value)
@@ -0,0 +1,231 @@
import time
from urllib.parse import urlparse
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 (
click_select2_option_by_text,
open_bootstrap_nav_submenu,
select_select2_value_by_container_id,
wait_for_select2_option_by_text,
)
TRADING_MARKET_LABELS = {
"GDAM": "GDAM Order Entry",
"DAM": "DAM Order Entry",
"HPDAM": "HPDAM Order Entry",
}
TRADING_MARKET_HREFS = {
"GDAM": "../IDAS/GDAMOrderEntry.aspx",
"DAM": "../IDAS/DAMOrderEntry.aspx",
"HPDAM": "../IDAS/HPDAMOrderEntry.aspx",
}
BIDDING_SCREEN_LABELS = {
"single_bid": "Single Bid",
"block_bid": "Block Bid",
}
BLOCK_BULK_UPLOAD_HREF = "../FrontOffice/BulkOrderUploadIDAS.aspx"
def open_idas_market_page(driver, trading_market):
open_bootstrap_nav_submenu(
driver,
"Trading",
"IDAS",
TRADING_MARKET_HREFS[trading_market]
)
wait = WebDriverWait(driver, 20)
wait.until(
EC.any_of(
EC.url_contains("OrderEntry.aspx"),
EC.visibility_of_element_located((By.XPATH, "//a[normalize-space()='Single Bid']")),
EC.visibility_of_element_located((By.ID, "s2id_ddlproduct"))
)
)
def open_idas_block_bulk_upload_page(driver):
open_bootstrap_nav_submenu(
driver,
"Trading",
"IDAS",
"BulkOrderUploadIDAS.aspx"
)
wait = WebDriverWait(driver, 20)
wait.until(
EC.any_of(
EC.url_contains("BulkOrderUploadIDAS.aspx"),
EC.visibility_of_element_located((By.XPATH, "//*[contains(normalize-space(),'IDAS Bulk Order Upload')]")),
EC.visibility_of_element_located((By.ID, "validateOrder"))
)
)
def open_idas_market_page_direct(driver, trading_market):
current_url = urlparse(driver.current_url)
base_url = f"{current_url.scheme}://{current_url.netloc}"
order_entry_path = TRADING_MARKET_HREFS[trading_market].replace("../", "/")
driver.get(f"{base_url}{order_entry_path}")
wait = WebDriverWait(driver, 20)
wait.until(
EC.any_of(
EC.url_contains("OrderEntry.aspx"),
EC.visibility_of_element_located((By.XPATH, "//a[normalize-space()='Single Bid']")),
EC.visibility_of_element_located((By.ID, "s2id_ddlproduct"))
)
)
def open_bidding_screen(driver, bidding_screen):
if bidding_screen == "single_bid":
return
tab_label = BIDDING_SCREEN_LABELS[bidding_screen]
wait = WebDriverWait(driver, 20)
active_or_visible_tab = wait.until(
EC.visibility_of_element_located(
(
By.XPATH,
f"//*[self::a or self::button or self::span][normalize-space()='{tab_label}']"
)
)
)
tab_classes = (active_or_visible_tab.get_attribute("class") or "").lower()
if "active" in tab_classes:
return
tab = wait.until(
EC.element_to_be_clickable(
(
By.XPATH,
f"//*[self::a or self::button or self::span][normalize-space()='{tab_label}']"
)
)
)
try:
tab.click()
except Exception:
driver.execute_script("arguments[0].click();", tab)
def select_bidding_filters(
driver,
product_code,
tm_id,
tac_id,
product_to_tm_delay_seconds=0,
tm_to_tac_delay_seconds=0,
tm_timeout_seconds=30,
tm_attempts=3,
tac_timeout_seconds=15,
tac_attempts=2,
bidding_screen_ready_wait_seconds=10
):
if bidding_screen_ready_wait_seconds > 0:
time.sleep(bidding_screen_ready_wait_seconds)
wait_for_bidding_product_configuration(driver)
select_select2_value_by_container_id(driver, "s2id_ddlproduct", product_code, exact=True)
if product_to_tm_delay_seconds > 0:
time.sleep(product_to_tm_delay_seconds)
wait_for_select2_option_by_text(
driver,
"s2id_ddlTMID",
tm_id,
timeout_seconds=tm_timeout_seconds,
attempts=tm_attempts
)
click_select2_option_by_text(driver, "s2id_ddlTMID", tm_id)
if tm_to_tac_delay_seconds > 0:
time.sleep(tm_to_tac_delay_seconds)
wait_for_select2_option_by_text(
driver,
"s2id_ddlTACID",
tac_id,
timeout_seconds=tac_timeout_seconds,
attempts=tac_attempts
)
click_select2_option_by_text(driver, "s2id_ddlTACID", tac_id)
def select_bidding_filters_for_trading_member(
driver,
product_code,
tac_id,
tac_timeout_seconds=15,
tac_attempts=2,
bidding_screen_ready_wait_seconds=10
):
if bidding_screen_ready_wait_seconds > 0:
time.sleep(bidding_screen_ready_wait_seconds)
wait_for_bidding_product_configuration(driver)
select_select2_value_by_container_id(driver, "s2id_ddlproduct", product_code, exact=True)
wait_for_select2_option_by_text(
driver,
"s2id_ddlTACID",
tac_id,
timeout_seconds=tac_timeout_seconds,
attempts=tac_attempts
)
click_select2_option_by_text(driver, "s2id_ddlTACID", tac_id)
def select_bidding_product_and_tm(
driver,
product_code,
tm_id,
product_to_tm_delay_seconds=0,
tm_timeout_seconds=30,
tm_attempts=3,
bidding_screen_ready_wait_seconds=10
):
if bidding_screen_ready_wait_seconds > 0:
time.sleep(bidding_screen_ready_wait_seconds)
wait_for_bidding_product_configuration(driver)
select_select2_value_by_container_id(driver, "s2id_ddlproduct", product_code, exact=True)
if product_to_tm_delay_seconds > 0:
time.sleep(product_to_tm_delay_seconds)
wait_for_select2_option_by_text(
driver,
"s2id_ddlTMID",
tm_id,
timeout_seconds=tm_timeout_seconds,
attempts=tm_attempts
)
click_select2_option_by_text(driver, "s2id_ddlTMID", tm_id)
def wait_for_bidding_product_configuration(driver, timeout_seconds=60):
wait = WebDriverWait(driver, timeout_seconds)
wait.until(
lambda _driver: not any(
element.is_displayed()
for element in _driver.find_elements(
By.XPATH,
"//*[contains(translate(normalize-space(), "
"'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), "
"'begin product configuration')]"
)
)
)
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,191 @@
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 ""
@@ -0,0 +1 @@
from tests.regression.auth import login_to_trading_application
@@ -0,0 +1 @@
@@ -0,0 +1,345 @@
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"
@@ -0,0 +1,207 @@
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."
)
@@ -0,0 +1,215 @@
import pytest
from selenium.webdriver.support.ui import WebDriverWait
from tests.regression.auth import login_to_trading_application
from tests.regression.config_reader import load_regression_config, should_run_flow
from tests.regression.exchange.idas.matching.actions import (
click_generate_flow_request,
configure_browser_downloads,
configure_matching_request,
export_review_result_for_source,
open_idas_matching_page,
poll_provisional_match_status,
prepare_matching_download_folder,
start_provisional_match,
wait_for_flow_request_success,
)
from tests.regression.exchange.idas.matching.bid_transfer_actions import (
SUPPORTED_MARKETS,
check_market_transfer_status,
close_market_transfer_status,
open_idas_bid_transfer_page,
select_bid_transfer_product,
transfer_market_orders,
)
from tests.step_actions import StepActions
def get_exchange_matching_context(driver, test_context):
config = load_regression_config()
assert config["selected_role"] == "exchange", (
"IDAS matching is available only for exchange login. "
"Set execution.login_role to 'exchange' in config/runtime.json."
)
if not should_run_flow("matching"):
pytest.skip(
f"Skipping exchange matching flow because flow_scope is '{config['selected_flow']}'."
)
wait = WebDriverWait(driver, 20)
step = StepActions(driver, wait, test_context)
return config, step
def open_app_and_login(driver, step, config):
driver.get(config["portal_urls"]["pratyay"])
step.capture("open_pratyay_application")
login_to_trading_application(
driver,
config["login"]["username"],
config["login"]["password"],
config["timing"],
login_role=config["selected_role"]
)
step.capture("pratyay_login_submitted")
def _step_slug(source_type):
return source_type.strip().lower().replace(" ", "_")
def _configured_transfer_markets(matching_config):
markets = matching_config.get("bid_transfer_markets") or list(SUPPORTED_MARKETS)
normalized_markets = [market.strip().upper() for market in markets]
if "ALL" in normalized_markets:
return list(SUPPORTED_MARKETS)
return list(dict.fromkeys(normalized_markets))
@pytest.mark.regression
@pytest.mark.exchange
@pytest.mark.matching
def test_idas_exchange_matching_flow(driver, test_context):
config, step = get_exchange_matching_context(driver, test_context)
matching_config = config["exchange_matching"]
if not matching_config.get("portals", {}).get("pratyay", {}).get("enabled", True):
pytest.skip("Pratyay matching preparation is disabled in config/runtime.json.")
open_app_and_login(driver, step, config)
product_config = config["product"]
transfer_markets = _configured_transfer_markets(matching_config)
transfer_timeout = matching_config.get("bid_transfer_timeout_seconds", 30)
accepted_transfer_statuses = {
status.strip().upper()
for status in matching_config.get(
"accepted_bid_transfer_statuses",
["ACCEPTED", "PARTIALLY_ACCEPTED", "NO_BIDS"],
)
}
open_idas_bid_transfer_page(driver)
step.capture("open_idas_bid_transfer")
select_bid_transfer_product(driver, product_config["product_code"])
step.capture("select_bid_transfer_product")
for index, market in enumerate(transfer_markets, start=1):
transfer_result = transfer_market_orders(
driver,
market,
timeout_seconds=transfer_timeout,
)
step.capture(f"bid_transfer_{index}_{market.lower()}")
status_result = check_market_transfer_status(
driver,
market,
timeout_seconds=transfer_timeout,
)
popup_status = status_result["normalized"]
step.capture(
f"bid_transfer_status_{index}_{market.lower()}_{popup_status.lower()}"
)
if popup_status == "REJECTED":
step.test_context["steps"][-1]["status"] = "failed"
raise AssertionError(
f"{market} MCO bid-transfer status is REJECTED. "
"Matching has been stopped before processing the next market."
)
if popup_status not in accepted_transfer_statuses:
step.test_context["steps"][-1]["status"] = "failed"
raise AssertionError(
f"{market} returned unsupported MCO bid-transfer status "
f"'{status_result['text']}'. Matching has been stopped. "
f"Accepted statuses: {sorted(accepted_transfer_statuses)}"
)
close_market_transfer_status(driver, status_result)
print(
f"IDAS bid transfer completed for market={market}, "
f"transfer_message={transfer_result['message']}, "
f"popup_status={status_result['text']}"
)
open_idas_matching_page(driver)
step.capture("open_idas_initiate_matching")
if matching_config.get("initiation_mode", "provisional_match") != "provisional_match":
raise ValueError(
"Only 'provisional_match' initiation_mode is supported right now."
)
download_folder = prepare_matching_download_folder(
test_context,
folder_name=matching_config.get("download_folder_name", "matching_files")
)
configure_browser_downloads(driver, download_folder)
source_types = matching_config.get(
"source_types",
["Green", "Conventional", "HPDAM"]
)
exported_files = []
for index, source_type in enumerate(source_types, start=1):
source_slug = _step_slug(source_type)
configure_matching_request(
driver,
product_config["product_code"],
matching_config["matching_type"],
source_type=source_type
)
step.capture(f"select_matching_filters_{index}_{source_slug}")
start_provisional_match(driver)
step.capture(f"start_provisional_match_{index}_{source_slug}")
final_status = poll_provisional_match_status(
driver,
interval_seconds=matching_config.get("check_status_interval_seconds", 2),
timeout_seconds=matching_config.get("check_status_timeout_seconds", 120)
)
step.capture(f"provisional_matching_completed_{index}_{source_slug}")
click_generate_flow_request(driver)
success_message = wait_for_flow_request_success(
driver,
timeout_seconds=matching_config.get("flow_request_success_timeout_seconds", 30)
)
step.capture(f"generate_flow_request_{index}_{source_slug}")
exported_file = export_review_result_for_source(
driver,
source_type,
download_folder,
timeout_seconds=matching_config.get("download_timeout_seconds", 60)
)
exported_files.append(exported_file)
step.capture(f"review_result_exported_{index}_{source_slug}")
print(
f"Matching review export completed for source_type={source_type}, "
f"status={final_status}, "
f"flow_request_message={success_message}, "
f"exported_file={exported_file}"
)
print(
"IDAS exchange matching completed for: "
f"matching_product_code={product_config['product_code']}, "
f"matching_type={matching_config['matching_type']}, "
f"source_types={source_types}, "
f"matching_files_folder={download_folder}, "
f"exported_files={exported_files}"
)
@@ -0,0 +1,42 @@
import pytest
from selenium.webdriver.support.ui import WebDriverWait
from tests.regression.auth import login_to_mco_application
from tests.regression.config_reader import load_regression_config, should_run_flow
from tests.step_actions import StepActions
@pytest.mark.regression
@pytest.mark.exchange
@pytest.mark.matching
@pytest.mark.mco
def test_idas_mco_login(driver, test_context):
config = load_regression_config()
matching_config = config["exchange_matching"]
mco_config = matching_config.get("portals", {}).get("mco", {})
if not should_run_flow("matching"):
pytest.skip(
f"Skipping MCO login because flow_scope is '{config['selected_flow']}'."
)
if not mco_config.get("enabled", False):
pytest.skip("MCO matching portal is disabled in config/runtime.json.")
step = StepActions(driver, WebDriverWait(driver, 20), test_context)
driver.get(config["portal_urls"]["mco"])
step.capture("open_mco_application")
login_to_mco_application(
driver,
config["mco_login"].get("username"),
config["mco_login"].get("password"),
config["timing"],
)
step.capture("mco_dashboard_loaded")
print(
"IDAS MCO login completed for "
f"environment={config['portal_urls']['environment']}, "
f"dashboard_url={driver.current_url}"
)
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1 @@
from tests.regression.auth import TradingLoginLocator
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1 @@
+264
View File
@@ -0,0 +1,264 @@
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.common.exceptions import StaleElementReferenceException, TimeoutException
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
def _menu_text_xpath(tag_name, label):
return (
f"//{tag_name}[contains(normalize-space(),\"{label}\")]"
f"|//{tag_name}[contains(.,\"{label}\")]"
)
def open_bootstrap_nav_submenu(driver, top_menu_label, submenu_label, final_href_part):
wait = WebDriverWait(driver, 20)
top_menu = wait.until(
EC.element_to_be_clickable(
(
By.XPATH,
"//ul[contains(@class,'nav') and contains(@class,'navbar-nav')]"
f"//a[contains(@class,'dropdown-toggle') and contains(normalize-space(),'{top_menu_label}')]"
)
)
)
try:
top_menu.click()
except Exception:
driver.execute_script("arguments[0].click();", top_menu)
top_menu_container = wait.until(
lambda _driver: _find_open_top_menu_container(_driver, top_menu_label)
)
submenu = wait.until(
lambda _driver: _find_clickable_submenu(_driver, top_menu_container, submenu_label)
)
try:
submenu.click()
except Exception:
driver.execute_script("arguments[0].click();", submenu)
final_option = wait.until(
lambda _driver: _find_clickable_final_option(_driver, top_menu_container, final_href_part)
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", final_option)
try:
final_option.click()
except Exception:
driver.execute_script("arguments[0].click();", final_option)
def _find_open_top_menu_container(driver, top_menu_label):
containers = driver.find_elements(
By.XPATH,
"//ul[contains(@class,'nav') and contains(@class,'navbar-nav')]/li"
)
normalized_label = top_menu_label.strip().lower()
for container in containers:
classes = (container.get_attribute("class") or "").lower()
text = container.text.strip().lower()
if normalized_label in text and "open" in classes:
return container
return False
def _find_clickable_submenu(driver, top_menu_container, submenu_label):
normalized_label = submenu_label.strip().lower()
submenu_links = top_menu_container.find_elements(
By.XPATH,
".//ul[contains(@class,'dropdown-menu')]//a"
)
for link in submenu_links:
text = link.text.strip().lower()
if normalized_label in text and link.is_displayed() and link.is_enabled():
return link
return False
def _find_clickable_final_option(driver, top_menu_container, final_href_part):
final_links = top_menu_container.find_elements(
By.XPATH,
f".//ul[contains(@class,'dropdown-menu')]//a[contains(@href,'{final_href_part}')]"
)
for link in final_links:
if link.is_displayed() and link.is_enabled():
return link
return False
def select_select2_value_by_container_id(driver, container_id, value, exact=False):
wait = WebDriverWait(driver, 20)
container = wait.until(
EC.element_to_be_clickable((By.ID, container_id))
)
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", container)
try:
container.click()
except Exception:
driver.execute_script("arguments[0].click();", container)
try:
search_input = wait.until(
EC.visibility_of_element_located(
(By.XPATH, "//div[contains(@class,'select2-drop-active')]//input[contains(@class,'select2-input')]")
)
)
search_input.click()
search_input.send_keys(Keys.CONTROL, "a")
search_input.send_keys(value)
except TimeoutException:
pass
try:
option = wait.until(
lambda _driver: _find_select2_option(_driver, value, exact=exact)
)
except TimeoutException as exc:
available_options = _get_visible_select2_option_texts(driver)
raise TimeoutException(
f"Option '{value}' did not load for {container_id}. "
f"Visible options: {available_options}"
) from exc
try:
option.click()
except Exception:
driver.execute_script("arguments[0].click();", option)
def _find_select2_option(driver, value, exact=False):
options = driver.find_elements(
By.XPATH,
"//div[contains(@class,'select2-drop-active')]//div[contains(@class,'select2-result-label')]"
)
normalized_value = value.strip().lower()
visible_options = []
for option in options:
try:
if option.is_displayed():
visible_options.append(option)
except StaleElementReferenceException:
return False
for option in visible_options:
text = _safe_element_text(option)
if text is None:
return False
if text == normalized_value:
return option
for option in visible_options:
text = _safe_element_text(option)
if text is None:
return False
option_code = text.split("(", 1)[0].strip()
if option_code == normalized_value:
return option
if exact:
return False
for option in visible_options:
text = _safe_element_text(option)
if text is None:
return False
if text.startswith(normalized_value):
return option
for option in visible_options:
text = _safe_element_text(option)
if text is None:
return False
if normalized_value in text:
return option
return False
def _safe_element_text(element):
try:
return element.text.strip().lower()
except StaleElementReferenceException:
return None
def click_select2_option_by_text(driver, container_id, value):
wait = WebDriverWait(driver, 20)
container = wait.until(EC.element_to_be_clickable((By.ID, container_id)))
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", container)
try:
container.click()
except Exception:
driver.execute_script("arguments[0].click();", container)
option = wait.until(lambda _driver: _find_select2_option(_driver, value))
driver.execute_script("arguments[0].scrollIntoView({block: 'nearest'});", option)
try:
option.click()
except Exception:
driver.execute_script("arguments[0].click();", option)
def wait_for_select2_option_by_text(driver, container_id, value, timeout_seconds=30, attempts=3):
wait = WebDriverWait(driver, timeout_seconds)
last_error = None
for _ in range(attempts):
container = wait.until(EC.element_to_be_clickable((By.ID, container_id)))
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", container)
try:
container.click()
except Exception:
driver.execute_script("arguments[0].click();", container)
try:
option = wait.until(lambda _driver: _find_select2_option(_driver, value))
try:
container.click()
except Exception:
driver.execute_script("arguments[0].click();", container)
return option
except TimeoutException as exc:
last_error = exc
try:
container.click()
except Exception:
driver.execute_script("arguments[0].click();", container)
if last_error:
available_options = _get_visible_select2_option_texts(driver)
raise TimeoutException(
f"Option '{value}' did not load for {container_id}. "
f"Visible options: {available_options}"
) from last_error
raise TimeoutException(f"Option '{value}' did not load for {container_id}")
def _get_visible_select2_option_texts(driver):
options = driver.find_elements(
By.XPATH,
"//div[contains(@class,'select2-drop-active')]//div[contains(@class,'select2-result-label')]"
)
return [option.text.strip() for option in options if option.is_displayed()]
+20
View File
@@ -0,0 +1,20 @@
import os
from datetime import datetime
def log_step(driver, request, step_name):
screenshots_dir = "reports/screenshots"
os.makedirs(screenshots_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{step_name}_{timestamp}.png"
filepath = os.path.join(screenshots_dir, filename)
driver.save_screenshot(filepath)
if hasattr(request.node, "extra"):
import pytest_html
request.node.extra.append(pytest_html.extras.image(filepath))
print(f"STEP: {step_name}")
+1
View File
@@ -0,0 +1 @@
+1
View File
@@ -0,0 +1 @@
+72
View File
@@ -0,0 +1,72 @@
from datetime import datetime, timedelta
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
import pytest
from tests.step_actions import StepActions
from tests.helpers import login, select_ui_dropdown
from tests.config_reader import is_combined_setup_enabled, is_setup_session_enabled, load_config
from tests.locators import Locator
pytestmark = pytest.mark.skipif(
is_combined_setup_enabled() or not is_setup_session_enabled(),
reason="Standalone session setup is disabled or runs inside combined setup",
)
def get_session_window(config):
session_config = config.get("setup", {}).get("session", {})
start_offset_minutes = session_config.get("start_offset_minutes", 7)
end_time = session_config.get("end_time", "20:30")
start_datetime = datetime.now() + timedelta(minutes=start_offset_minutes)
today = datetime.now().strftime("%d/%m/%Y")
return start_datetime.strftime("%d/%m/%Y %H:%M"), f"{today} {end_time}"
def create_session(driver, step, config):
step.click((By.XPATH, Locator.tradingmaintenance), "trading_maintenance")
step.click((By.XPATH, Locator.tradingsession), "trading_session")
step.click((By.XPATH, Locator.addsessionbtn), "add_session")
select_ui_dropdown(driver, "meInstanceId", config["product"]["me_instance"])
step.capture("select_product")
start_date_time, end_date_time = get_session_window(config)
step.type((By.XPATH, Locator.inputstartdate), start_date_time, "start_date")
step.type((By.XPATH, Locator.inputenddate), end_date_time, "end_date")
step.click((By.XPATH, Locator.sessionsavebtn), "save_session")
@pytest.mark.session
def test_session(driver, test_context):
config = load_config()
wait = WebDriverWait(driver, 20)
step = StepActions(driver, wait, test_context)
driver.get(config["url"])
step.capture("open_application")
login(
driver,
config["login"]["username"],
config["login"]["password"]
)
step.capture("login")
create_session(driver, step, config)
+46
View File
@@ -0,0 +1,46 @@
import pytest
from selenium.webdriver.support.ui import WebDriverWait
from tests.config_reader import is_combined_setup_enabled, load_config
from tests.helpers import login, logout, save_session_backup
from tests.setup.sessions.test_session import create_session
from tests.setup.tokens.test_trading_token import create_and_activate_token, get_product_codes
from tests.step_actions import StepActions
pytestmark = pytest.mark.skipif(
not is_combined_setup_enabled(),
reason="Combined setup requires setup.run_session and setup.run_token to both be true",
)
@pytest.mark.session
@pytest.mark.trading_token
def test_session_and_token_setup(driver, test_context):
config = load_config()
token = config.get("trading_token_idas")
assert token, "Add trading_token_idas in config.json before running token setup."
wait = WebDriverWait(driver, 20)
step = StepActions(driver, wait, test_context)
driver.get(config["url"])
step.capture("open_application")
login(
driver,
config["login"]["username"],
config["login"]["password"]
)
step.capture("login")
save_session_backup(driver, test_context, "login_session_backup")
create_session(driver, step, config)
for product_code in get_product_codes(token):
product_token = dict(token)
product_token["product_code"] = product_code
create_and_activate_token(driver, step, product_token)
logout(driver)
+1
View File
@@ -0,0 +1 @@
+249
View File
@@ -0,0 +1,249 @@
from datetime import datetime, timedelta
import pytest
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from tests.config_reader import is_combined_setup_enabled, is_setup_token_enabled, load_config, parse_product_codes
from tests.helpers import (
click_visible_menu_item,
click_first_grid_button,
click_grid_button_for_row_text,
handle_transient_popups,
login,
logout,
open_trading_maintenance_page,
pause_after_step,
save_session_backup,
select_bootstrap_dropdown_by_id,
select_labeled_ui_dropdown,
type_first_visible_input,
)
from tests.locators import Locator
from tests.step_actions import StepActions
pytestmark = pytest.mark.skipif(
is_combined_setup_enabled() or not is_setup_token_enabled(),
reason="Standalone token setup is disabled or runs inside combined setup",
)
def get_token_context(driver, test_context):
config = load_config()
wait = WebDriverWait(driver, 20)
step = StepActions(driver, wait, test_context)
token = config.get("trading_token_idas")
assert token, "Add trading_token_idas in config.json before running this test."
return config, step, token
def open_app_and_login(driver, step, config):
driver.get(config["url"])
step.capture("open_application")
login(
driver,
config["login"]["username"],
config["login"]["password"]
)
step.capture("login")
def get_delivery_date_and_expiry(token):
delivery_offset_days = token.get("delivery_offset_days", 1)
expiry_time = token.get("expiry_time", "20:30")
delivery_date = (datetime.now() + timedelta(days=delivery_offset_days)).strftime("%d/%m/%Y")
token_expiry = f"{datetime.now().strftime('%d/%m/%Y')} {expiry_time}"
return delivery_date, token_expiry
def get_time_session_times(token, row_text, default_start_time, default_end_time):
time_session_config = token.get("time_sessions", {}).get(row_text, {})
start_time = time_session_config.get("start_time")
if start_time is None:
start_offset_minutes = time_session_config.get("start_offset_minutes")
if start_offset_minutes is not None:
start_time = (datetime.now() + timedelta(minutes=start_offset_minutes)).strftime("%H:%M")
return (
start_time or default_start_time,
time_session_config.get("end_time", default_end_time)
)
def get_product_codes(token):
product_codes = parse_product_codes(token.get("product_codes"))
if product_codes:
return product_codes
product_code = token.get("product_code")
assert product_code, "Add product_code or product_codes in config.json trading_token_idas."
return parse_product_codes(product_code)
def edit_time_session_row(driver, step, row_text, start_time, end_time):
today = datetime.now().strftime("%d/%m/%Y")
click_grid_button_for_row_text(driver, row_text, "Details")
step.capture(f"open_{row_text.lower()}_details")
step.click((By.XPATH, Locator.edittimesessionbtn), f"edit_{row_text.lower()}_time_session")
step.type((By.XPATH, Locator.timesessionstartdate), today, f"enter_{row_text.lower()}_start_date")
step.type((By.XPATH, Locator.timesessionenddate), today, f"enter_{row_text.lower()}_end_date")
type_first_visible_input(driver, Locator.timesessionstarttime, start_time)
step.capture(f"enter_{row_text.lower()}_start_time")
type_first_visible_input(driver, Locator.timesessionendtime, end_time)
step.capture(f"enter_{row_text.lower()}_end_time")
step.click((By.XPATH, Locator.savetimesessionbtn), f"save_{row_text.lower()}_time_session")
handle_transient_popups(driver)
def complete_time_session_edits(driver, step, token, next_day):
open_trading_maintenance_page(driver, Locator.tokenandtimesessionview)
step.capture("open_token_and_time_session_view")
select_bootstrap_dropdown_by_id(driver, Locator.viewproducttype, token["product_type"])
step.capture("select_view_product_type")
select_bootstrap_dropdown_by_id(driver, Locator.viewproductcode, token["product_code"])
step.capture("select_view_product_code")
step.type((By.XPATH, Locator.viewdeliveryfromdate), next_day, "enter_view_delivery_from_date")
step.type((By.XPATH, Locator.viewdeliverytodate), next_day, "enter_view_delivery_to_date")
step.click((By.XPATH, Locator.viewsearchbtn), "search_token_and_time_session")
click_grid_button_for_row_text(driver, "12-Mar-2026", "Time Session Details")
step.capture("open_time_session_details")
step.capture("before_edit_order_normal")
order_nor_start_time, order_nor_end_time = get_time_session_times(
token,
"ORDER_NOR",
(datetime.now() + timedelta(minutes=10)).strftime("%H:%M"),
"18:30"
)
order_exch_start_time, order_exch_end_time = get_time_session_times(token, "ORDER_EXCH", "18:30", "19:00")
trade_start_time, trade_end_time = get_time_session_times(token, "TRADE", "19:00", "20:00")
edit_time_session_row(
driver,
step,
"ORDER_NOR",
order_nor_start_time,
order_nor_end_time
)
edit_time_session_row(driver, step, "ORDER_EXCH", order_exch_start_time, order_exch_end_time)
edit_time_session_row(driver, step, "TRADE", trade_start_time, trade_end_time)
click_first_grid_button(driver, "Token Details")
step.capture("open_token_details")
step.click((By.XPATH, Locator.savetokenandactivatebtn), "save_and_activate_token")
handle_transient_popups(driver)
pause_after_step()
def complete_time_session_edits_from_saved_token(driver, step, token):
step.click((By.XPATH, Locator.tokentimesessionbtn), "open_time_session")
click_first_grid_button(driver, "Time Session Details")
step.capture("open_time_session_details")
step.capture("before_edit_order_normal")
order_nor_start_time, order_nor_end_time = get_time_session_times(
token,
"ORDER_NOR",
(datetime.now() + timedelta(minutes=10)).strftime("%H:%M"),
"18:30"
)
order_exch_start_time, order_exch_end_time = get_time_session_times(token, "ORDER_EXCH", "18:30", "19:00")
trade_start_time, trade_end_time = get_time_session_times(token, "TRADE", "19:00", "20:00")
edit_time_session_row(
driver,
step,
"ORDER_NOR",
order_nor_start_time,
order_nor_end_time
)
edit_time_session_row(driver, step, "ORDER_EXCH", order_exch_start_time, order_exch_end_time)
edit_time_session_row(driver, step, "TRADE", trade_start_time, trade_end_time)
click_first_grid_button(driver, "Token Details")
step.capture("open_token_details")
step.click((By.XPATH, Locator.savetokenandactivatebtn), "save_and_activate_token")
handle_transient_popups(driver)
pause_after_step()
def create_and_activate_token(driver, step, token):
product_code = token["product_code"]
token_menu_items = driver.find_elements(By.XPATH, Locator.tokengeneration)
if any(item.is_displayed() for item in token_menu_items):
click_visible_menu_item(driver, Locator.tokengeneration)
else:
open_trading_maintenance_page(driver, Locator.tokengeneration)
handle_transient_popups(driver)
step.capture(f"open_token_generation_{product_code.lower()}")
select_bootstrap_dropdown_by_id(driver, Locator.tokenmarketcode, token["market_code"])
step.capture(f"select_market_code_{product_code.lower()}")
select_labeled_ui_dropdown(driver, "PRODUCT CODE", product_code)
step.capture(f"select_product_code_{product_code.lower()}")
step.click((By.XPATH, Locator.addnewtokenbtn), f"click_add_new_token_{product_code.lower()}")
next_day, token_expiry = get_delivery_date_and_expiry(token)
step.type((By.XPATH, Locator.tokendeliverystartdate), next_day, f"enter_delivery_start_date_{product_code.lower()}")
step.type((By.XPATH, Locator.tokendeliveryenddate), next_day, f"enter_delivery_end_date_{product_code.lower()}")
step.type((By.XPATH, Locator.tokenexpirydate), token_expiry, f"enter_token_expiry_date_{product_code.lower()}")
step.click((By.XPATH, Locator.savetokenbtn), f"save_token_{product_code.lower()}")
handle_transient_popups(driver)
complete_time_session_edits_from_saved_token(driver, step, token)
@pytest.mark.trading_token
def test_trading_token_creation_idas(driver, test_context):
config, step, token = get_token_context(driver, test_context)
open_app_and_login(driver, step, config)
save_session_backup(driver, test_context, "login_session_backup")
for product_code in get_product_codes(token):
product_token = dict(token)
product_token["product_code"] = product_code
create_and_activate_token(driver, step, product_token)
logout(driver)
# @pytest.mark.trading_token
# def test_trading_token_time_session_edit_idas(driver, test_context):
#
# config, step, token = get_token_context(driver, test_context)
# open_app_and_login(driver, step, config)
#
# next_day, _ = get_delivery_date_and_expiry(token)
#
# complete_time_session_edits(driver, step, token, next_day)
+68
View File
@@ -0,0 +1,68 @@
import os
from time import sleep
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import ElementClickInterceptedException
from .config_reader import get_step_delay_seconds
STEP_DELAY_SECONDS = get_step_delay_seconds()
class StepActions:
def __init__(self, driver, wait, test_context):
self.driver = driver
self.wait = wait
self.folder = test_context["folder"]
self.test_context = test_context
self.step = 1
self.test_context.setdefault("steps", [])
def capture(self, name):
filename = f"{self.step:02d}_{name}.png"
path = os.path.join(self.folder, filename)
self.driver.save_screenshot(path)
self.test_context["steps"].append(
{
"index": self.step,
"name": name,
"path": path,
"status": "passed",
}
)
print(f"STEP {self.step}: {name}")
self.step += 1
sleep(STEP_DELAY_SECONDS)
def click(self, locator, name):
element = self.wait.until(
EC.element_to_be_clickable(locator)
)
self.driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", element)
try:
element.click()
except ElementClickInterceptedException:
self.driver.execute_script("arguments[0].click();", element)
self.capture(name)
def type(self, locator, value, name):
element = self.wait.until(
EC.visibility_of_element_located(locator)
)
element.clear()
element.send_keys(value)
self.capture(name)
+95
View File
@@ -0,0 +1,95 @@
import random
import time
import pytest
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from .config_reader import load_config
from .helpers import (
clear_pending_console_input,
click_page_body,
enter_pressed,
login,
open_trading_maintenance_page,
select_bootstrap_dropdown_by_id,
select_ui_dropdown,
toggle_bootstrap_dropdown_by_id,
toggle_labeled_ui_dropdown,
)
from .locators import Locator
from .step_actions import StepActions
def _bounce_token_generation(driver, step, config):
token = config["trading_token_idas"]
interactions = random.randint(2, 4)
open_trading_maintenance_page(driver, Locator.tokengeneration)
step.capture("open_token_generation_fake")
for index in range(interactions):
if random.choice([True, False]):
toggle_bootstrap_dropdown_by_id(driver, Locator.tokenmarketcode)
step.capture(f"toggle_market_dropdown_{index + 1}")
click_page_body(driver)
else:
select_bootstrap_dropdown_by_id(driver, Locator.tokenmarketcode, token["market_code"])
step.capture(f"select_market_code_{index + 1}")
toggle_labeled_ui_dropdown(driver, "PRODUCT CODE")
step.capture(f"toggle_product_dropdown_{index + 1}")
click_page_body(driver)
time.sleep(random.uniform(0.5, 1.5))
def _bounce_trading_session(driver, step, config):
open_trading_maintenance_page(driver, Locator.tradingsession)
step.capture("open_trading_session_fake")
step.click((By.XPATH, Locator.addsessionbtn), "open_add_session_fake")
interactions = random.randint(2, 4)
for index in range(interactions):
if random.choice([True, False]):
select_ui_dropdown(driver, "meInstanceId", config["product"]["me_instance"])
step.capture(f"select_me_instance_{index + 1}")
else:
click_page_body(driver)
step.capture(f"idle_on_session_form_{index + 1}")
time.sleep(random.uniform(0.5, 1.5))
@pytest.mark.fake_activity
def test_random_fake_activity_no_save(driver, test_context):
config = load_config()
wait = WebDriverWait(driver, 20)
step = StepActions(driver, wait, test_context)
driver.get(config["url"])
step.capture("open_application")
login(
driver,
config["login"]["username"],
config["login"]["password"]
)
step.capture("login")
actions = [_bounce_token_generation, _bounce_trading_session]
random.shuffle(actions)
clear_pending_console_input()
print("Fake activity started. Press Enter in the terminal any time to stop.")
while True:
random.choice(actions)(driver, step, config)
if enter_pressed():
break
step.capture("manual_stop_point")
+7
View File
@@ -0,0 +1,7 @@
import pytest
pytest.skip(
"Session test moved to tests/setup/sessions/test_session.py",
allow_module_level=True,
)
+7
View File
@@ -0,0 +1,7 @@
import pytest
pytest.skip(
"Trading token test moved to tests/setup/tokens/test_trading_token.py",
allow_module_level=True,
)