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..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)