273 lines
9.3 KiB
Python
273 lines
9.3 KiB
Python
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", {})
|
|
}
|