diff --git a/.gitignore b/.gitignore index 81f7099..e5fd208 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,6 @@ reports/logs/* reports/html/* !reports/html/.gitkeep + +# CAPTCHA stress-test screenshots and CSV reports +reports/captcha_login_stress/ diff --git a/config.example.json b/config.example.json index 0d08d1d..06660ff 100644 --- a/config.example.json +++ b/config.example.json @@ -1,209 +1,249 @@ { - "url": "https://utilseastern.pxil.in/Pxil/", - "stagingurl": "https://utilseastern.pxil.in/Pxil/", - "uaturl": "https://utilsuat.pxil.in/Pxil/", - "login": { - "username": "Tejas", - "password": "Pxil@123" - }, - "login_captcha_wait_seconds": 10, - "login_captcha_max_retries": 10, - "login_post_submit_wait_seconds": 10, - "step_delay_seconds": 1, - "setup": { - "run_session": true, - "run_token": false, - "session": { - "start_offset_minutes": 5, - "end_time": "20:30" + "url": "https://utilseastern.pxil.in/Pxil/", + "stagingurl": "https://utilseastern.pxil.in/Pxil/", + "uaturl": "https://utilsuat.pxil.in/Pxil/", + "login": { + "username": "Tejas", + "password": "Pxil@123" + }, + "login_captcha_wait_seconds": 10, + "login_captcha_max_retries": 10, + "login_post_submit_wait_seconds": 10, + "step_delay_seconds": 1, + "setup": { + "run_session": true, + "run_token": false, + "session": { + "time_mode": "offset", + "start_offset_minutes": 5, + "start_time": "14:30", + "end_time": "20:30" + } + }, + "products": { + "IDAS": { + "enabled": true, + "run_session": true, + "run_token": false, + "me_instance": "55", + "session": {}, + "token": { + "market_code": "NATIONAL", + "product_type": "IDAS", + "product_code": "IDAS4", + "product_codes": "IDAS,IDASTEST01,IDASTEST02,IDASTEST03,IDASTEST04,IDASTEST05,IDASTEST06", + "available_codes": ",IDASTEST06,IDASTEST07,IDASTEST08,IDASTEST09,IDASTEST10,IDASTEST11,IDASTEST12,IDASTEST13,IDASTEST14,IDASTEST15,IDASTEST16,IDASTEST17,IDASTEST18,IDASTEST19,IDASTEST20,IDASTEST21,IDASTEST22,IDASTEST23,IDASTEST24,IDASTEST25,IDASTEST26,IDASTEST27,IDASTEST28,IDASTEST29,IDASTEST30", + "delivery_offset_days": 1, + "expiry_time": "20:30", + "time_sessions": { + "ORDER_NOR": { + "start_offset_minutes": 10, + "end_time": "19:30" + }, + "ORDER_EXCH": { + "start_time": "19:30", + "end_time": "19:45" + }, + "TRADE": { + "start_time": "19:45", + "end_time": "19:50" + } + } + } + }, + "RTM": { + "enabled": false, + "run_session": false, + "run_token": true, + "me_instance": "52", + "session": { + "time_mode": "offset", + "start_offset_minutes": 7, + "start_time": "14:30", + "end_time": "20:30" + }, + "token": { + "market_code": "NATIONAL", + "product_type": "RTM", + "product_codes": [ + "RTM" + ], + "generation_mode": "RTM", + "delivery_offset_days": 0, + "delivery_from_time": "00:00", + "delivery_to_time": "23:59", + "time_mode": "offset", + "session_start_offset_minutes": 5, + "session_start_time": "14:30", + "session_end_time": "20:00" + } + }, + "UPC": { + "enabled": true, + "run_session": true, + "run_token": true, + "session_variants": { + "DAILY": { + "enabled": true, + "run_session": false, + "run_token": true, + "me_instance": "58", + "token": { + "generation_mode": "AnyDayUPC", + "create_product_codes": [ + "DailyC_E" + ], + "available_product_codes": [ + "DailyC_D", + "DailyGR_D", + "DailyC_RTC", + "DailyC_E", + "DailyC_M" + ] + } + }, + "WEEKLY": { + "enabled": true, + "run_session": false, + "run_token": true, + "me_instance": "59", + "token": { + "generation_mode": "WeeklyUPC", + "create_product_codes": [ + "WeeklyC_E" + ], + "available_product_codes": [ + "WeeklyC_D", + "WeeklyC_N", + "WeeklyGR_D", + "WeeklyC_RTC" + ] + } + }, + "MONTHLY": { + "enabled": true, + "run_session": false, + "run_token": true, + "me_instance": "60", + "token": { + "generation_mode": "MonthlyUPC", + "create_product_codes": [ + "MonthlyC_N" + ], + "available_product_codes": [ + "MonthlyC_D", + "MonthlyGR_D", + "MonthlyC_RTC" + ] + } + } + }, + "session": {}, + "token": { + "market_code": "NATIONAL", + "product_type": "UPC", + "product_codes": [], + "delivery_offset_days": 1, + "expiry_time": "20:30", + "upc_times": { + "time_mode": "manual", + "order_normal_start_time": "13:00", + "order_normal_start_offset_minutes": 10, + "order_normal_end_time": "17:00", + "trade_session_start_time": "17:00", + "trade_session_end_time": "22:00" + } + } + }, + "RA": { + "enabled": false, + "run_session": true, + "run_token": false, + "me_instance": "64", + "session": {}, + "token": { + "market_code": "NATIONAL", + "product_type": "RA", + "product_codes": [], + "delivery_offset_days": 1, + "expiry_time": "20:30", + "time_sessions": {} + } + }, + "REC": { + "enabled": false, + "run_session": true, + "run_token": true, + "me_instance": "25", + "session": {}, + "token": { + "market_code": "NATIONAL", + "product_type": "REC", + "product_codes": [ + "REC" + ], + "delivery_offset_days": 1, + "expiry_time": "20:30", + "type_prefix": "REC", + "time_sessions": { + "ORDER_NOR": { + "time_mode": "offset", + "start_offset_minutes": 10, + "start_time": "14:30", + "end_time": "19:30" + }, + "ORDER_EXCH": { + "start_time": "19:30", + "end_time": "19:45" + }, + "TRADE": { + "start_time": "19:45", + "end_time": "19:50" + } + } + } + }, + "INTRADAY": { + "enabled": true, + "run_session": true, + "run_token": true, + "me_instance": "53", + "session": {}, + "token": { + "market_code": "NATIONAL", + "product_type": "INTRADAY", + "product_codes": [ + "I_DayStatic_C" + ], + "generation_mode": "Intraday24by7", + "delivery_start_offset_minutes": 5, + "trading_start_time": "00:20", + "trading_start_days_before_delivery": 1, + "duration_before_delivery_end_time": 150, + "delivery_to_time": "23:59" + } + }, + "DAC": { + "enabled": false, + "run_session": true, + "run_token": true, + "me_instance": "57", + "session": { + "time_mode": "offset", + "start_offset_minutes": 5, + "end_time": "23:00" + }, + "token": { + "market_code": "NATIONAL", + "product_type": "DAC", + "product_codes": ["DACStatic"], + "generation_mode": "DAC", + "delivery_start_offset_minutes": 5, + "delivery_to_time": "23:59", + "session_start_time": "13:00", + "session_end_time": "23:00", + "delivery_time_period": 15 + } + } } - }, - "products": { - "IDAS": { - "enabled": true, - "run_session": true, - "run_token": false, - "me_instance": "55", - "session": {}, - "token": { - "market_code": "NATIONAL", - "product_type": "IDAS", - "product_code": "IDAS4", - "product_codes": "IDAS,IDASTEST01,IDASTEST02,IDASTEST03,IDASTEST04,IDASTEST05,IDASTEST06", - "available_codes": ",IDASTEST06,IDASTEST07,IDASTEST08,IDASTEST09,IDASTEST10,IDASTEST11,IDASTEST12,IDASTEST13,IDASTEST14,IDASTEST15,IDASTEST16,IDASTEST17,IDASTEST18,IDASTEST19,IDASTEST20,IDASTEST21,IDASTEST22,IDASTEST23,IDASTEST24,IDASTEST25,IDASTEST26,IDASTEST27,IDASTEST28,IDASTEST29,IDASTEST30", - "delivery_offset_days": 1, - "expiry_time": "20:30", - "time_sessions": { - "ORDER_NOR": { - "start_offset_minutes": 10, - "end_time": "19:30" - }, - "ORDER_EXCH": { - "start_time": "19:30", - "end_time": "19:45" - }, - "TRADE": { - "start_time": "19:45", - "end_time": "19:50" - } - } - } - }, - "RTM": { - "enabled": false, - "run_session": false, - "run_token": true, - "me_instance": "52", - "session": { - "start_offset_minutes": 7, - "end_time": "20:30" - }, - "token": { - "market_code": "NATIONAL", - "product_type": "RTM", - "product_codes": [ - "RTM" - ], - "generation_mode": "RTM", - "delivery_offset_days": 0, - "delivery_from_time": "00:00", - "delivery_to_time": "23:59", - "session_start_offset_minutes": 5, - "session_end_time": "20:00" - } - }, - "UPC": { - "enabled": true, - "run_session": true, - "run_token": true, - "session_variants": { - "DAILY": { - "enabled": true, - "run_session": false, - "run_token": true, - "me_instance": "58", - "token": { - "generation_mode": "AnyDayUPC", - "create_product_codes": [ - "DailyC_E" - ], - "available_product_codes": [ - "DailyC_D", - "DailyGR_D", - "DailyC_RTC", - "DailyC_E", - "DailyC_M" - ] - } - }, - "WEEKLY": { - "enabled": true, - "run_session": false, - "run_token": true, - "me_instance": "59", - "token": { - "generation_mode": "WeeklyUPC", - "create_product_codes": [ - "WeeklyC_E" - ], - "available_product_codes": [ - "WeeklyC_D", - "WeeklyC_N", - "WeeklyGR_D", - "WeeklyC_RTC" - ] - } - }, - "MONTHLY": { - "enabled": true, - "run_session": false, - "run_token": true, - "me_instance": "60", - "token": { - "generation_mode": "MonthlyUPC", - "create_product_codes": [ - "MonthlyC_N" - ], - "available_product_codes": [ - "MonthlyC_D", - "MonthlyGR_D", - "MonthlyC_RTC" - ] - } - } - }, - "session": {}, - "token": { - "market_code": "NATIONAL", - "product_type": "UPC", - "product_codes": [], - "delivery_offset_days": 1, - "expiry_time": "20:30", - "upc_times": { - "order_normal_start_time": "13:00", - "order_normal_end_time": "17:00", - "trade_session_start_time": "17:00", - "trade_session_end_time": "22:00" - }, - "time_sessions": { - "ORDER_NOR": { - "start_offset_minutes": 10, - "end_time": "19:30" - }, - "ORDER_EXCH": { - "start_time": "19:30", - "end_time": "19:45" - }, - "TRADE": { - "start_time": "19:45", - "end_time": "19:50" - } - } - } - }, - "RA": { - "enabled": false, - "run_session": true, - "run_token": false, - "me_instance": "64", - "session": {}, - "token": { - "market_code": "NATIONAL", - "product_type": "RA", - "product_codes": [], - "delivery_offset_days": 1, - "expiry_time": "20:30", - "time_sessions": {} - } - }, - "REC": { - "enabled": false, - "run_session": true, - "run_token": true, - "me_instance": "25", - "session": {}, - "token": { - "market_code": "NATIONAL", - "product_type": "REC", - "product_codes": ["REC"], - "delivery_offset_days": 1, - "expiry_time": "20:30", - "type_prefix": "REC", - "time_sessions": { - "ORDER_NOR": { - "start_offset_minutes": 10, - "end_time": "19:30" - }, - "ORDER_EXCH": { - "start_time": "19:30", - "end_time": "19:45" - }, - "TRADE": { - "start_time": "19:45", - "end_time": "19:50" - } - } - } - } - } } diff --git a/pytest.ini b/pytest.ini index e984976..4c3f267 100644 --- a/pytest.ini +++ b/pytest.ini @@ -12,3 +12,9 @@ markers = reports: report download coverage order_book: order book coverage trade_book: trade book coverage + captcha: CAPTCHA login stress test + +filterwarnings = + ignore::DeprecationWarning:easyocr.* + ignore::DeprecationWarning:torch.* + ignore::UserWarning:torch.* diff --git a/requirements.txt b/requirements.txt index 3dde252..a58577b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,14 @@ # Core test runner -pytest==8.4.1 +pytest==9.1.1 pytest-html==4.2.0 # Browser automation -selenium==4.34.2 +selenium==4.50.0 + +# CAPTCHA OCR and image processing +easyocr==1.7.2 +opencv-python==5.0.0.93 +Pillow==12.3.0 +numpy==2.5.3 +torch==2.14.1 +torchvision==0.29.1 diff --git a/tests/browser.py b/tests/browser.py index f692ec3..01b50ae 100644 --- a/tests/browser.py +++ b/tests/browser.py @@ -1,4 +1,7 @@ +import os + from selenium import webdriver +from selenium.webdriver.chrome.service import Service def create_chrome_driver(): @@ -10,5 +13,11 @@ def create_chrome_driver(): options.add_argument("--disable-features=CalculateNativeWinOcclusion") options.add_argument("--disable-hang-monitor") options.add_argument("--disable-popup-blocking") + options.add_argument("--disable-background-networking") + options.add_argument("--disable-component-update") + options.add_argument("--disable-sync") + options.add_argument("--disable-default-apps") + options.add_experimental_option("excludeSwitches", ["enable-logging"]) - return webdriver.Chrome(options=options) + service = Service(log_output=os.devnull) + return webdriver.Chrome(service=service, options=options) diff --git a/tests/captcha_solver.py b/tests/captcha_solver.py new file mode 100644 index 0000000..6d81fbe --- /dev/null +++ b/tests/captcha_solver.py @@ -0,0 +1,353 @@ +""" +CAPTCHA Solver for PXIL Login using EasyOCR. +Extracts base64 from input's CSS background-image, runs EasyOCR OCR, fills the input. +""" +import re +import base64 +import time +from typing import Optional + +from PIL import Image +import cv2 +import numpy as np +from selenium.webdriver.common.by import By +from selenium.webdriver.remote.webdriver import WebDriver +from selenium.webdriver.remote.webelement import WebElement +from selenium.webdriver.support.ui import WebDriverWait +from selenium.webdriver.support import expected_conditions as EC +from selenium.common.exceptions import StaleElementReferenceException, TimeoutException + + +# Initialize EasyOCR once (module level) +# First run downloads models, then works offline +try: + import easyocr + _EASYOCR_READER = easyocr.Reader(['en'], gpu=False) +except ImportError: + _EASYOCR_READER = None + + +# Character confusion correction map for common OCR errors +CONFUSION_MAP = { + '0': 'O', 'O': '0', 'D': '0', + '1': 'I', 'I': '1', 'L': '1', 'l': '1', + '5': 'S', 'S': '5', + '8': 'B', 'B': '8', + '2': 'Z', 'Z': '2', + '6': 'G', 'G': '6', + '9': 'g', 'g': '9', + 'Q': 'O', + 'U': 'V', 'V': 'U', + 'n': 'h', 'h': 'n', + 'r': 'f', 'f': 'r', + 'c': 'e', 'e': 'c', + 'o': '0', '0': 'o', + 's': '5', '5': 's', + 'z': '2', '2': 'z', + 'i': '1', '1': 'i', + 'l': '1', '1': 'l', +} + + +def find_visible_captcha_input(driver: WebDriver) -> WebElement: + """Find the visible captcha input element (name='captcha').""" + wait = WebDriverWait(driver, 15) + inputs = wait.until(EC.presence_of_all_elements_located((By.NAME, "captcha"))) + for inp in inputs: + if inp.is_displayed(): + return inp + raise RuntimeError("No visible captcha input found") + + +def extract_captcha_base64(captcha_input: WebElement) -> bytes: + """Extract base64 image data from input's style attribute. Returns raw JPEG bytes.""" + style = captcha_input.get_attribute("style") or "" + match = re.search(r'background:\s*url\(["\']?data:image/[^;]+;base64,([^"\')]+)["\']?\)', style) + if not match: + raise RuntimeError(f"No base64 background-image found in style: {style[:200]}") + + b64_data = match.group(1) + if not b64_data: + raise RuntimeError("Base64 data is empty (captcha not loaded yet)") + + b64_data += "=" * ((4 - len(b64_data) % 4) % 4) # Fix padding + return base64.b64decode(b64_data) + + +def wait_for_captcha_load(driver: WebDriver, captcha_input: WebElement, timeout: int = 15) -> bool: + """Wait for captcha base64 to be populated with actual image data.""" + end_time = time.time() + timeout + while time.time() < end_time: + try: + style = captcha_input.get_attribute("style") or "" + match = re.search(r'background:\s*url\(["\']?data:image/[^;]+;base64,([^"\')]+)["\']?\)', style) + if match and match.group(1): + return True + except StaleElementReferenceException: + captcha_input = find_visible_captcha_input(driver) + time.sleep(0.3) + return False + + +def preprocess_captcha_image(image_bytes: bytes) -> Image.Image: + """Preprocess image for better OCR accuracy - gentle preprocessing.""" + nparr = np.frombuffer(image_bytes, np.uint8) + img = cv2.imdecode(nparr, cv2.IMREAD_COLOR) # Load as color (BGR) + + if img is None: + raise RuntimeError("Failed to decode image bytes") + + # Convert BGR to RGB + img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + + # Convert to grayscale + gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY) + + # Simple threshold - keep it simple + _, gray = cv2.threshold(gray, 128, 255, cv2.THRESH_BINARY) + + # Light denoise + gray = cv2.fastNlMeansDenoising(gray, h=3) + + # Very light morphological close - only connect broken parts + kernel = np.ones((1, 1), np.uint8) + gray = cv2.morphologyEx(gray, cv2.MORPH_CLOSE, kernel) + + # Resize 6x for better recognition (larger for 6-char captcha) + gray = cv2.resize(gray, None, fx=6, fy=6, interpolation=cv2.INTER_CUBIC) + + # Convert back to RGB (3-channel) for EasyOCR + rgb = cv2.cvtColor(gray, cv2.COLOR_GRAY2RGB) + + return Image.fromarray(rgb) + + +def apply_confusion_corrections(text: str) -> str: + """Apply character confusion corrections based on common OCR errors.""" + corrected = [] + for ch in text: + corrected.append(CONFUSION_MAP.get(ch, ch)) + return ''.join(corrected) + + +def ocr_captcha(image: Image.Image) -> str: + """Run OCR on preprocessed captcha image using EasyOCR.""" + if _EASYOCR_READER is None: + raise RuntimeError("EasyOCR not installed. Run: pip install easyocr") + + img_array = np.array(image) + + # EasyOCR with alphanumeric allowlist for 6-char captcha + results = _EASYOCR_READER.readtext( + img_array, + detail=0, + allowlist='ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789', + width_ths=0.5, + paragraph=False + ) + + # Combine all detected text + result = ''.join(results) + + # Clean result - keep only alphanumeric, max 6 chars + cleaned = ''.join(filter(str.isalnum, result))[:6] + + print(f" DEBUG OCR: result='{result}', cleaned='{cleaned}', len={len(cleaned)}") + return cleaned + + +def find_refresh_button(driver: WebDriver, captcha_input: WebElement) -> Optional[WebElement]: + """Find captcha refresh button near the captcha input.""" + parent = captcha_input.find_element(By.XPATH, "..") + for sibling in parent.find_elements(By.XPATH, "*"): + onclick = (sibling.get_attribute("onclick") or "").lower() + class_attr = (sibling.get_attribute("class") or "").lower() + id_attr = (sibling.get_attribute("id") or "").lower() + if any(kw in onclick for kw in ["refresh", "captcha", "reload", "newcaptcha"]): + return sibling + if any(kw in class_attr for kw in ["refresh", "captcha-refresh", "reload"]): + return sibling + if any(kw in id_attr for kw in ["refresh", "captcha", "reload"]): + return sibling + + # Fallback: search globally for refresh-like elements + try: + candidates = driver.find_elements(By.XPATH, "//*[self::a or self::button or self::img or self::span][contains(translate(@onclick, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'refresh') or contains(translate(@class, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'refresh') or contains(translate(@title, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'refresh')]") + for cand in candidates: + if cand.is_displayed(): + return cand + except: + pass + return None + + +def wait_for_new_captcha(driver: WebDriver, captcha_input: WebElement, timeout: int = 10) -> bool: + """Wait for captcha background-image to change (new challenge loaded).""" + old_style = captcha_input.get_attribute("style") or "" + end_time = time.time() + timeout + while time.time() < end_time: + try: + new_style = captcha_input.get_attribute("style") or "" + if new_style != old_style and "data:image" in new_style: + return True + except StaleElementReferenceException: + captcha_input = find_visible_captcha_input(driver) + old_style = captcha_input.get_attribute("style") or "" + time.sleep(0.3) + return False + + +def solve_and_fill_captcha(driver: WebDriver, max_retries: int = 5, attempt_callback=None) -> bool: + """ + Main entry point: solve captcha and fill input. + Returns True if filled successfully, False if max retries exceeded. + Each retry gets a fresh captcha via refresh button. + """ + for attempt in range(1, max_retries + 1): + captcha_text = "" + try: + captcha_input = find_visible_captcha_input(driver) + + # Wait for captcha base64 to be populated (async load) + if not wait_for_captcha_load(driver, captcha_input): + print(f" [Attempt {attempt}] Captcha failed to load in time, retrying...") + if attempt < max_retries: + refresh_btn = find_refresh_button(driver, captcha_input) + if refresh_btn: + refresh_btn.click() + wait_for_new_captcha(driver, captcha_input) + if attempt_callback: + attempt_callback(attempt, "", "", "not_loaded") + continue + + image_bytes = extract_captcha_base64(captcha_input) + pil_image = preprocess_captcha_image(image_bytes) + captcha_text = ocr_captcha(pil_image) + + if not captcha_text or len(captcha_text) < 6: + print(f" [Attempt {attempt}] OCR weak result: '{captcha_text}' (len={len(captcha_text)}), retrying...") + if attempt < max_retries: + refresh_btn = find_refresh_button(driver, captcha_input) + if refresh_btn: + refresh_btn.click() + wait_for_new_captcha(driver, captcha_input) + if attempt_callback: + attempt_callback(attempt, captcha_text, "", "ocr_failed") + continue + + # Apply confusion corrections to get best variant + corrected = apply_confusion_corrections(captcha_text) + + # Generate case variants for case-sensitive captcha (mixed case + numbers) + variants = [ + captcha_text, # as-is (from OCR) + corrected, # with confusion corrections + captcha_text.upper(), # all uppercase + captcha_text.capitalize(), # first letter uppercase + captcha_text.title(), # each word capitalized + captcha_text.swapcase(), # swap case + ] + + # Add variations with first char variations + if len(captcha_text) >= 1: + variants.append(captcha_text[0].upper() + captcha_text[1:]) + variants.append(captcha_text[0].lower() + captcha_text[1:]) + if len(captcha_text) >= 2: + variants.append(captcha_text[:2].upper() + captcha_text[2:]) + variants.append(captcha_text[:2].lower() + captcha_text[2:]) + if len(captcha_text) >= 3: + variants.append(captcha_text[:3].upper() + captcha_text[3:]) + variants.append(captcha_text[:3].lower() + captcha_text[3:]) + + # Remove duplicates while preserving order, keep only 6-char variants + seen = set() + unique_variants = [] + for v in variants: + if v not in seen and len(v) == 6: + seen.add(v) + unique_variants.append(v) + variants = unique_variants + + if attempt_callback: + screenshot_path = attempt_callback(attempt, captcha_text, "", "before_entry") or "" + + success = False + # Try ALL variants before giving up on this attempt + for variant in variants: + if len(variant) == 6: + print(f" [Attempt {attempt}] Trying variant: '{variant}'") + captcha_input.clear() + captcha_input.send_keys(variant) + success = True + if attempt_callback: + attempt_callback(attempt, captcha_text, variant, "filled") + break + + if not success: + print(f" [Attempt {attempt}] No valid 6-char variant found for '{captcha_text}', retrying...") + if attempt < max_retries: + refresh_btn = find_refresh_button(driver, captcha_input) + if refresh_btn: + refresh_btn.click() + wait_for_new_captcha(driver, captcha_input) + continue + + # Click submit + submit_btn = WebDriverWait(driver, 10).until( + EC.element_to_be_clickable((By.XPATH, "//button[@value='Login']")) + ) + submit_btn.click() + + # Wait a bit for response + time.sleep(2) + + # Check for invalid captcha error using the exact xpath + try: + # First handle any alerts that might be open + try: + alert = driver.switch_to.alert + alert_text = alert.text + print(f" DEBUG: Alert detected after submit: {alert_text}") + if "logout" in alert_text.lower() or "session" in alert_text.lower() or "previous" in alert_text.lower(): + alert.accept() + print(" DEBUG: Accepted session/logout alert") + time.sleep(2) + except: + pass + + invalid_captcha_elem = WebDriverWait(driver, 3).until( + EC.visibility_of_element_located(( + By.XPATH, + "//div[@id='error']//strong[contains(normalize-space(), 'Invalid Captcha')]", + )) + ) + print(f" [Attempt {attempt}] Invalid captcha detected: {invalid_captcha_elem.text}") + if attempt < max_retries: + refresh_btn = find_refresh_button(driver, captcha_input) + if refresh_btn: + refresh_btn.click() + wait_for_new_captcha(driver, captcha_input) + continue + except TimeoutException: + # No invalid captcha error - likely success + print(f" [Attempt {attempt}] No invalid captcha error detected, assuming success") + return True + except Exception as e: + # Handle other exceptions like NoSuchElementException + print(f" [Attempt {attempt}] Error checking invalid captcha: {type(e).__name__}: {e}") + if attempt < max_retries: + refresh_btn = find_refresh_button(driver, captcha_input) + if refresh_btn: + refresh_btn.click() + wait_for_new_captcha(driver, captcha_input) + continue + + except Exception as e: + print(f" [Attempt {attempt}] Error: {e}") + if attempt < max_retries: + refresh_btn = find_refresh_button(driver, captcha_input) + if refresh_btn: + refresh_btn.click() + wait_for_new_captcha(driver, captcha_input) + + return False diff --git a/tests/config_reader.py b/tests/config_reader.py index 84a5063..ba59aab 100644 --- a/tests/config_reader.py +++ b/tests/config_reader.py @@ -67,7 +67,10 @@ def get_setup_configs(mode, config=None): session = setup.get("run_session", True) token = setup.get("run_token", True) matches = { - "combined": session and token or (name.startswith("UPC_") and token), + # The full setup run owns every enabled product that has work to do. + # Products without tokens (for example RA) still belong in this + # shared browser flow; the combined test skips only their token step. + "combined": session or token, "session": session and not token, "token": token and not session, } diff --git a/tests/conftest.py b/tests/conftest.py index d4bdb3c..d79a94b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -47,6 +47,30 @@ def pytest_configure(config): config.option.self_contained_html = True +def pytest_collection_modifyitems(config, items): + """Keep the full setup run limited to the shared combined flow.""" + invocation = " ".join( + str(arg).replace("\\", "/") for arg in config.invocation_params.args + ).lower() + full_setup_run = ( + "tests/setup" in invocation + and "tests/setup/sessions" not in invocation + and "tests/setup/tokens" not in invocation + and "test_session.py" not in invocation + and "test_trading_token.py" not in invocation + ) + if not full_setup_run: + return + + deselected = [ + item for item in items + if "/tests/setup/sessions/" in f"/{item.nodeid.replace('\\', '/')}" + or "/tests/setup/tokens/" in f"/{item.nodeid.replace('\\', '/')}" + ] + items[:] = [item for item in items if item not in deselected] + config.hook.pytest_deselected(items=deselected) + + # ----------------------------- # WebDriver Fixture # ----------------------------- diff --git a/tests/helpers.py b/tests/helpers.py index 4259378..1bb0046 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -57,43 +57,216 @@ def enter_pressed(): return pressed -def login(driver, username, password): +def check_captcha_error(driver) -> str: + """Check if there's a CAPTCHA error message on the page.""" + try: + # First handle any alerts + try: + alert = driver.switch_to.alert + alert_text = alert.text + print(f" DEBUG: Alert detected in check_captcha_error: {alert_text}") + if "logout" in alert_text.lower() or "session" in alert_text.lower() or "previous" in alert_text.lower(): + alert.accept() + print(" DEBUG: Accepted session/logout alert") + time.sleep(2) + except: + pass + + error_selectors = [ + "//div[@id='error']//strong[normalize-space()='Invalid Captcha']", + "//div[@id='error']//span[contains(@class,'help-block')]//strong[contains(text(),'Invalid')]", + "//*[contains(@class,'error') and contains(text(),'Invalid Captcha')]", + ] + for selector in error_selectors: + elements = driver.find_elements(By.XPATH, selector) + for el in elements: + if el.is_displayed() and el.text.strip(): + return el.text.strip() + except Exception as e: + print(f" DEBUG: check_captcha_error exception: {e}") + return "" + + +def check_login_success(driver) -> bool: + """Check if login was successful using multiple indicators.""" + try: + # Handle any pending alerts first + try: + alert = driver.switch_to.alert + alert_text = alert.text + print(f" DEBUG: Alert detected: {alert_text}") + if "logout" in alert_text.lower() or "session" in alert_text.lower() or "previous" in alert_text.lower(): + alert.accept() + print(" DEBUG: Accepted session/logout alert") + time.sleep(2) + except: + pass + + # Wait a bit for page to load after submit + time.sleep(3) + + # Print current URL and page title for debugging + print(f" DEBUG: Current URL: {driver.current_url}") + print(f" DEBUG: Page title: {driver.title}") + + # Multiple dashboard indicators + dashboard_indicators = [ + "//span[normalize-space()='TRADING MAINTENANCE']", + "//a[contains(@href, 'tradingMaintenance')]", + "//span[contains(text(), 'Dashboard')]", + "//h1[contains(text(), 'Dashboard')]", + "//div[contains(@class, 'dashboard')]", + "//a[contains(text(), 'Trading')]", + "//div[contains(@class, 'navbar')]//span[contains(text(), 'Trading')]", + ] + + wait = WebDriverWait(driver, 15) + for indicator in dashboard_indicators: + try: + wait.until(EC.visibility_of_element_located((By.XPATH, indicator))) + print(f" DEBUG: Found dashboard indicator: {indicator}") + return True + except: + continue + + # URL-based check + if "home" in driver.current_url.lower() or "dashboard" in driver.current_url.lower(): + print(" DEBUG: Login success detected via URL") + return True + + return False + except Exception as e: + print(f" DEBUG: check_login_success failed: {e}") + return False + + +def login( + driver, + username, + password, + max_captcha_retries=5, + captcha_metrics=None, + max_login_duration_seconds=1200, +): wait = WebDriverWait(driver, 20) - # username - userid_field = wait.until( - EC.visibility_of_element_located((By.XPATH, Locator.userid)) - ) - userid_field.send_keys(username) + # --- AUTOMATED CAPTCHA SOLVING WITH RETRY ON FAILURE --- + from tests.captcha_solver import solve_and_fill_captcha, find_refresh_button, wait_for_new_captcha + + login_deadline = time.monotonic() + max_login_duration_seconds + captcha_attempt = 0 + while True: + captcha_attempt += 1 - # password - password_field = driver.find_element(By.XPATH, Locator.password) - password_field.send_keys(password) + # Refreshing after a failed CAPTCHA clears the login form, so refill + # both credentials before every automatic or manual retry. + userid_field = wait.until( + EC.visibility_of_element_located((By.XPATH, Locator.userid)) + ) + userid_field.clear() + userid_field.send_keys(username) - print(f"Solve captcha within {LOGIN_CAPTCHA_WAIT_SECONDS} seconds...") + password_field = wait.until( + EC.visibility_of_element_located((By.XPATH, Locator.password)) + ) + password_field.clear() + password_field.send_keys(password) - # wait for manual captcha solve - time.sleep(LOGIN_CAPTCHA_WAIT_SECONDS) + if captcha_metrics is not None: + captcha_metrics["tries"] = captcha_attempt + captcha_metrics["max_retries"] = max_captcha_retries + print(f"Solving CAPTCHA (attempt {captcha_attempt})...") - # click submit - submit_btn = wait.until( - EC.element_to_be_clickable((By.XPATH, Locator.submitbtn)) - ) - submit_btn.click() + # Once automatic solving has run for the configured duration, ask for + # the CAPTCHA value in a browser prompt and continue the same login flow. + if time.monotonic() >= login_deadline: + print("Automatic CAPTCHA retries timed out; waiting for manual CAPTCHA entry.") + manual_value = driver.execute_script( + "return window.prompt('Automatic CAPTCHA solving timed out. Enter the CAPTCHA shown on the page:');" + ) + if not manual_value: + raise RuntimeError("Manual CAPTCHA entry was cancelled") - time.sleep(LOGIN_POST_SUBMIT_WAIT_SECONDS) + captcha_input = wait.until( + EC.visibility_of_element_located((By.NAME, "captcha")) + ) + captcha_input.clear() + captcha_input.send_keys(manual_value.strip()) + wait.until(EC.element_to_be_clickable((By.XPATH, Locator.submitbtn))).click() + time.sleep(LOGIN_POST_SUBMIT_WAIT_SECONDS) + accept_browser_alert_if_present(driver) + handle_transient_popups(driver) - accept_browser_alert_if_present(driver) + captcha_error = check_captcha_error(driver) + if captcha_error: + print(f" Manual CAPTCHA rejected: {captcha_error}") + continue - # 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") + if check_login_success(driver): + print("Login successful after manual CAPTCHA entry") + if captcha_metrics is not None: + captcha_metrics["correct_attempt"] = captcha_attempt + captcha_metrics.setdefault("captcha_attempts", []).append({ + "attempt": captcha_attempt, + "captcha_text": "manual", + "variant": manual_value.strip(), + "status": "manual", + "screenshot": "", + "error": "", + }) + close_tac_expiry_modal_if_present(driver) + return True + continue + + def attempt_callback(attempt_num, captcha_text, variant, status, screenshot_path="", error=""): + if captcha_metrics is not None: + captcha_metrics.setdefault("captcha_attempts", []).append({ + "attempt": attempt_num, + "captcha_text": captcha_text, + "variant": variant, + "status": status, + "screenshot": screenshot_path, + "error": error + }) + + if not solve_and_fill_captcha(driver, max_retries=1, attempt_callback=attempt_callback): + print(" Automatic CAPTCHA solve failed; retrying until the login deadline.") + driver.refresh() + time.sleep(3) + continue + + # solve_and_fill_captcha already clicks submit, so just wait and check result + time.sleep(LOGIN_POST_SUBMIT_WAIT_SECONDS) + + accept_browser_alert_if_present(driver) + + # Handle any popups that appear + handle_transient_popups(driver) + + # Check for CAPTCHA error (invalid captcha message) + captcha_error = check_captcha_error(driver) + if captcha_error: + print(f" CAPTCHA validation failed: {captcha_error}") + # Reload the page to get a fresh CAPTCHA and keep retrying until + # the overall login deadline is reached. + print(f" Reloading page to get fresh CAPTCHA (attempt {captcha_attempt + 1})...") + driver.refresh() + time.sleep(3) + continue + + # Check for successful login - use improved detection + if check_login_success(driver): + print("Login successful") + if captcha_metrics is not None: + captcha_metrics["correct_attempt"] = captcha_attempt + close_tac_expiry_modal_if_present(driver) + return True + + # If we get here, something else is wrong + print(" Login not confirmed, retrying...") + driver.refresh() + time.sleep(3) def accept_browser_alert_if_present(driver, timeout=5): @@ -309,23 +482,34 @@ 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) + def visible_elements(xpath): + return [element for element in driver.find_elements(By.XPATH, xpath) if element.is_displayed()] - submenu = driver.find_elements(By.XPATH, menu_xpath) - submenu_visible = any(item.is_displayed() for item in submenu) + submenu_visible = any( + item.is_enabled() for item in visible_elements(menu_xpath) + ) if not submenu_visible: + maintenance_items = visible_elements(Locator.tradingmaintenance) + if not maintenance_items: + wait.until(lambda current_driver: visible_elements(Locator.tradingmaintenance)) + maintenance_items = visible_elements(Locator.tradingmaintenance) + + maintenance = maintenance_items[0] + driver.execute_script( + "arguments[0].scrollIntoView({block: 'center'});", maintenance + ) try: - wait.until(EC.element_to_be_clickable((By.XPATH, Locator.tradingmaintenance))) - trading_maintenance.click() + wait.until(lambda current_driver: maintenance.is_enabled()) + maintenance.click() except Exception: - driver.execute_script("arguments[0].click();", trading_maintenance) + driver.execute_script("arguments[0].click();", maintenance) menu_item = wait.until( - EC.element_to_be_clickable((By.XPATH, menu_xpath)) + lambda current_driver: next( + (item for item in visible_elements(menu_xpath) if item.is_enabled()), + False, + ) ) try: menu_item.click() diff --git a/tests/locators.py b/tests/locators.py index 5ce4763..3e8aa71 100644 --- a/tests/locators.py +++ b/tests/locators.py @@ -4,11 +4,13 @@ class Locator(object): userid="//input[@placeholder='User ID']" password="//input[@placeholder='Password']" submitbtn="//button[@value='Login']" - dashboard_marker="//span[normalize-space()='TRADING MAINTENANCE']" + dashboard_marker="//*[contains(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']" + # Click the Angular menu anchor, not the nested label span. The anchor + # owns openSubMenu(item), which is required to render its child links. + tradingmaintenance="//a[.//span[normalize-space()='TRADING MAINTENANCE']]" tradingsession="//a[normalize-space()='Trading Session Master']" tokengeneration="//a[@href='#/tradingMaintenance/generateToken' or normalize-space()='Token Generation']" tokengenerator = "//a[@href='#/tradingMaintenance/tokenGenerator']" @@ -37,6 +39,16 @@ class Locator(object): tokendeliveryenddate = "//input[@placeholder='DELIVERY_END_DATE']" tokenexpirydate = "//input[@placeholder='Token Expiry Date']" tokentype = "//input[@name='TYPE' or @placeholder='TYPE']" + intradaydeliveryfromdate = "//input[@placeholder='Delivery From Date']" + intradaydeliverytodate = "//input[@placeholder='Delivery To Date']" + intradaytradingstarttime = "//input[@placeholder='Trading Start Time']" + intradaydaysbeforedelivery = "//input[@placeholder='Trading Start days before delivery']" + intradaydurationbeforeend = "//input[@placeholder='Duration before delivery end time']" + intradaysuccessmessage = "//*[contains(normalize-space(text()), 'Tokens generated successfully')]" + dactradingfromdate = "//input[@placeholder='Trading From Date']" + dactradingtodate = "//input[@placeholder='Trading To Date']" + dacsessionstarttime = "//input[@placeholder='Session Start Time']" + dacsessionendtime = "//input[@placeholder='Session End Time']" tokentradedate = "//input[@placeholder='Trade Date']" upcordernormalstart = "//input[@name='ses3StartTime']" upcordernormalend = "//input[@name='ses3EndTime']" diff --git a/tests/setup/sessions/test_session.py b/tests/setup/sessions/test_session.py index dce576b..c2d1294 100644 --- a/tests/setup/sessions/test_session.py +++ b/tests/setup/sessions/test_session.py @@ -13,13 +13,21 @@ from tests.locators import Locator def get_session_window(config): session_config = config.get("setup", {}).get("session", {}) - start_offset_minutes = session_config.get("start_offset_minutes", 7) + time_mode = session_config.get("time_mode", "offset").lower() + if time_mode == "manual": + start_time = session_config.get("start_time") + if not start_time: + raise ValueError("Session time_mode 'manual' requires start_time") + elif time_mode == "offset": + start_offset_minutes = session_config.get("start_offset_minutes", 7) + start_time = (datetime.now() + timedelta(minutes=start_offset_minutes)).strftime("%H:%M") + else: + raise ValueError("Session time_mode must be 'manual' or 'offset'") 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}" + return f"{today} {start_time}", f"{today} {end_time}" def create_session(driver, step, config): @@ -44,7 +52,13 @@ def create_session(driver, step, config): @pytest.mark.session @pytest.mark.parametrize("config", get_setup_configs("session"), ids=lambda c: c["product_name"]) -def test_session(driver, test_context, config): +def test_session(driver, test_context, config, request): + + # The full setup command runs session-only products (such as RA) through + # test_combined_setup.py so they share the same browser/login context. + invocation = " ".join(str(arg).replace("\\", "/") for arg in request.config.invocation_params.args) + if "tests/setup" in invocation and "tests/setup/sessions" not in invocation and "test_session.py" not in invocation: + pytest.skip("Handled by the shared full setup flow") wait = WebDriverWait(driver, 20) diff --git a/tests/setup/test_combined_setup.py b/tests/setup/test_combined_setup.py index 24c5629..33b91ac 100644 --- a/tests/setup/test_combined_setup.py +++ b/tests/setup/test_combined_setup.py @@ -31,8 +31,9 @@ def test_session_and_token_setup(driver, test_context, config): if config.get("setup", {}).get("run_session", True): create_session(driver, step, config) - 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) + if config.get("setup", {}).get("run_token", True): + 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) diff --git a/tests/setup/tokens/test_trading_token.py b/tests/setup/tokens/test_trading_token.py index 06346ca..f557e2c 100644 --- a/tests/setup/tokens/test_trading_token.py +++ b/tests/setup/tokens/test_trading_token.py @@ -56,15 +56,28 @@ def get_delivery_date_and_expiry(token): 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 start_time is None: + if time_mode == "offset": 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") + 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, @@ -72,6 +85,29 @@ def get_time_session_times(token, row_text, default_start_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"))) @@ -203,10 +239,19 @@ def open_rtm_new_token(driver, step, token): 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')}", - (now + timedelta(minutes=token.get("session_start_offset_minutes", 5))).strftime("%H:%M"), + session_start_time, token.get("session_end_time", "20:00"), ) @@ -227,6 +272,76 @@ def create_rtm_token(driver, step, token): 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) @@ -240,7 +355,7 @@ def create_upc_token(driver, step, token): 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, timings.get("order_normal_start_time")), + ("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")), @@ -254,6 +369,12 @@ def create_upc_token(driver, step, token): 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 @@ -291,7 +412,11 @@ def create_and_activate_token(driver, step, token): step.click((By.XPATH, Locator.addnewtokenbtn), f"click_add_new_token_{product_code.lower()}") - next_day, token_expiry = get_delivery_date_and_expiry(token) + 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") @@ -303,9 +428,7 @@ def create_and_activate_token(driver, step, token): 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 = timings.get("order_normal_start_time") - if normal_start is None and timings.get("order_normal_start_offset_minutes") is not None: - normal_start = (datetime.now() + timedelta(minutes=timings["order_normal_start_offset_minutes"])).strftime("%H:%M") + normal_start = get_upc_order_normal_start_time(timings) values = { Locator.upcordernormalstart: normal_start, Locator.upcordernormalend: timings.get("order_normal_end_time"), @@ -317,14 +440,17 @@ def create_and_activate_token(driver, step, token): continue type_first_visible_input(driver, locator, value) - if ( + 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 token.get("generation_mode") not in ("WeeklyUPC", "MonthlyUPC"): + 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()}") @@ -334,12 +460,19 @@ def create_and_activate_token(driver, step, token): step.click((By.XPATH, Locator.tokengeneratorresetbtn), f"reset_token_generator_{product_code.lower()}") return - complete_time_session_edits_from_saved_token(driver, step, token) + 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): +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) diff --git a/tests/test_captcha_login_stress.py b/tests/test_captcha_login_stress.py new file mode 100644 index 0000000..ad50783 --- /dev/null +++ b/tests/test_captcha_login_stress.py @@ -0,0 +1,124 @@ +import csv +import os +import time +from datetime import datetime + +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.browser import create_chrome_driver +from tests.config_reader import load_config +from tests.helpers import login, logout +from tests.locators import Locator + + +LOGIN_ATTEMPTS = 50 +MAX_CAPTCHA_RETRIES = 7 + + +@pytest.mark.captcha +def test_captcha_login_stress(): + """Run bounded EasyOCR login attempts and retain auditable CAPTCHA evidence.""" + config = load_config() + output_dir = os.path.abspath(os.path.join("reports", "captcha_login_stress")) + os.makedirs(output_dir, exist_ok=True) + started = datetime.now().strftime("%Y%m%d_%H%M%S") + csv_path = os.path.join(output_dir, f"captcha_login_{started}.csv") + rows = [] + passed = 0 + failed = 0 + + def record(row): + rows.append(row) + + for login_number in range(1, LOGIN_ATTEMPTS + 1): + login_started = datetime.now().isoformat(timespec="seconds") + captcha_rows = [] + + driver = create_chrome_driver() + try: + driver.get(config["url"]) + + login_status = "FAILED" + logout_status = "NOT_ATTEMPTED" + error_text = "" + captcha_metrics = {"captcha_attempts": []} + try: + login( + driver, + config["login"]["username"], + config["login"]["password"], + max_captcha_retries=MAX_CAPTCHA_RETRIES, + captcha_metrics=captcha_metrics, + max_login_duration_seconds=1200, + ) + login_status = "PASSED" + passed += 1 + except Exception as exc: + failed += 1 + error_text = f"{type(exc).__name__}: {exc}" + try: + driver.save_screenshot(os.path.join(output_dir, f"login_{login_number:03d}_failure.png")) + except Exception: + pass + finally: + # Explicitly log out after a successful authenticated flow, + # then close the browser session. + if login_status == "PASSED": + try: + logout_status = "PASSED" if logout(driver) else "FAILED" + except Exception as exc: + logout_status = "FAILED" + print(f"Logout error: {type(exc).__name__}: {exc}") + + try: + driver.quit() + except Exception: + pass + + captcha_rows = captcha_metrics.get("captcha_attempts", []) or [{ + "attempt": 0, "captcha_text": "", "variant": "", + "status": "no_captcha_record", "screenshot": "", "error": error_text, + }] + for item in captcha_rows: + record({ + "login_attempt": login_number, + "started_at": login_started, + "login_status": login_status, + "logout_status": logout_status, + "captcha_attempt": item["attempt"], + "attempted_captcha": item["variant"], + "ocr_captcha": item["captcha_text"], + "screenshot_path": item["screenshot"], + "captcha_status": item["status"], + "captcha_entered_count": sum( + 1 for captcha_row in captcha_rows if captcha_row["variant"] + ), + "error": item["error"] or error_text, + "attempts_to_result": len(captcha_rows), + "passed_count": passed, + "failed_count": failed, + }) + finally: + # Clean up the driver after all attempts + try: + driver.quit() + except Exception: + pass + + fieldnames = [ + "login_attempt", "started_at", "login_status", "captcha_attempt", + "logout_status", + "attempted_captcha", "ocr_captcha", "screenshot_path", "captcha_status", + "captcha_entered_count", "error", "attempts_to_result", "passed_count", + "failed_count", + ] + with open(csv_path, "w", newline="", encoding="utf-8") as report: + writer = csv.DictWriter(report, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(rows) + + print(f"CAPTCHA stress report: {csv_path}") + print(f"Passed: {passed}; Failed: {failed}; Total: {LOGIN_ATTEMPTS}")