498 lines
21 KiB
Python
498 lines
21 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
import pytest
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
|
|
from tests.config_reader import get_setup_configs, 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
|
|
|
|
|
|
def get_token_context(driver, test_context, config):
|
|
wait = WebDriverWait(driver, 20)
|
|
step = StepActions(driver, wait, test_context)
|
|
token = config["token"]
|
|
|
|
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_intraday_delivery_window(token, now=None):
|
|
"""Return an Intraday delivery window that cannot start in the past."""
|
|
now = datetime.now() if now is None else now
|
|
start = now + timedelta(minutes=token.get("delivery_start_offset_minutes", 5))
|
|
start_text = start.strftime("%d/%m/%Y %H:%M")
|
|
end_text = f"{start.strftime('%d/%m/%Y')} {token.get('delivery_to_time', '23:59')}"
|
|
return start_text, end_text
|
|
|
|
|
|
def get_time_session_times(token, row_text, default_start_time, default_end_time):
|
|
|
|
time_session_config = token.get("time_sessions", {}).get(row_text, {})
|
|
time_mode = time_session_config.get("time_mode")
|
|
start_time = time_session_config.get("start_time")
|
|
|
|
if time_mode == "offset":
|
|
start_offset_minutes = time_session_config.get("start_offset_minutes")
|
|
if start_offset_minutes is None:
|
|
raise ValueError(f"{row_text} offset mode requires start_offset_minutes")
|
|
start_time = (datetime.now() + timedelta(minutes=start_offset_minutes)).strftime("%H:%M")
|
|
elif time_mode == "manual" and not start_time:
|
|
raise ValueError(f"{row_text} manual mode requires start_time")
|
|
|
|
return (
|
|
start_time or default_start_time,
|
|
time_session_config.get("end_time", default_end_time)
|
|
)
|
|
|
|
|
|
def get_upc_order_normal_start_time(timings, now=None):
|
|
"""Resolve UPC order-normal start time from manual or offset mode."""
|
|
mode = timings.get("time_mode", "manual").lower()
|
|
if mode == "offset":
|
|
offset = timings.get("order_normal_start_offset_minutes")
|
|
if offset is None:
|
|
raise ValueError(
|
|
"UPC time_mode 'offset' requires "
|
|
"order_normal_start_offset_minutes"
|
|
)
|
|
now = datetime.now() if now is None else now
|
|
return (now + timedelta(minutes=offset)).strftime("%H:%M")
|
|
if mode != "manual":
|
|
raise ValueError("UPC time_mode must be 'manual' or 'offset'")
|
|
|
|
start_time = timings.get("order_normal_start_time")
|
|
if not start_time:
|
|
raise ValueError(
|
|
"UPC time_mode 'manual' requires order_normal_start_time"
|
|
)
|
|
return start_time
|
|
|
|
|
|
def get_product_codes(token):
|
|
|
|
product_codes = parse_product_codes(token.get("create_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 products.<product>.token."
|
|
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_NORMAL",
|
|
order_nor_start_time,
|
|
order_nor_end_time
|
|
)
|
|
edit_time_session_row(driver, step, "ORDER_EXCHANGE", 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")
|
|
WebDriverWait(driver, 20).until(EC.url_contains("timeSessionView"))
|
|
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_NORMAL",
|
|
order_nor_start_time,
|
|
order_nor_end_time
|
|
)
|
|
edit_time_session_row(driver, step, "ORDER_EXCHANGE", 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 open_rtm_new_token(driver, step, token):
|
|
"""Prepare the RTM generator through Add New Token."""
|
|
open_trading_maintenance_page(driver, Locator.tokengenerator)
|
|
handle_transient_popups(driver)
|
|
step.capture("open_token_generator_rtm")
|
|
|
|
for element_id, value, name in (
|
|
(Locator.generatormarketcode, token["market_code"], "market_code"),
|
|
(Locator.generatorproductcode, token["product_code"], "product_code"),
|
|
(Locator.generatorgenerationmode, token["generation_mode"], "generation_mode"),
|
|
):
|
|
select_bootstrap_dropdown_by_id(driver, element_id, value, exact=True)
|
|
step.capture(f"select_{name}_rtm")
|
|
|
|
# Product Type is populated by the application after selecting Product Code.
|
|
step.click((By.XPATH, Locator.addnewtokenbtn), "click_add_new_token_rtm")
|
|
|
|
|
|
def get_rtm_token_times(token, now=None):
|
|
now = datetime.now() if now is None else now
|
|
delivery_date = (now + timedelta(days=token.get("delivery_offset_days", 0))).strftime("%d/%m/%Y")
|
|
time_mode = token.get("time_mode", "offset").lower()
|
|
if time_mode == "manual":
|
|
session_start_time = token.get("session_start_time")
|
|
if not session_start_time:
|
|
raise ValueError("RTM time_mode 'manual' requires session_start_time")
|
|
elif time_mode == "offset":
|
|
session_start_time = (now + timedelta(minutes=token.get("session_start_offset_minutes", 5))).strftime("%H:%M")
|
|
else:
|
|
raise ValueError("RTM time_mode must be 'manual' or 'offset'")
|
|
return (
|
|
f"{delivery_date} {token.get('delivery_from_time', '00:00')}",
|
|
f"{delivery_date} {token.get('delivery_to_time', '23:59')}",
|
|
session_start_time,
|
|
token.get("session_end_time", "20:00"),
|
|
)
|
|
|
|
|
|
def create_rtm_token(driver, step, token):
|
|
open_rtm_new_token(driver, step, token)
|
|
values = get_rtm_token_times(token)
|
|
fields = (
|
|
(Locator.rtmdeliveryfromdate, "delivery_from_date"),
|
|
(Locator.rtmdeliverytodate, "delivery_to_date"),
|
|
(Locator.rtmsessionstarttime, "session_start_time"),
|
|
(Locator.rtmsessionendtime, "session_end_time"),
|
|
)
|
|
for (locator, name), value in zip(fields, values):
|
|
type_first_visible_input(driver, locator, value)
|
|
step.capture(f"enter_{name}_rtm")
|
|
step.click((By.XPATH, Locator.savetokenbtn), "save_token_rtm")
|
|
handle_transient_popups(driver)
|
|
|
|
|
|
def create_intraday_token(driver, step, token):
|
|
"""Create an Intraday token through the Token Generator page."""
|
|
open_trading_maintenance_page(driver, Locator.tokengenerator)
|
|
handle_transient_popups(driver)
|
|
step.capture("open_token_generator_intraday")
|
|
|
|
for element_id, value, name in (
|
|
(Locator.generatormarketcode, token["market_code"], "market_code"),
|
|
(Locator.generatorproductcode, token["product_code"], "product_code"),
|
|
(Locator.generatorgenerationmode, token["generation_mode"], "generation_mode"),
|
|
):
|
|
select_bootstrap_dropdown_by_id(driver, element_id, value, exact=True)
|
|
step.capture(f"select_{name}_intraday")
|
|
|
|
step.click((By.XPATH, Locator.addnewtokenbtn), "click_add_new_token_intraday")
|
|
|
|
delivery_start, delivery_end = get_intraday_delivery_window(token)
|
|
values = (
|
|
(Locator.intradaydeliveryfromdate, delivery_start, "delivery_from_date"),
|
|
(Locator.intradaydeliverytodate, delivery_end, "delivery_to_date"),
|
|
(Locator.intradaytradingstarttime, token.get("trading_start_time", "00:20"), "trading_start_time"),
|
|
(Locator.intradaydaysbeforedelivery, str(token.get("trading_start_days_before_delivery", 1)), "days_before_delivery"),
|
|
)
|
|
for locator, value, name in values:
|
|
type_first_visible_input(driver, locator, value)
|
|
step.capture(f"enter_{name}_intraday")
|
|
|
|
step.click((By.XPATH, Locator.savetokenbtn), "save_token_intraday")
|
|
WebDriverWait(driver, 30).until(
|
|
EC.visibility_of_element_located((By.XPATH, Locator.intradaysuccessmessage))
|
|
)
|
|
step.capture("intraday_token_generated_successfully")
|
|
handle_transient_popups(driver)
|
|
|
|
|
|
def create_dac_token(driver, step, token):
|
|
"""Create a DAC token through the Token Generator page."""
|
|
open_trading_maintenance_page(driver, Locator.tokengenerator)
|
|
handle_transient_popups(driver)
|
|
step.capture("open_token_generator_dac")
|
|
|
|
for element_id, value, name in (
|
|
(Locator.generatormarketcode, token["market_code"], "market_code"),
|
|
(Locator.generatorproductcode, token["product_code"], "product_code"),
|
|
(Locator.generatorgenerationmode, token["generation_mode"], "generation_mode"),
|
|
):
|
|
select_bootstrap_dropdown_by_id(driver, element_id, value, exact=True)
|
|
step.capture(f"select_{name}_dac")
|
|
|
|
step.click((By.XPATH, Locator.addnewtokenbtn), "click_add_new_token_dac")
|
|
|
|
trading_start, trading_end = get_intraday_delivery_window(token)
|
|
values = (
|
|
(Locator.dactradingfromdate, trading_start, "trading_from_date"),
|
|
(Locator.dactradingtodate, trading_end, "trading_to_date"),
|
|
(Locator.dacsessionstarttime, token.get("session_start_time", "13:00"), "session_start_time"),
|
|
(Locator.dacsessionendtime, token.get("session_end_time", "23:00"), "session_end_time"),
|
|
)
|
|
for locator, value, name in values:
|
|
type_first_visible_input(driver, locator, value)
|
|
step.capture(f"enter_{name}_dac")
|
|
|
|
step.click((By.XPATH, Locator.savetokenbtn), "save_token_dac")
|
|
WebDriverWait(driver, 30).until(
|
|
EC.visibility_of_element_located((By.XPATH, Locator.intradaysuccessmessage))
|
|
)
|
|
step.capture("dac_token_generated_successfully")
|
|
handle_transient_popups(driver)
|
|
|
|
|
|
def create_upc_token(driver, step, token):
|
|
open_trading_maintenance_page(driver, Locator.tokengenerator)
|
|
handle_transient_popups(driver)
|
|
step.capture("open_token_generator_upc")
|
|
fields = [(Locator.generatormarketcode, token["market_code"], "market_code"), (Locator.generatorproductcode, token["product_code"], "product_code")]
|
|
if driver.find_elements(By.ID, Locator.generatorgenerationmode):
|
|
fields.append((Locator.generatorgenerationmode, token["generation_mode"], "generation_mode"))
|
|
for element_id, value, name in fields:
|
|
select_bootstrap_dropdown_by_id(driver, element_id, value, exact=True)
|
|
step.capture(f"select_{name}_upc")
|
|
step.click((By.XPATH, Locator.addnewtokenbtn), "click_add_new_token_upc")
|
|
timings = token.get("upc_times", {})
|
|
for name, locator, value in (
|
|
("order_normal_start", Locator.upcorderstart, get_upc_order_normal_start_time(timings)),
|
|
("order_normal_end", Locator.upcorderend, timings.get("order_normal_end_time")),
|
|
("trade_start", Locator.upctradestart, timings.get("trade_session_start_time")),
|
|
("trade_end", Locator.upctradeend, timings.get("trade_session_end_time")),
|
|
):
|
|
if value:
|
|
type_first_visible_input(driver, locator, value)
|
|
step.capture(f"enter_{name}_upc")
|
|
step.click((By.XPATH, Locator.savetokenbtn), "save_token_upc")
|
|
handle_transient_popups(driver)
|
|
step.click((By.XPATH, Locator.tokengeneratorresetbtn), "reset_token_generator_upc")
|
|
|
|
|
|
def run_token_setup(driver, step, token):
|
|
if token.get("product_type", "").upper() == "DAC":
|
|
create_dac_token(driver, step, token)
|
|
return
|
|
if token.get("product_type", "").upper() == "INTRADAY":
|
|
create_intraday_token(driver, step, token)
|
|
return
|
|
if token.get("product_type", "").upper() == "UPC":
|
|
create_upc_token(driver, step, token)
|
|
return
|
|
if token.get("product_type", "").upper() == "RTM":
|
|
create_rtm_token(driver, step, token)
|
|
else:
|
|
create_and_activate_token(driver, step, token)
|
|
|
|
|
|
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()}")
|
|
|
|
generation_mode = token.get("generation_mode")
|
|
if generation_mode:
|
|
mode_elements = driver.find_elements(By.ID, Locator.generatorgenerationmode)
|
|
if mode_elements:
|
|
select_bootstrap_dropdown_by_id(driver, Locator.generatorgenerationmode, generation_mode, exact=True)
|
|
step.capture(f"select_generation_mode_{product_code.lower()}")
|
|
|
|
step.click((By.XPATH, Locator.addnewtokenbtn), f"click_add_new_token_{product_code.lower()}")
|
|
|
|
is_intraday = token.get("product_type", "").upper() == "INTRADAY"
|
|
if is_intraday:
|
|
delivery_start, delivery_end = get_intraday_delivery_window(token)
|
|
else:
|
|
next_day, token_expiry = get_delivery_date_and_expiry(token)
|
|
|
|
if token.get("product_type", "").upper() == "REC":
|
|
token_type_prefix = token.get("type_prefix", "REC")
|
|
token_type = f"{token_type_prefix}-{datetime.now().strftime('%d%m%Y')}"
|
|
step.type((By.XPATH, Locator.tokentype), token_type, f"enter_type_{product_code.lower()}")
|
|
|
|
if token.get("product_type", "").upper() == "UPC":
|
|
today = datetime.now().strftime("%d/%m/%Y")
|
|
if token.get("generation_mode") == "AnyDayUPC":
|
|
step.type((By.XPATH, Locator.tokentradedate), today, f"enter_trade_date_{product_code.lower()}")
|
|
timings = token.get("upc_times", {})
|
|
normal_start = get_upc_order_normal_start_time(timings)
|
|
values = {
|
|
Locator.upcordernormalstart: normal_start,
|
|
Locator.upcordernormalend: timings.get("order_normal_end_time"),
|
|
Locator.upctradesessionstart: timings.get("trade_session_start_time"),
|
|
Locator.upctradesessionend: timings.get("trade_session_end_time"),
|
|
}
|
|
for locator, value in values.items():
|
|
if value is None:
|
|
continue
|
|
type_first_visible_input(driver, locator, value)
|
|
|
|
if is_intraday:
|
|
step.type((By.XPATH, Locator.tokendeliverystartdate), delivery_start, f"enter_delivery_start_date_{product_code.lower()}")
|
|
step.type((By.XPATH, Locator.tokendeliveryenddate), delivery_end, f"enter_delivery_end_date_{product_code.lower()}")
|
|
elif (
|
|
token.get("generation_mode") not in ("WeeklyUPC", "MonthlyUPC")
|
|
and token.get("product_type", "").upper() != "REC"
|
|
):
|
|
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()}")
|
|
|
|
if not is_intraday and token.get("generation_mode") not in ("WeeklyUPC", "MonthlyUPC"):
|
|
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)
|
|
|
|
if token.get("product_type", "").upper() == "UPC":
|
|
step.click((By.XPATH, Locator.tokengeneratorresetbtn), f"reset_token_generator_{product_code.lower()}")
|
|
return
|
|
|
|
if not is_intraday:
|
|
complete_time_session_edits_from_saved_token(driver, step, token)
|
|
|
|
|
|
@pytest.mark.trading_token
|
|
@pytest.mark.parametrize("config", get_setup_configs("token"), ids=lambda c: c["product_name"])
|
|
def test_trading_token_creation(driver, test_context, config, request):
|
|
|
|
# The full setup command runs token-enabled products through the combined
|
|
# flow so session and token creation use one browser/login context.
|
|
invocation = " ".join(str(arg).replace("\\", "/") for arg in request.config.invocation_params.args)
|
|
if "tests/setup" in invocation and "tests/setup/tokens" not in invocation and "test_trading_token.py" not in invocation:
|
|
pytest.skip("Handled by the shared full setup flow")
|
|
|
|
config, step, token = get_token_context(driver, test_context, config)
|
|
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
|
|
run_token_setup(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, config)
|
|
# open_app_and_login(driver, step, config)
|
|
#
|
|
# next_day, _ = get_delivery_date_and_expiry(token)
|
|
#
|
|
# complete_time_session_edits(driver, step, token, next_day)
|