#!/usr/bin/env python3 """ E2E API Test Runner for NexusCRM. Parses JSON test cases and executes them against the NexusCRM API. """ import sys sys.stdout.reconfigure(line_buffering=True) # Force line buffering for real-time logs in CI import argparse import glob import hashlib import json import os import re import signal import sys import threading import time import uuid from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait from datetime import datetime, timezone from typing import Dict, List, Any, Optional from urllib.parse import quote_plus import config import api import mysql.connector E2E_CONNECTED_APP_ID = "test-gosf-client" E2E_CONNECTED_APP_NAME = "Test GoSF Client" E2E_CONNECTED_APP_SECRET = "test-gosf-secret-123" class CaseTimeoutError(RuntimeError): """Raised when a case exceeds its total runtime budget.""" class RunTimeoutError(RuntimeError): """Raised when the whole run exceeds its total runtime budget.""" class ActiveCaseRegistry: """Track active cases across parallel scene workers for CI heartbeat logs.""" def __init__(self): self._lock = threading.Lock() self._active: Dict[str, Dict[str, Any]] = {} def start(self, scene: str, case_id: str, case_name: str) -> None: key = f"{scene}:{case_id}" with self._lock: self._active[key] = { "scene": scene, "case_id": case_id, "case_name": case_name, "started_at": time.time(), } def finish(self, scene: str, case_id: str) -> None: key = f"{scene}:{case_id}" with self._lock: self._active.pop(key, None) def snapshot(self) -> List[Dict[str, Any]]: now = time.time() with self._lock: items = list(self._active.values()) items.sort(key=lambda item: item["started_at"]) snapshot = [] for item in items: snapshot.append({ "scene": item["scene"], "case_id": item["case_id"], "case_name": item["case_name"], "elapsed_seconds": int(now - item["started_at"]), }) return snapshot ACTIVE_CASES = ActiveCaseRegistry() class TestRunner: """Executes E2E API test cases from JSON files.""" BILLING_COUNTRY_CODE_POOL = ( "US", "CA", "GB", "DE", "FR", "IT", "ES", "NL", "BE", "CH", "AT", "SE", "NO", "DK", "FI", "IE", "PT", "PL", "CZ", "HU", "RO", "GR", "TR", "IL", "AE", "SA", "ZA", "EG", "IN", "SG", "MY", "TH", "VN", "PH", "ID", "JP", "KR", "AU", "NZ", "BR", "AR", "CL", "CO", "PE", "MX", "CR", "PA", "DO", "UY", "EC", ) def __init__(self, base_path: str = None): self.base_path = base_path or os.path.dirname(os.path.abspath(__file__)) self.cases_dir = os.path.join(self.base_path, "cases") self.seed_data = self._load_seed_data() self.results = [] self.client = api.NexusAPIClient() self._client_cache = {} self.variables = {} # Runtime variables (e.g., created record IDs) self._db_pool: Optional[mysql.connector.MySQLConnection] = None self._run_deadline: Optional[float] = None self._case_deadline: Optional[float] = None self._current_case_id: str = "" self._current_scene: str = "" self._current_case_name: str = "" self._last_http_request: Optional[Dict[str, Any]] = None self._last_query_after: Optional[Dict[str, Any]] = None self._last_teardown_step: Optional[Dict[str, Any]] = None def _load_seed_data(self) -> Dict[str, List[Dict]]: """Load seed data from seed_data.json.""" seed_path = os.path.join(self.base_path, "seed_data.json") if os.path.exists(seed_path): with open(seed_path, "r") as f: return json.load(f) return {} def _build_case_variables(self, case: Dict) -> Dict[str, Any]: """Build per-case runtime variables, including unique built-ins.""" now = datetime.now(timezone.utc) case_key = f"{case.get('scene', '')}:{case.get('id', '')}:{uuid.uuid4()}" variables = { "run_uuid": str(uuid.uuid4()), "timestamp": now.strftime("%Y%m%d-%H%M%S"), "today": now.strftime("%Y-%m-%d"), "db_timestamp": now.strftime("%Y-%m-%d %H:%M:%S"), # Spread compound autonumber prefixes across real ISO-like values so # shared ci-test runs do not all pile onto the same per-prefix/day counter. "billing_country_code": self._select_billing_country_code(case_key), } for name, value in case.get("variables", {}).items(): variables[name] = self._substitute_variables(value, variables) return variables def _select_billing_country_code(self, seed: str) -> str: digest = hashlib.sha256(seed.encode("utf-8")).digest() index = int.from_bytes(digest[:2], byteorder="big") % len(self.BILLING_COUNTRY_CODE_POOL) return self.BILLING_COUNTRY_CODE_POOL[index] def _lookup_seed_value(self, pattern: str) -> Any: """Resolve seed references from seed_data.json. Supported forms: {seed_Object_0} -> seed_data[object][0]["Id"] (default Id) {seed_Object_0_FieldName} -> seed_data[object][0]["FieldName"] (exact case-sensitive field key lookup, falls back to case-insensitive match) """ parts = pattern.split("_") if len(parts) < 2: return None table_name = parts[0].lower() try: index = int(parts[1]) except ValueError: return None if table_name not in self.seed_data: return None records = self.seed_data[table_name] if index >= len(records): return None record = records[index] if len(parts) >= 3: # Rejoin remaining parts — field names can contain underscores # (e.g. Account_Name_for_email_template__c would split). field = "_".join(parts[2:]) if field in record: return record[field] # Case-insensitive fallback lowered = {k.lower(): v for k, v in record.items()} return lowered.get(field.lower()) return record.get("Id", record.get("id")) def _substitute_variables(self, obj: Any, variables: Optional[Dict[str, Any]] = None) -> Any: """ Replace {seed_Account_0}, {seed_Lead_1}, etc. with actual values from seed_data.json. Also handles {variable_name} for runtime variables. """ variables = self.variables if variables is None else variables if isinstance(obj, str): full_seed_match = re.fullmatch(r"\{seed_([^}]+)\}", obj) if full_seed_match: seed_value = self._lookup_seed_value(full_seed_match.group(1)) return obj if seed_value is None else seed_value full_var_match = re.fullmatch(r"\{([^}]+)\}", obj) if full_var_match: var_name = full_var_match.group(1) if var_name in variables: return variables[var_name] def replace_seed(match): seed_value = self._lookup_seed_value(match.group(1)) return match.group(0) if seed_value is None else str(seed_value) result = re.sub(r"\{seed_([^}]+)\}", replace_seed, obj) def replace_var(match): var_name = match.group(1) if var_name not in variables: return match.group(0) return str(variables[var_name]) result = re.sub(r"\{([^}]+)\}", replace_var, result) return result if isinstance(obj, dict): return {k: self._substitute_variables(v, variables) for k, v in obj.items()} if isinstance(obj, list): return [self._substitute_variables(item, variables) for item in obj] return obj def _resolve_path(self, data: Any, path: str) -> Any: """Resolve dotted paths like records.0.Id against dict/list payloads.""" if path == "": return data current = data for segment in path.split("."): if segment == "": continue if segment == "length": if isinstance(current, (list, dict, str)): current = len(current) continue return None if isinstance(current, list): if not segment.isdigit(): return None index = int(segment) if index < 0 or index >= len(current): return None current = current[index] continue if isinstance(current, dict): if segment not in current: return None current = current[segment] continue return None return current def _normalize_steps(self, steps: Any) -> List[Dict[str, Any]]: """Normalize setup/teardown blocks into a list of step dicts.""" if not steps: return [] if isinstance(steps, list): return steps if isinstance(steps, dict): return [steps] raise ValueError(f"Unsupported step definition: {type(steps).__name__}") def _default_expected_status(self, method: str) -> int: """Return the default success status for an HTTP method.""" method = method.upper() if method == "GET": return 200 if method == "POST": return 201 if method in ("PATCH", "DELETE"): return 204 return 200 def _resolve_case_timeout_seconds(self, case: Dict[str, Any]) -> int: """Return total runtime budget for one case.""" raw = case.get("case_timeout_seconds") if raw is None: raw = os.environ.get("E2E_API_CASE_TIMEOUT_SECONDS", "90") timeout = int(raw) return max(timeout, 1) def _resolve_run_timeout_seconds(self) -> int: """Return total runtime budget for the whole run, or 0 when disabled.""" raw = os.environ.get("E2E_API_SOFT_RUN_TIMEOUT_SECONDS", "0").strip() if not raw: return 0 timeout = int(raw) return max(timeout, 0) def _remaining_run_budget(self) -> Optional[float]: """Return remaining seconds in the whole-run budget.""" if self._run_deadline is None: return None return self._run_deadline - time.monotonic() def _build_timeout_diagnostics(self, stage: str) -> Dict[str, Any]: """Collect useful diagnostics for case/run timeout failures.""" return { "stage": stage, "case_id": self._current_case_id, "scene": self._current_scene, "active_cases": ACTIVE_CASES.snapshot(), "last_http_request": self._last_http_request, "last_query_after": self._last_query_after, "last_teardown_step": self._last_teardown_step, } def _assert_run_budget(self, stage: str) -> None: """Raise a rich timeout error when a whole run exceeds its budget.""" remaining = self._remaining_run_budget() if remaining is not None and remaining <= 0: diagnostics = self._build_timeout_diagnostics(stage) raise RunTimeoutError(f"Run timeout exceeded: {json.dumps(diagnostics, ensure_ascii=False)}") def _remaining_case_budget(self) -> Optional[float]: """Return remaining seconds in the current case budget.""" if self._case_deadline is None: return None return self._case_deadline - time.monotonic() def _assert_case_budget(self, stage: str) -> None: """Raise a rich timeout error when a case exceeds its budget.""" remaining = self._remaining_case_budget() if remaining is not None and remaining <= 0: diagnostics = self._build_timeout_diagnostics(stage) raise CaseTimeoutError(f"Case timeout exceeded: {json.dumps(diagnostics, ensure_ascii=False)}") def _assert_time_budgets(self, stage: str) -> None: """Check both whole-run and per-case budgets before doing more work.""" self._assert_run_budget(stage) self._assert_case_budget(stage) def _bounded_request_timeout(self, requested_timeout: int) -> int: """Clamp a per-request timeout to the remaining case budget.""" remaining = self._remaining_case_budget() if remaining is None: return requested_timeout # Leave a tiny buffer so we can surface diagnostics in the runner. bounded = min(int(max(1, remaining)), int(requested_timeout)) return max(bounded, 1) def _emit_active_case_heartbeat(self) -> None: """Print currently running cases so CI logs show what is stuck.""" snapshot = ACTIVE_CASES.snapshot() if not snapshot: return rendered = ", ".join( f"{item['scene']}:{item['case_id']}({item['elapsed_seconds']}s)" for item in snapshot ) print(f"[heartbeat] Active cases: {rendered}") def _build_timeout_results(self, cases: List[Dict[str, Any]], error: str) -> List[Dict[str, Any]]: """Mark not-started cases as errored when the run budget is exhausted.""" results = [] for case in cases: results.append({ "id": case.get("id", "unknown"), "name": case.get("name", ""), "status": "ERROR", "error": error, "expected": case.get("expects", {}), "actual": {}, "bug_ref": case.get("bug_ref"), "scene": case.get("scene"), }) return results def _has_unresolved_placeholders(self, obj: Any) -> bool: """Detect unresolved {token} placeholders after substitution, ignoring %s SQL placeholders.""" if isinstance(obj, str): # Ignore %s placeholders in SQL statements if "%s" in obj: return bool(re.search(r"\{[^}]+\}", obj)) return bool(re.search(r"\{[^}]+\}", obj)) if isinstance(obj, dict): return any(self._has_unresolved_placeholders(v) for v in obj.values()) if isinstance(obj, list): return any(self._has_unresolved_placeholders(v) for v in obj) return False def _extract_step_variables(self, extract_map: Dict[str, str], response: Any, context: str = ""): """Extract variables from a response using dotted paths.""" for variable_name, path in extract_map.items(): value = self._resolve_path(response, path) if value is None: print(f" [DEBUG] Warning: Failed to extract {variable_name} from {path} (context: {context})") self.variables[variable_name] = value def _coerce_number(self, value: Any) -> Optional[float]: """Convert comparable numeric values to float when possible.""" if value is None or isinstance(value, bool): return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): try: return float(value) except ValueError: return None return None def _values_equal(self, actual: Any, expected: Any) -> bool: """Compare values with numeric coercion when both sides look numeric.""" actual_number = self._coerce_number(actual) expected_number = self._coerce_number(expected) if actual_number is not None and expected_number is not None: return actual_number == expected_number return str(actual) == str(expected) def _evaluate_assertions(self, data: Any, assertions: List[Dict[str, Any]]) -> tuple: """Evaluate assertion definitions against a response payload.""" for assertion in assertions: field = assertion.get("field") operator = (assertion.get("operator") or "eq").lower() expected_value = self._substitute_variables(assertion.get("value")) actual_value = self._resolve_path(data, field) if operator in ("eq", "equals"): if not self._values_equal(actual_value, expected_value): return False, f"Field {field}: {actual_value} != {expected_value}" elif operator == "ne": if self._values_equal(actual_value, expected_value): return False, f"Field {field}: {actual_value} == {expected_value}" elif operator == "gt": actual_number = self._coerce_number(actual_value) expected_number = self._coerce_number(expected_value) if actual_number is None or expected_number is None or actual_number <= expected_number: return False, f"Field {field}: {actual_value} is not > {expected_value}" elif operator == "contains": if expected_value not in str(actual_value): return False, f"Field {field} does not contain {expected_value}" elif operator == "not_contains": if expected_value in str(actual_value): return False, f"Field {field} unexpectedly contains {expected_value}" elif operator == "is_null": should_be_null = expected_value is not False if should_be_null and actual_value is not None: return False, f"Field {field} should be null" if not should_be_null and actual_value is None: return False, f"Field {field} should not be null" elif operator == "is_not_null": should_not_be_null = expected_value is not False if should_not_be_null and actual_value is None: return False, f"Field {field} should not be null" if not should_not_be_null and actual_value is not None: return False, f"Field {field} should be null" else: return False, f"Unsupported operator: {operator}" return True, "OK" def _db_connect(self) -> mysql.connector.MySQLConnection: """Get or create a thread-local database connection. Uses threading.local() to ensure each thread has its own connection, avoiding segfaults from shared connections in parallel execution. """ import threading if not hasattr(self, '_db_thread_local'): self._db_thread_local = threading.local() conn = getattr(self._db_thread_local, 'conn', None) if conn is None or not conn.is_connected(): conn = mysql.connector.connect( host=config.TIDB_HOST, port=config.TIDB_PORT, user=config.TIDB_USER, password=config.TIDB_PASSWORD, database=os.environ.get("TIDB_DATABASE", "nexuscrm"), connection_timeout=30, ) self._db_thread_local.conn = conn return conn def _discard_db_connection(self) -> None: """Drop the thread-local DB connection after TiDB Cloud disconnects.""" if not hasattr(self, '_db_thread_local'): return conn = getattr(self._db_thread_local, 'conn', None) if conn is not None: try: conn.close() except Exception: pass self._db_thread_local.conn = None @staticmethod def _is_mysql_connection_error(exc: BaseException) -> bool: """Return true for transient MySQL/TiDB connection failures.""" errno = getattr(exc, "errno", None) if errno in {2006, 2013, 2055}: return True text = str(exc).lower() return any( marker in text for marker in ( "lost connection", "server has gone away", "invalid connection", "unexpected eof", "ssl connection error", ) ) def close(self): """Close thread-local DB connection and HTTP sessions if they exist.""" import threading if hasattr(self, '_db_thread_local'): conn = getattr(self._db_thread_local, 'conn', None) if conn and conn.is_connected(): conn.close() self._db_thread_local.conn = None if getattr(self, "client", None) is not None: self.client.session.close() for client in getattr(self, "_client_cache", {}).values(): client.session.close() self._client_cache.clear() def _get_required_seed_columns(self, cursor, db_table: str) -> List[str]: """Return NOT NULL columns without defaults for the given table.""" table_name = db_table.replace("`", "") cursor.execute( """ SELECT COLUMN_NAME FROM information_schema.columns WHERE table_schema = %s AND table_name = %s AND is_nullable = 'NO' AND column_default IS NULL AND extra NOT LIKE %s ORDER BY ordinal_position """, (os.environ.get("TIDB_DATABASE", "nexuscrm"), table_name, "%auto_increment%"), ) return [row[0] for row in cursor.fetchall()] def _apply_seed_defaults(self, record: Dict[str, Any], required_columns: List[str]) -> Dict[str, Any]: """Backfill common system fields required by strict-mode TiDB schemas.""" normalized = dict(record) timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") default_values = { "Id": lambda: normalized.get("Id") or str(uuid.uuid4()), "CreatedDate": lambda: normalized.get("CreatedDate") or timestamp, "LastModifiedDate": lambda: normalized.get("LastModifiedDate") or timestamp, "SystemModstamp": lambda: normalized.get("SystemModstamp") or timestamp, } for column in required_columns: if column in normalized and normalized[column] not in ("", None): continue factory = default_values.get(column) if factory: normalized[column] = factory() return normalized def _setup_seed_data(self): """Load seed_data.json records into TiDB. Run once before all tests. Truncates relevant tables and re-inserts seed records so that {seed_*} variable references resolve to real DB rows. """ # Table name mapping: seed_data.json key → actual DB table name table_map = { "user": None, # skip - User table requires Password which seed doesn't have "lead": "`Lead`", "account": "Account", "contact": "Contact", "opportunity": "Opportunity", "opportunitylineitem": "OpportunityLineItem", "quote": "Quote", "order": "`Order`", # Order is a reserved word, needs backticks "contract": "Contract", "campaign": "Campaign", "processdefinition": "ProcessDefinition", "processinstance": "ProcessInstance", "product2": "Product2", # skip - table doesn't exist in DB "pricebook2": "Pricebook2", "pricebookentry": "PricebookEntry", "orderitem": "OrderItem", "recordtype": "RecordType", } # Dependency order for INSERT (parent tables first), then TRUNCATE in REVERSE insert_order = ["account", "campaign", "product2", "pricebook2", "processdefinition", "lead", "contact", "opportunity", "opportunitylineitem", "quote", "order", "contract", "pricebookentry", "processinstance"] insert_order = [t for t in insert_order if t in table_map and table_map[t]] max_attempts = 3 for attempt in range(1, max_attempts + 1): try: self._setup_seed_data_once(table_map, insert_order) return except Exception as e: if not self._is_mysql_connection_error(e) or attempt == max_attempts: raise print(f"[setup] TiDB connection lost during seed data setup; retrying {attempt}/{max_attempts}: {e}") self._discard_db_connection() time.sleep(2 * attempt) def _setup_seed_data_once(self, table_map: Dict[str, Optional[str]], insert_order: List[str]): print("[setup] Initializing seed data...") conn = self._db_connect() cursor = conn.cursor() try: # Truncate tables in REVERSE dependency order (children first, FK-safe) # Also truncate FK child tables that reference our tables extra_tables = ["ProcessAction", "ProcessInstanceNode", "OrderItem", "QuoteLineItem", "Product2"] # First, truncate extra tables (including ProcessAction which has FK to ProcessDefinition) for db_table in extra_tables: try: cursor.execute(f"TRUNCATE TABLE {db_table}") print(f" [setup] truncated {db_table}") except mysql.connector.Error as e: if self._is_mysql_connection_error(e): raise print(f" [setup] warning: could not truncate {db_table}: {e}") # Then truncate other tables in reverse order for table in reversed(insert_order): db_table = table_map[table] # Skip ProcessDefinition to avoid FK errors, we already handled its dependencies if db_table == "ProcessDefinition": continue try: cursor.execute(f"TRUNCATE TABLE {db_table}") print(f" [setup] truncated {db_table}") except mysql.connector.Error as e: if self._is_mysql_connection_error(e): raise print(f" [setup] warning: could not truncate {db_table}: {e}") # Finally, truncate ProcessDefinition after all dependencies are truncated if "processdefinition" in table_map and table_map["processdefinition"] == "ProcessDefinition": try: cursor.execute("TRUNCATE TABLE ProcessDefinition") print(f" [setup] truncated ProcessDefinition") except mysql.connector.Error as e: if self._is_mysql_connection_error(e): raise print(f" [setup] warning: could not truncate ProcessDefinition: {e}") conn.commit() # Batch commit: one commit for all TRUNCATE statements print("[setup] Seed data truncation complete.") # Insert seed data in dependency order (parents first) for seed_key in insert_order: db_table = table_map[seed_key] records = self.seed_data.get(seed_key, []) if not records: continue required_columns = self._get_required_seed_columns(cursor, db_table) inserted = 0 for index, record in enumerate(records): record_with_defaults = self._apply_seed_defaults(record, required_columns) # Persist generated defaults like Id so later {seed_*} references stay valid. self.seed_data[seed_key][index] = record_with_defaults # Build INSERT: skip fields that are "NULL" string or empty string # Map seed field name to DB column name (Title case, e.g. lastName → LastName) fields = [] values = [] for k, v in record_with_defaults.items(): if v == "NULL" or v == "": continue # Convert lastName → LastName (Title case the first letter) col = k[0].upper() + k[1:] fields.append(f"`{col}`") values.append(v) if not fields: continue cols_str = ", ".join(fields) placeholders = ", ".join(["%s"] * len(values)) sql = f"INSERT IGNORE INTO {db_table} ({cols_str}) VALUES ({placeholders})" try: cursor.execute(sql, values) inserted += 1 except mysql.connector.Error as e: if self._is_mysql_connection_error(e): raise print(f" [setup] warning: could not insert into {db_table}: {e}") print(f" [setup] inserted {inserted}/{len(records)} rows into {db_table}") # Batch commit all INSERT statements once at the end conn.commit() print("[setup] Seed data inserted successfully.") finally: try: cursor.close() except Exception: pass self._discard_db_connection() def _setup_admin_user(self): """Initialize admin user and test-gosf-client connected app. This method is always called before tests run, regardless of skip_setup setting. It ensures the admin user exists for API authentication. """ max_attempts = 3 for attempt in range(1, max_attempts + 1): try: self._setup_admin_user_once() return except Exception as e: if not self._is_mysql_connection_error(e) or attempt == max_attempts: raise print(f"[setup] TiDB connection lost during admin setup; retrying {attempt}/{max_attempts}: {e}") self._discard_db_connection() time.sleep(2 * attempt) def _setup_admin_user_once(self): print("[setup] Initializing admin user and connected app...") conn = self._db_connect() cursor = conn.cursor() try: # Check if admin user exists cursor.execute("SELECT Id FROM `User` WHERE Email = 'admin@test.com'") admin_user = cursor.fetchall() if not admin_user: cursor.execute(""" INSERT INTO `User` ( `Id`, `Email`, `Username`, `FirstName`, `LastName`, `Password`, `IsActive`, `IsDeleted`, `CreatedDate`, `LastModifiedDate`, `CreatedById`, `LastModifiedById`, `ProfileId` ) VALUES ( '10000000-0000-0000-0000-000000000001', 'admin@test.com', 'admin', 'System', 'Administrator', '$2a$10$kjUgeoFlBhW3CxluLzIoI.vYtMpSJWJ3yA.M5E8sGAEZxXEleMRuy', 1, 0, NOW(), NOW(), '10000000-0000-0000-0000-000000000001', '10000000-0000-0000-0000-000000000001', 'system_admin' ) """) print(" [setup] inserted admin@test.com user") else: print(" [setup] admin@test.com user already exists") # Keep the OAuth test client in the current system table schema. try: cursor.execute(""" SELECT COUNT(*) as count FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = '_System_ConnectedApp' """) result = cursor.fetchall() if result and result[0][0] > 0: cursor.execute( """ INSERT INTO `_System_ConnectedApp` ( `Id`, `Name`, `ConsumerKey`, `ConsumerSecret`, `IsActive`, `CreatedDate`, `LastModifiedDate` ) VALUES ( %s, %s, %s, %s, 1, NOW(), NOW() ) ON DUPLICATE KEY UPDATE `Name` = VALUES(`Name`), `ConsumerSecret` = VALUES(`ConsumerSecret`), `IsActive` = VALUES(`IsActive`), `LastModifiedDate` = NOW() """, ( E2E_CONNECTED_APP_ID, E2E_CONNECTED_APP_NAME, E2E_CONNECTED_APP_ID, E2E_CONNECTED_APP_SECRET, ), ) print(" [setup] test-gosf-client connected app ready") else: print(" [setup] _System_ConnectedApp table does not exist, skipping") except Exception as e: print(f" [setup] failed to ensure connected app: {e}") raise conn.commit() print("[setup] Admin user and connected app ready.") except Exception as e: print(f" [setup] failed to initialize admin user: {e}") if self._is_mysql_connection_error(e): try: conn.rollback() except Exception: pass raise conn.rollback() raise finally: try: cursor.close() except Exception: pass self._discard_db_connection() def _parse_scene_filters(self, scene: Optional[str]) -> List[str]: """Parse comma-separated scene filters into a normalized list.""" if not scene: return [] return [item.strip().lower() for item in scene.split(",") if item.strip()] def _load_cases(self, scene: str = None, case_id: str = None) -> List[Dict]: """Load test cases from JSON files.""" cases = [] scene_filters = self._parse_scene_filters(scene) if case_id: # Load single case by ID — scan all scene subdirectories # (case ID prefix may differ from directory name, e.g. "val" vs "validation") parts = case_id.split("-") if len(parts) >= 2: for scene_dir in os.listdir(self.cases_dir): case_path = os.path.join(self.cases_dir, scene_dir, f"{case_id}.json") if os.path.exists(case_path): with open(case_path, "r") as f: cases.append(json.load(f)) break # Also try lowercase version of case_id case_path_lower = os.path.join(self.cases_dir, scene_dir, f"{case_id.lower()}.json") if os.path.exists(case_path_lower): with open(case_path_lower, "r") as f: cases.append(json.load(f)) break return cases # Load all cases or filtered by scene pattern = os.path.join(self.cases_dir, "**", "*.json") for case_path in glob.glob(pattern, recursive=True): with open(case_path, "r") as f: case = json.load(f) if scene_filters: # Filter by scene name case_scene = case.get("scene", "").lower() if any( case_scene == scene_filter or case_scene.startswith(scene_filter) for scene_filter in scene_filters ): cases.append(case) else: cases.append(case) # Sort by priority (P0 first) priority_order = {"P0": 0, "P1": 1, "P2": 2} cases.sort(key=lambda c: priority_order.get(c.get("priority", "P2"), 2)) return cases def _execute_http_request( self, method: str, endpoint: str, payload: Any, timeout_seconds: Optional[int] = None, auth: Optional[Dict[str, Any]] = None, ) -> tuple: """Execute a raw HTTP request against the NexusCRM API.""" self._assert_time_budgets(f"http:{method}:{endpoint}") requested_timeout = 30 if timeout_seconds is None else int(timeout_seconds) timeout = self._bounded_request_timeout(requested_timeout) self._last_http_request = { "method": method, "endpoint": endpoint, "timeout_seconds": timeout, } client = self._get_http_client(auth) if method == "GET": return client.get(endpoint, timeout_seconds=timeout) if method == "POST": return client.post(endpoint, payload, timeout_seconds=timeout) if method == "PATCH": return client.patch(endpoint, payload, timeout_seconds=timeout) if method == "DELETE": return client.delete(endpoint, timeout_seconds=timeout) return 400, {"error": f"Unsupported method: {method}"} def _resolve_auth_value(self, value: Any) -> str: """Resolve auth values from literals, runtime variables, or env refs.""" if value is None: return "" substituted = self._substitute_variables(value) if not isinstance(substituted, str): return str(substituted) if substituted.startswith("$"): return os.environ.get(substituted[1:], "") return substituted def _get_http_client(self, auth: Optional[Dict[str, Any]] = None): """Return the default client or a cached client for step-level auth.""" if not auth: return self.client username = self._resolve_auth_value(auth.get("username") or auth.get("user")) password = self._resolve_auth_value(auth.get("password")) if not username or not password: raise ValueError("HTTP auth requires both username and password") cache_key = (username, password) if cache_key not in self._client_cache: self._client_cache[cache_key] = api.NexusAPIClient( username=username, password=password, ) return self._client_cache[cache_key] def _execute_request(self, case: Dict) -> tuple: """Execute the HTTP request for a test case.""" method = case.get("method", "GET").upper() endpoint = self._substitute_variables(case.get("endpoint", "")) payload = self._substitute_variables(case.get("payload")) timeout_seconds = case.get("timeout_seconds") auth = case.get("auth") if self._has_unresolved_placeholders(endpoint) or self._has_unresolved_placeholders(payload): return 400, {"error": "Unresolved placeholder(s) in request"} return self._execute_http_request( method, endpoint, payload, timeout_seconds=timeout_seconds, auth=auth, ) def _store_created_id( self, created_id: Optional[str], object_name: str = "", case_id: str = "", id_var: str = "", set_default_id: bool = False, ): """Store created record IDs into runtime variables for later substitution.""" if not created_id: return if id_var: self.variables[id_var] = created_id if object_name: self.variables[f"{object_name}_id"] = created_id if case_id: self.variables[f"{case_id}_id"] = created_id if set_default_id: self.variables["id"] = created_id # Matches "INSERT INTO (Id, ...)" — not "INSERT IGNORE", not # "INSERT ... (X, Id, ...)". Captures the table name (with or without # backticks) so we can emit a targeted pre-DELETE for idempotent seed. _INSERT_WITH_ID_PK = re.compile( r"^\s*INSERT\s+INTO\s+(?P
`[^`]+`|\w+)\s*\(\s*(?:Id|id)\b", re.IGNORECASE, ) @classmethod def _extract_insert_id_target(cls, sql: str) -> Optional[str]: """Return the table name if sql is an `INSERT INTO (Id, ...)` form. Returns None for `INSERT IGNORE`, for INSERTs whose first column is not `Id`, and for non-INSERT statements. The caller uses this to decide whether a `DELETE FROM WHERE Id = %s` pre-clean is safe. """ match = cls._INSERT_WITH_ID_PK.match(sql or "") return match.group("table") if match else None def _execute_sql_step(self, step: Dict[str, Any], commit: bool = True) -> tuple: """Execute a SQL setup/teardown step using the runner's TiDB connection. Setup INSERTs whose first column is `Id` are made idempotent by auto-prepending a `DELETE FROM WHERE Id = ` inside the same connection. Prevents 1062 PK collisions when a prior run's teardown did not execute (process killed / exception mid-case) on the shared ci-test TiDB. Args: step: Step definition dict with 'sql' and optional 'params'. commit: If True, commit the transaction after executing. Callers that want batch commit should pass False and commit once at the end of a multi-step sequence. """ self._assert_time_budgets("sql") sql = self._substitute_variables(step.get("sql", "")) params = self._substitute_variables(step.get("params", [])) if self._has_unresolved_placeholders(sql) or self._has_unresolved_placeholders(params): print(f" [DEBUG] Unresolved placeholder in SQL step: sql={sql}, params={params}") print(f" [DEBUG] Current variables: {list(self.variables.keys())}") raise ValueError("Unresolved placeholder(s) in SQL step") statement = sql.strip() is_select = statement.upper().startswith("SELECT") conn = self._db_connect() cursor = conn.cursor(dictionary=is_select) try: pre_clean_table = self._extract_insert_id_target(statement) if pre_clean_table and params: cursor.execute( f"DELETE FROM {pre_clean_table} WHERE Id = %s", (params[0],), ) cursor.execute(statement, params) if is_select: fetch = (step.get("fetch") or "all").lower() if fetch == "one": response = cursor.fetchone() or {} else: response = cursor.fetchall() else: if commit: conn.commit() response = {"rows_affected": cursor.rowcount} finally: cursor.close() # For non-SELECTs with commit=False, we intentionally skip commit here # so the caller can batch multiple SQL steps before a single commit. # For non-SELECTs with commit=True, conn.commit() was already called above # (making in_transaction=False), so this is a no-op. if not is_select and not commit and conn.is_connected(): pass # No auto-commit — let caller batch if step.get("extract"): self._extract_step_variables(step["extract"], response) return 200, response def _execute_step(self, step: Dict[str, Any], phase: str) -> tuple: """Execute a setup/teardown step and optionally capture variables.""" step_type = (step.get("type") or "http").lower() step_name = step.get("name") or step.get("endpoint") or step.get("sql") or phase if step_type == "sql": status_code, response = self._execute_sql_step(step) # For SQL steps with id_var, extract ID and store it id_var = step.get("id_var") if id_var and response and isinstance(response, dict): created_id = None # First check for explicit id_value in the step (for idempotent inserts) if step.get("id_value"): created_id = step.get("id_value") # For SELECT queries, extract from response elif isinstance(response, list) and response: created_id = response[0].get("id") or response[0].get("Id") elif isinstance(response, dict) and response.get("id") or response.get("Id"): created_id = response.get("id") or response.get("Id") if created_id: self._store_created_id( created_id, object_name=step.get("object", ""), id_var=id_var, ) # Process extract field for SQL steps (e.g., to extract custom fields like csa_rt_id) if step.get("extract"): self._extract_step_variables(step["extract"], response) return True, "OK", { "name": step_name, "type": "sql", "status_code": status_code, "response": response, } method = step.get("method", "GET").upper() endpoint = self._substitute_variables(step.get("endpoint", "")) payload = self._substitute_variables(step.get("payload")) timeout_seconds = step.get("timeout_seconds") auth = step.get("auth") if self._has_unresolved_placeholders(endpoint) or self._has_unresolved_placeholders(payload): raise ValueError(f"Unresolved placeholder(s) in {phase} step: {step_name}") status_code, response = self._execute_http_request( method, endpoint, payload, timeout_seconds=timeout_seconds, auth=auth, ) created_id = self._extract_id_from_response(response) self._store_created_id( created_id, object_name=step.get("object", ""), id_var=step.get("id_var", ""), ) if step.get("extract"): self._extract_step_variables(step["extract"], response) expected = dict(step.get("expects", {})) if "expected_status" in step and "status_code" not in expected: expected["status_code"] = step["expected_status"] assertion_block = step.get("response_assertions") or step.get("assertions") or [] passed, message = self._assert_expected_response( status_code, response, expected=expected, response_assertions=assertion_block, default_expected_status=self._default_expected_status(method), ) if not passed: return False, f"{step_name}: {message}", { "name": step_name, "type": "http", "status_code": status_code, "response": response, } return True, "OK", { "name": step_name, "type": "http", "status_code": status_code, "response": response, } def _execute_setup(self, case: Dict) -> tuple: """Execute case-level setup steps before the main request.""" details = [] for step in self._normalize_steps(case.get("setup")): passed, message, detail = self._execute_step(step, "setup") details.append(detail) if not passed: return False, message, details return True, "OK", details def _response_as_text(self, response: Any) -> str: """Render a response payload as searchable text for substring checks.""" if isinstance(response, str): return response try: return json.dumps(response, ensure_ascii=False, sort_keys=True) except TypeError: return str(response) def _assert_body_contains(self, response: Any, body_contains: Any) -> tuple: """Support dict, list, and string forms for expects.body_contains.""" if isinstance(body_contains, dict): for field, value in body_contains.items(): actual_value = self._resolve_path(response, field) if not self._values_equal(actual_value, self._substitute_variables(value)): return False, f"Field {field} = {actual_value}, expected {value}" return True, "OK" haystack = self._response_as_text(response) snippets = body_contains if isinstance(body_contains, list) else [body_contains] for snippet in snippets: resolved = str(self._substitute_variables(snippet)) if resolved not in haystack: return False, f"Response does not contain {resolved}" return True, "OK" def _assert_expected_response( self, status_code: int, response: Any, expected: Optional[Dict[str, Any]] = None, response_assertions: Optional[List[Dict[str, Any]]] = None, default_expected_status: Optional[int] = None, ) -> tuple: """Shared HTTP response assertion helper for case and step execution.""" expected = expected or {} response_assertions = response_assertions or [] expected_status = expected.get("status_code") if expected_status is None: expected_status = default_expected_status if expected_status is not None and status_code != expected_status: return False, f"Status code {status_code} != expected {expected_status}" if "body_contains" in expected: passed, message = self._assert_body_contains(response, expected["body_contains"]) if not passed: return False, message if response_assertions: passed, message = self._evaluate_assertions(response, response_assertions) if not passed: return False, message return True, "OK" def _assert_response(self, case: Dict, status_code: int, response: Any) -> tuple: """Assert response matches expectations. Returns (passed, message).""" return self._assert_expected_response( status_code, response, expected=case.get("expects", {}), response_assertions=case.get("response_assertions", []), ) def _build_request_steps(self, case: Dict) -> List[Dict[str, Any]]: """Normalize case-level request execution into a sequential step list.""" if case.get("request_steps"): return self._normalize_steps(case.get("request_steps")) main_step = { "name": case.get("name") or case.get("id") or "request", "type": "http", "method": case.get("method", "GET"), "endpoint": case.get("endpoint", ""), "payload": case.get("payload"), "timeout_seconds": case.get("timeout_seconds"), "object": case.get("object", ""), "extract": case.get("extract"), "expects": case.get("expects", {}), "response_assertions": case.get("response_assertions", []), "auth": case.get("auth"), } steps = [main_step] for legacy_key in ("approval_step", "retry_approval"): legacy_step = case.get(legacy_key) if legacy_step: legacy_copy = dict(legacy_step) legacy_copy.setdefault("type", "http") legacy_copy.setdefault("method", "POST") legacy_copy.setdefault("name", legacy_key) steps.append(legacy_copy) return steps def _execute_request_steps(self, case: Dict) -> tuple: """Execute the case's main HTTP phase, including chained follow-up steps.""" details = [] last_status_code = None last_response = None last_created_id = None for step in self._build_request_steps(case): passed, message, detail = self._execute_step(step, "request") details.append(detail) last_status_code = detail["status_code"] last_response = detail["response"] created_id = self._extract_id_from_response(last_response) if created_id: last_created_id = created_id self._store_created_id( created_id, object_name=step.get("object", ""), case_id=case.get("id", ""), set_default_id=True, ) if not passed: return False, message, details, last_status_code, last_response, last_created_id return True, "OK", details, last_status_code, last_response, last_created_id def _build_query_after_order_by(self, sort_definition: Any) -> str: """Build an ORDER BY clause for non-Id query_after fetches.""" if not sort_definition: return "CreatedDate DESC" if isinstance(sort_definition, str): return sort_definition if isinstance(sort_definition, dict): sort_definition = [sort_definition] if not isinstance(sort_definition, list): raise ValueError("query_after.sort must be a string, object, or list") clauses = [] for item in sort_definition: if not isinstance(item, dict): raise ValueError("query_after.sort list entries must be objects") field = item.get("field") if not field: raise ValueError("query_after.sort entries require a field") direction = str(item.get("direction") or "ASC").upper() if direction not in ("ASC", "DESC"): raise ValueError("query_after.sort direction must be ASC or DESC") clauses.append(f"{field} {direction}") if not clauses: raise ValueError("query_after.sort must contain at least one field") return ", ".join(clauses) def _execute_query_after(self, case: Dict, created_id: Optional[str] = None) -> tuple: """Execute query_after assertions. Returns (passed, message, details). Supports optional retry for async-settled state (e.g., AfterSave record-triggered flows that write via Outbox worker): query_after.retry int Max retries after the first attempt (default 0). query_after.delay_ms int Sleep between retries in ms (default 2000, also used when key is present with a null value). query_after.select_fields list Explicit SOQL SELECT fields for non-Id lookups. query_after.sort str|obj|list Explicit ORDER BY for non-Id lookups. query_after.limit int Explicit LIMIT for non-Id lookups (default 1). query_after.response_assertions list Assertions against the raw fetched payload. Retries re-run the full fetch + assertion sequence and pass as soon as any attempt succeeds; the last failure is returned if all retries exhaust. """ query_after = case.get("query_after") if not query_after: return True, "OK", None # Use `is None` guards so explicit `0` keeps its meaning (no delay / # no retry); `or` would coerce 0 to the default and surprise callers. retry_raw = query_after.get("retry") retry = 0 if retry_raw is None else int(retry_raw) delay_raw = query_after.get("delay_ms") delay_ms = 2000 if delay_raw is None else int(delay_raw) last_passed = False last_message = "not executed" last_response = None # retry counts attempts after the first, so total attempts = retry + 1 for attempt in range(retry + 1): self._assert_time_budgets("query_after") last_passed, last_message, last_response = self._query_after_once( query_after, created_id ) if last_passed: return True, "OK", last_response if attempt < retry: self._last_query_after = { "object": query_after.get("object"), "attempt": attempt + 1, "retry": retry, "delay_ms": delay_ms, "message": last_message, } delay_seconds = delay_ms / 1000.0 remaining = self._remaining_case_budget() if remaining is not None and remaining <= delay_seconds: self._assert_case_budget("query_after_sleep") self._assert_run_budget("query_after_sleep") time.sleep(delay_seconds) return False, last_message, last_response def _query_after_once(self, query_after: Dict, created_id: Optional[str] = None) -> tuple: """Single query_after attempt. Extracted so _execute_query_after can retry.""" object_name = query_after.get("object") id_field = query_after.get("id_field", "Id") assertions = query_after.get("assertions", []) response_assertions = query_after.get("response_assertions", []) # Get the record record_id = self._substitute_variables(query_after.get("record_id")) if query_after.get("record_id") else None record_id = record_id or created_id or self.variables.get(f"{object_name}_id") if not record_id: return False, f"No ID found for {object_name}", None if id_field == "Id": endpoint = f"/services/data/v63.0/sobjects/{object_name}/{record_id}" status_code, response = self.client.get(endpoint) if status_code not in (200, 201): return False, f"Failed to query {object_name}: {status_code}", response assertion_target = response else: select_fields = query_after.get("select_fields") if select_fields is None: query_fields = ["Id", id_field] for assertion in assertions: field = assertion.get("field") if field and "." not in field: query_fields.append(field) elif isinstance(select_fields, str): query_fields = ["Id", id_field, select_fields] else: query_fields = ["Id", id_field, *list(select_fields)] unique_fields = [] for field in query_fields: if field not in unique_fields: unique_fields.append(field) order_by = self._build_query_after_order_by(query_after.get("sort")) limit_raw = query_after.get("limit") limit = 1 if limit_raw is None else int(limit_raw) if limit < 1: limit = 1 soql = ( f"SELECT {', '.join(unique_fields)} FROM {object_name} " f"WHERE {id_field}='{record_id}' ORDER BY {order_by} LIMIT {limit}" ) endpoint = f"/services/data/v63.0/query/?q={quote_plus(soql)}" status_code, response = self.client.get(endpoint) if status_code not in (200, 201): return False, f"Failed to query {object_name}: {status_code}", response has_records = ( isinstance(response, dict) and response.get("totalSize", 0) >= 1 and isinstance(response.get("records"), list) and len(response["records"]) >= 1 ) assertion_target = response["records"][0] if has_records else None if response_assertions: passed, message = self._evaluate_assertions(response, response_assertions) if not passed: return False, message, response if assertion_target is None: if not assertions and response_assertions: return True, "OK", response return False, f"No {object_name} records found for {id_field}={record_id}", response passed, message = self._evaluate_assertions(assertion_target, assertions) if not passed: return False, message, response return True, "OK", response def _execute_teardown(self, case: Dict) -> bool: """Execute teardown operations. Always returns True - teardown failures are silently ignored. Teardown failures (SQL errors, connection issues, etc.) are logged as warnings but do not affect the test case result. This is intentional because: 1. Teardown is for cleanup, not validation 2. TiDB Cloud connections can be unstable 3. Test case result should reflect the business logic, not cleanup infrastructure """ steps = self._normalize_steps(case.get("teardown")) if not steps: return True has_sql_teardown = any(step.get("type", "http").lower() == "sql" for step in steps) conn = None if has_sql_teardown: try: conn = self._db_connect() except Exception as e: print(f" Teardown warning: could not open DB connection: {e}") for step in steps: try: self._last_teardown_step = { "type": step.get("type", "http"), "name": step.get("name") or step.get("endpoint") or step.get("sql"), } if step.get("type", "http").lower() == "sql": self._execute_sql_step(step) continue endpoint = self._substitute_variables(step.get("endpoint", "")) payload = self._substitute_variables(step.get("payload")) if self._has_unresolved_placeholders(endpoint) or self._has_unresolved_placeholders(payload): print(f" Teardown warning: Unresolved placeholder(s) in HTTP step") continue method = step.get("method", "DELETE").upper() self._execute_http_request( method, endpoint, payload, timeout_seconds=step.get("timeout_seconds"), ) except Exception as e: print(f" Teardown warning: {e}") # Batch commit all teardown SQL changes once at the end try: if conn is not None and conn.in_transaction: conn.commit() except Exception: pass return True def _extract_id_from_response(self, response: Any) -> Optional[str]: """Extract ID from API response.""" if isinstance(response, dict): return response.get("id") or response.get("Id") if isinstance(response, list) and response and isinstance(response[0], dict): return response[0].get("id") or response[0].get("Id") return None def run_case(self, case: Dict) -> Dict: """Run a single test case. Returns result dict.""" case_id = case.get("id", "unknown") case_name = case.get("name", "") previous_variables = self.variables self.variables = self._build_case_variables(case) self._current_case_id = case_id self._current_scene = case.get("scene", "") self._current_case_name = case_name self._case_deadline = time.monotonic() + self._resolve_case_timeout_seconds(case) self._last_http_request = None self._last_query_after = None self._last_teardown_step = None result = { "id": case_id, "name": case_name, "status": "PASS", "expected": case.get("expects", {}), "actual": {}, "bug_ref": case.get("bug_ref"), "scene": case.get("scene") } try: self._assert_time_budgets("case_start") setup_passed, setup_message, setup_details = self._execute_setup(case) if setup_details: result["actual"]["setup"] = setup_details if not setup_passed: result["status"] = "FAIL" result["error"] = f"setup: {setup_message}" return result request_passed, request_message, request_details, status_code, response, created_id = ( self._execute_request_steps(case) ) if request_details: result["actual"]["requests"] = request_details result["actual"]["status_code"] = status_code result["actual"]["response"] = response if not request_passed: result["status"] = "FAIL" result["error"] = request_message return result # Execute query_after if present query_after = case.get("query_after") if query_after: print(f" [DEBUG] Before query_after, order_id: {self.variables.get('order_id')}") passed, message, query_response = self._execute_query_after(case, created_id) if not passed: result["status"] = "FAIL" result["error"] = f"query_after: {message}" result["actual"]["query_response"] = query_response return result result["actual"]["query_response"] = query_response except CaseTimeoutError as e: result["status"] = "FAIL" result["error"] = str(e) result["actual"]["timeout"] = { "case_id": self._current_case_id, "scene": self._current_scene, "last_http_request": self._last_http_request, "last_query_after": self._last_query_after, "last_teardown_step": self._last_teardown_step, } except Exception as e: result["status"] = "ERROR" result["error"] = str(e) finally: self._execute_teardown(case) self.variables = previous_variables self._case_deadline = None self._current_case_id = "" self._current_scene = "" self._current_case_name = "" return result def _run_scene_cases(self, scene_name: str, scene_cases: List[Dict], case_filter: Optional[str] = None) -> List[Dict]: """Run all cases in one scene sequentially with this runner instance only.""" print(f"[{scene_name}] Starting {len(scene_cases)} case(s)...") scene_results = [] for index, case in enumerate(scene_cases): case_id_val = case.get("id", "unknown") try: self._assert_run_budget(f"scene:{scene_name}:before_case:{case_id_val}") except RunTimeoutError as e: timeout_results = self._build_timeout_results(scene_cases[index:], str(e)) scene_results.extend(timeout_results) print(f"[{scene_name}] STOP: {e}") break print(f" [{case_id_val}] {case.get('name', '')}...") ACTIVE_CASES.start(scene_name, case_id_val, case.get("name", "")) # A direct case filter should override skip flags for focused debugging. should_skip = case.get("skip") and case_filter is None if should_skip: result = {"id": case_id_val, "status": "SKIP", "scene": case.get("scene"), "skip": True} scene_results.append(result) print(" SKIP") ACTIVE_CASES.finish(scene_name, case_id_val) continue try: result = self.run_case(case) scene_results.append(result) if result["status"] == "PASS": print(" PASS") elif result["status"] == "FAIL": result_id = result.get("id", case.get("id", "unknown")) print(f" FAIL [{result_id}]: {result.get('error', 'Unknown')}") else: result_id = result.get("id", case.get("id", "unknown")) print(f" ERROR [{result_id}]: {result.get('error', 'Unknown')}") finally: ACTIVE_CASES.finish(scene_name, case_id_val) passed_in_scene = len([r for r in scene_results if r["status"] == "PASS"]) skipped_in_scene = len([r for r in scene_results if r["status"] == "SKIP"]) print(f"[{scene_name}] Done: {passed_in_scene}/{len(scene_cases)} passed, {skipped_in_scene} skipped") return scene_results def run(self, scene: str = None, case_id: str = None, skip_setup: bool = False, parallel: int = 1, output_format: str = "json") -> Dict: """Run all matching test cases. Returns report dict.""" run_timeout_seconds = self._resolve_run_timeout_seconds() self._run_deadline = None if run_timeout_seconds > 0: self._run_deadline = time.monotonic() + run_timeout_seconds # Always initialize admin user and connected app (needed for API authentication) self._assert_run_budget("run_start") self._setup_admin_user() # If running a single specific case, skip scene-level seed setup and rely on # the case's own inline setup/teardown. A self-contained case needs isolation. effective_skip_setup = skip_setup or (case_id is not None) if not effective_skip_setup: self._setup_seed_data() cases = self._load_cases(scene, case_id) print(f"=== NexusCRM E2E API Tests ===") print(f"Loaded {len(cases)} case(s)") if scene: print(f"Scene filter: {scene}") if case_id: print(f"Case filter: {case_id}") print(f"Parallel workers: {parallel}") if run_timeout_seconds > 0: print(f"Soft run timeout: {run_timeout_seconds}s") print() # Group cases by scene for scene-level execution from collections import defaultdict scenes_to_cases = defaultdict(list) for case in cases: scene_name = case.get("scene", "unknown") scenes_to_cases[scene_name].append(case) passed = 0 failed = 0 errors = 0 skipped = 0 if parallel > 1 and len(scenes_to_cases) > 1: max_workers = min(parallel, len(scenes_to_cases)) scene_queue = list(scenes_to_cases.items()) next_scene_index = 0 print(f"Running {len(scenes_to_cases)} scene(s) in parallel (workers={max_workers})...") with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_scene = {} def submit_until_full() -> None: nonlocal next_scene_index while len(future_to_scene) < max_workers and next_scene_index < len(scene_queue): self._assert_run_budget("scene_submit") scene_name, scene_cases = scene_queue[next_scene_index] next_scene_index += 1 future = executor.submit( _run_scene_worker, scene_name, scene_cases, self.base_path, self.seed_data, case_id, self._run_deadline, ) future_to_scene[future] = scene_name try: submit_until_full() except RunTimeoutError as e: remaining_queue = scene_queue[next_scene_index:] for _, queued_cases in remaining_queue: self.results.extend(self._build_timeout_results(queued_cases, str(e))) errors += len(queued_cases) future_to_scene = {} pending = set(future_to_scene.keys()) heartbeat_seconds = int(os.environ.get("E2E_API_HEARTBEAT_SECONDS", "15")) while pending: done, pending = wait( pending, timeout=heartbeat_seconds, return_when=FIRST_COMPLETED, ) if not done: self._emit_active_case_heartbeat() continue for future in done: scene_name = future_to_scene.pop(future) try: scene_results = future.result() self.results.extend(scene_results) for r in scene_results: if r["status"] == "PASS": passed += 1 elif r["status"] == "FAIL": failed += 1 elif r["status"] == "SKIP": skipped += 1 else: errors += 1 except Exception as e: print(f"[{scene_name}] ERROR: {e}") errors += len(scenes_to_cases[scene_name]) try: submit_until_full() except RunTimeoutError as e: remaining_queue = scene_queue[next_scene_index:] for _, queued_cases in remaining_queue: self.results.extend(self._build_timeout_results(queued_cases, str(e))) errors += len(queued_cases) next_scene_index = len(scene_queue) pending = set(future_to_scene.keys()) else: # Serial execution: scenes run sequentially, cases within scene run sequentially scene_queue = list(scenes_to_cases.items()) for scene_index, (scene_name, scene_cases) in enumerate(scene_queue): try: self._assert_run_budget(f"scene:{scene_name}:start") except RunTimeoutError as e: for _, queued_cases in scene_queue[scene_index:]: timeout_results = self._build_timeout_results(queued_cases, str(e)) self.results.extend(timeout_results) errors += len(timeout_results) break scene_results = self._run_scene_cases(scene_name, scene_cases, case_id) self.results.extend(scene_results) for result in scene_results: if result["status"] == "PASS": passed += 1 elif result["status"] == "FAIL": failed += 1 elif result["status"] == "SKIP": skipped += 1 else: errors += 1 # Generate report total = len(cases) timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") report_path = os.path.join(self.base_path, f"report-{timestamp}.json") # Group by scene coverage = {} for r in self.results: scene_name = r.get("scene", "unknown") if scene_name not in coverage: coverage[scene_name] = {"total": 0, "passed": 0, "failed": 0, "errors": 0} coverage[scene_name]["total"] += 1 if r["status"] == "PASS": coverage[scene_name]["passed"] += 1 elif r["status"] == "FAIL": coverage[scene_name]["failed"] += 1 else: coverage[scene_name]["errors"] += 1 report = { "timestamp": datetime.now().isoformat(), "total": total, "passed": passed, "failed": failed, "errors": errors, "pass_rate": f"{(passed/total*100):.0f}%" if total > 0 else "0%", "coverage": coverage, "results": self.results } with open(report_path, "w") as f: json.dump(report, f, indent=2) print(f"JSON report: {report_path}") # Generate JUnit XML report if requested if output_format in ("junit", "both"): try: from generate_junit import json_to_junit junit_path = json_to_junit(report_path) print(f"JUnit XML report: {junit_path}") except Exception as e: print(f"Warning: Failed to generate JUnit XML: {e}") print() print("=== Summary ===") print(f"Total: {total}, Passed: {passed}, Skipped: {skipped}, Failed: {failed}, Errors: {errors}") print(f"Pass rate: {report['pass_rate']}") self.close() return report def _run_case_worker(case: Dict, base_path: str, seed_data: Dict) -> Dict: """Worker function for parallel case execution. Creates an isolated TestRunner with its own NexusAPIClient to avoid shared auth state between parallel workers. """ runner = TestRunner(base_path=base_path) runner.seed_data = seed_data # Use pre-loaded seed data to avoid re-reading return runner.run_case(case) def _run_scene_worker( scene_name: str, scene_cases: List[Dict], base_path: str, seed_data: Dict, case_filter: Optional[str] = None, run_deadline: Optional[float] = None, ) -> List[Dict]: """Run one scene in an isolated runner to avoid cross-thread shared state.""" runner = TestRunner(base_path=base_path) runner.seed_data = seed_data runner._run_deadline = run_deadline try: return runner._run_scene_cases(scene_name, scene_cases, case_filter) finally: runner.close() def _handle_termination_signal(signum, _frame) -> None: """Print active-case diagnostics before the outer timeout kills the process.""" signal_name = signal.Signals(signum).name snapshot = ACTIVE_CASES.snapshot() if snapshot: rendered = ", ".join( f"{item['scene']}:{item['case_id']}({item['elapsed_seconds']}s)" for item in snapshot ) print(f"[signal] Received {signal_name}; active cases: {rendered}") else: print(f"[signal] Received {signal_name}; no active cases recorded") raise SystemExit(124) def main(): signal.signal(signal.SIGTERM, _handle_termination_signal) signal.signal(signal.SIGINT, _handle_termination_signal) parser = argparse.ArgumentParser(description="NexusCRM E2E API Test Runner") parser.add_argument("--scene", type=str, help="Run only cases in this scene or comma-separated scene list") parser.add_argument("--case", type=str, help="Run only this specific case ID") parser.add_argument("--all", action="store_true", default=True, help="Run all cases (default)") parser.add_argument("--skip-setup", action="store_true", help="Skip seed data setup (use after manual setup)") parser.add_argument( "--parallel", type=int, default=1, help="Number of cases to run in parallel (default: 1, serial)", ) parser.add_argument( "--fail-on-failures", action="store_true", help="Exit with code 1 when any case fails or errors", ) parser.add_argument( "--format", type=str, default="json", choices=["json", "junit", "both"], help="Output format: json, junit, or both (default: json)", ) args = parser.parse_args() runner = TestRunner() report = runner.run( scene=args.scene, case_id=args.case, skip_setup=args.skip_setup, parallel=args.parallel, output_format=args.format, ) # Exit with non-zero if there are failures or errors (only when fail-on-failures is set) if args.fail_on_failures and (report["failed"] > 0 or report["errors"] > 0): sys.exit(1) sys.exit(0) if __name__ == "__main__": main()