diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 748e24f..5a6ae2a 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,3 +1,24 @@ +# Release Notes: FlowRunner CLI (Unreleased) + +## ⚠️ Cross-app schema gate — severity: HIGH — unknown MAJOR `schemaVersion` is now rejected loudly + +**What changed.** The shared `.flow.json` format carries an OPTIONAL top-level `schemaVersion` string `"MAJOR.MINOR"` (absence ⇒ `"1.0"`, HAR `log.version` precedent). The CLI's `FlowMap` parser now **version-gates** on it: + +- **Absent / `"1.0"` / any `"1.x"`** ⇒ accepted and run unchanged. An unknown **MINOR** (e.g. `"1.5"`) is **tolerated with a warning**; any unrecognized construct still degrades gracefully (skip-with-warning) exactly as before. +- **Unknown MAJOR (`>= 2`, e.g. `"2.0"`)** ⇒ **rejected loudly** with a `ValidationError` attributable to `schemaVersion` (naming the offending version), instead of best-effort mis-executing a genuinely newer format against live customer traffic. +- A non-string value (e.g. integer `2`) is **coerced-and-warned** (`2` ⇒ `2.0`), then gated on its MAJOR like any other value — never a silent crash. + +This converts *silent wrong-execution* — the single most damaging failure for a "what you see is what actually ran" demo tool — into a principled, auditable refusal. It is additive and backward-compatible: a golden conformance test (`tests/unit/test_golden_old_flow.py`) proves a real pre-sprint flow parses to an **identical** execution model with no `schemaVersion`, with `"1.0"`, and with an unknown MINOR. See the FlowRunner UI repo's `docs/schema-versioning.md`. + +## ✨ Additive request-step fields honored: `retries` and `assertions` + +- **`step.retries = {count, delayMs}`** (severity: LOW — additive, opt-in). Per-request retry policy mirroring the FlowRunner UI JS engine: an outer retry loop re-issues the whole request on a non-2xx status **or** a network/fetch error, sleeping `delayMs` between attempts and issuing a fresh request each pass. `count` defaults to `0` ⇒ a single attempt, **IDENTICAL** to prior behavior. A user-requested stop is **never** retried. This wraps — and is orthogonal to — the built-in connection/5xx resilience loop. +- **`step.assertions[]`** (severity: LOW — additive, diagnostic-only). Declarative assertions evaluated against the response after each request, **reusing the frozen `conditionData` operator vocabulary** (same operators, same coercion, same missing-target handling). Results are recorded into the execution context under `response__assertions` (a per-assertion `{name, variable, operator, value, passed}` list) and `response__assertions_passed` (aggregate boolean). Assertions are diagnostic: they **never** change flow control and **never** crash. Unknown operators / missing targets degrade to a FAILED assertion with a warning. + +Both fields are ignored by older CLIs (`extra='ignore'`), so files that use them still run everywhere. This is part of the cross-app FlowMap additive-evolution strategy (see the FlowRunner UI repo's `docs/flowmap-evolution.md`). + +--- + # Release Notes: FlowRunner CLI v1.2.0 ## Highlights diff --git a/flow_runner.py b/flow_runner.py index 6fc6944..db3ad34 100644 --- a/flow_runner.py +++ b/flow_runner.py @@ -79,6 +79,20 @@ class BaseStep(BaseModel): name: Optional[str] = Field(None, description="Human-readable name for the step") # type will be defined in subclasses using Literal +class RetryConfig(BaseModel): + """Additive per-request retry policy. Mirrors the FlowRunner UI JS engine. + + ``count`` = number of RETRIES after the first attempt (0 => single attempt, + IDENTICAL to prior behavior). ``delayMs`` = fixed backoff slept between + attempts. A retry fires on a non-2xx HTTP status OR a network/fetch error, + but NEVER after a user-requested stop. + """ + count: int = Field(0, ge=0, description="Number of retries after the first attempt (default 0).") + delayMs: int = Field(0, ge=0, description="Delay in milliseconds slept between attempts.") + + model_config = ConfigDict(extra="ignore") + + class RequestStep(BaseStep): type: Literal['request'] = Field(..., description="Specifies the step type as 'request'") method: str = Field(..., description="HTTP method (GET, POST, PUT, etc.)") @@ -91,6 +105,14 @@ class RequestStep(BaseStep): body: Optional[Union[Dict[str, Any], str]] = Field(None, description="Request body (JSON object or raw string). Can contain {{variables}}.") extract: Optional[Dict[str, str]] = Field(default_factory=dict, description="Mapping of variable names to extract from response using path notation (e.g., 'token': 'body.data.sessionToken', 'firstId': 'body.data.items[0].id', 'status_code': '.status', 'header_val': 'headers.Content-Type')") # Updated description with prefixes onFailure: Literal['stop', 'continue'] = Field(..., description="Action on request failure (status >= 300): 'stop' or 'continue'.") # Added onFailure field + retries: Optional[RetryConfig] = Field( + None, + description="Additive per-request retry policy {count, delayMs}. Absent/None => single attempt.", + ) + assertions: Optional[List["Assertion"]] = Field( + None, + description="Additive declarative assertions evaluated against the response using the conditionData operator vocabulary.", + ) @field_validator('method') def validate_method(cls, v): @@ -108,6 +130,24 @@ class ConditionData(BaseModel): operator: str = Field("", description="Operation to perform (e.g., 'equals', 'exists', 'is_number', 'greater_than')") value: Optional[str] = Field("", description="Value to compare against (for operators that need it)") +class Assertion(BaseModel): + """Additive declarative assertion on a request step's result. + + Reuses the frozen ``conditionData`` operator vocabulary (see + ``_evaluate_structured_condition``) so authoring stays consistent across the + UI, CLI, and portal. ``variable`` is a context path evaluated AFTER the + request completes (e.g. ``response__status``, ``body.data.ok``, or any + extracted variable). An unknown operator or missing target degrades to a + failed assertion with a warning — it never crashes the run. + """ + name: Optional[str] = Field(None, description="Human-readable label for the assertion (optional).") + variable: str = Field("", description="Context path to evaluate (e.g. 'response_s1_status', 'body.data.id').") + operator: str = Field("", description="Operator from the conditionData vocabulary (e.g. 'equals', 'exists', 'greater_than').") + value: Optional[str] = Field("", description="Comparison value for operators that need it.") + + model_config = ConfigDict(extra="ignore") + + class ConditionStep(BaseStep): type: Literal['condition'] = Field(..., description="Specifies the step type as 'condition'") condition: Optional[str] = Field(None, description="DEPRECATED/LEGACY: Original JavaScript-like condition string. Use conditionData instead.") @@ -163,10 +203,63 @@ class TransformStep(BaseStep): ] # Update nested references in ConditionStep and LoopStep +RequestStep.model_rebuild() # resolves the forward ref to Assertion ConditionStep.model_rebuild() LoopStep.model_rebuild() TransformStep.model_rebuild() +# --- Cross-app schemaVersion gate --------------------------------------- +# The shared .flow.json format carries an OPTIONAL top-level ``schemaVersion`` +# string "MAJOR.MINOR" (e.g. "1.0"). ABSENCE means "1.0" (HAR log.version +# precedent, zero-migration). Contract (docs/schema-versioning.md in the +# FlowRunner UI repo): +# * Unknown MINOR within the supported MAJOR -> TOLERATE (accept, warn once, +# let per-construct graceful degradation handle anything newer). +# * Unknown MAJOR (>= 2) -> REJECT LOUDLY, so a 24/7 +# container never best-effort mis-executes a genuinely newer format. +# Never make schemaVersion required and never gate the wire format on a MINOR. +SUPPORTED_SCHEMA_MAJOR = 1 +_SCHEMA_VERSION_RE = re.compile(r"^(\d+)\.(\d+)$") + + +def parse_schema_version(raw: Any) -> "tuple[int, int]": + """Coerce a raw ``schemaVersion`` value to a ``(major, minor)`` tuple. + + Accepts a "MAJOR.MINOR" string, a bare integer (coerced to ``.0`` with + a warning), ``None`` (=> ``(1, 0)``), or a bare "MAJOR" string. Raises + ``ValueError`` for anything unparseable so the caller can reject it cleanly. + """ + if raw is None or raw == "": + return (SUPPORTED_SCHEMA_MAJOR, 0) + + # Non-string (e.g. integer 2 / float 1.0): coerce-and-warn, never crash. + if isinstance(raw, bool): # guard: bools are ints in Python + raise ValueError(f"schemaVersion must be a 'MAJOR.MINOR' string, got boolean {raw!r}") + if isinstance(raw, int): + logger.warning( + f"schemaVersion was provided as integer {raw!r}; coercing to string '{raw}.0'. " + "The canonical form is a 'MAJOR.MINOR' string (e.g. \"1.0\")." + ) + return (raw, 0) + if isinstance(raw, float): + # e.g. 1.0 -> (1, 0); avoid float precision surprises via string form. + raw = repr(raw) + + if isinstance(raw, str): + text = raw.strip() + m = _SCHEMA_VERSION_RE.match(text) + if m: + return (int(m.group(1)), int(m.group(2))) + # Tolerate a bare "MAJOR" (e.g. "2") as MAJOR.0. + if text.isdigit(): + return (int(text), 0) + raise ValueError( + f"schemaVersion '{raw}' is not a valid 'MAJOR.MINOR' version string." + ) + + raise ValueError(f"schemaVersion has unsupported type {type(raw).__name__}: {raw!r}") + + class FlowMap(BaseModel): id: Optional[str | int] = Field( None, @@ -180,10 +273,49 @@ class FlowMap(BaseModel): headers: Optional[Dict[str, str]] = Field(default_factory=dict, description="Global headers applied to all requests in the flow. Can contain {{variables}}.") steps: List[FlowStep] = Field(..., description="The sequence of steps defining the flow") staticVars: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Global static variables accessible anywhere in the flow (referenced as {{varName}}). Values can be strings, numbers, booleans.") # Allow Any type + schemaVersion: Optional[Any] = Field( + None, + description=( + "OPTIONAL cross-app schema version 'MAJOR.MINOR' (absence => '1.0'). " + "Unknown MINOR is tolerated; unknown MAJOR (>= 2) is rejected." + ), + ) - # Ignore any extra fields when parsing flow definitions + # Ignore any OTHER extra fields when parsing flow definitions (additive-safe). model_config = ConfigDict(extra="ignore") + @field_validator("schemaVersion") + @classmethod + def _gate_schema_version(cls, v: Any) -> Any: + """Version-gate: reject unknown MAJOR loudly, tolerate unknown MINOR.""" + try: + major, minor = parse_schema_version(v) + except ValueError as e: + raise ValueError(str(e)) + + if major > SUPPORTED_SCHEMA_MAJOR: + raise ValueError( + f"Unsupported schemaVersion '{v}': this FlowRunner CLI supports " + f"schema MAJOR {SUPPORTED_SCHEMA_MAJOR}.x but the flow requires " + f"MAJOR {major}. Refusing to run rather than mis-execute a newer " + "format. Upgrade the CLI (see docs/schema-versioning.md)." + ) + if major < SUPPORTED_SCHEMA_MAJOR: + # A lower MAJOR than we support should not happen (1 is the floor), + # but tolerate it: the format only grew additively above it. + logger.warning( + f"schemaVersion '{v}' has a MAJOR below the supported " + f"{SUPPORTED_SCHEMA_MAJOR}; treating as compatible." + ) + elif minor > 0: + # Same MAJOR, newer MINOR: additive; tolerate + warn once. + logger.warning( + f"schemaVersion '{v}' has an unknown MINOR (supported MAJOR is " + f"{SUPPORTED_SCHEMA_MAJOR}). Proceeding; any unrecognized " + "constructs will degrade gracefully (skip-with-warning)." + ) + return v + # Ensure FlowMap uses the updated FlowStep FlowMap.model_rebuild() @@ -2015,6 +2147,102 @@ def _evaluate_structured_condition(self, condition_data: ConditionData, context: return False + # Operators recognized by _evaluate_structured_condition. Kept in sync with + # that method so assertions can warn distinctly on an unknown operator + # (degrade-with-warning) rather than silently treating it as a plain failure. + _KNOWN_ASSERTION_OPERATORS = frozenset({ + 'exists', 'not_exists', + 'is_number', 'is_text', 'is_boolean', 'is_array', + 'is_true', 'is_false', + 'equals', 'not_equals', + 'greater_than', 'less_than', 'greater_equals', 'less_equals', + 'contains', 'starts_with', 'ends_with', 'matches_regex', + }) + + def _evaluate_assertions( + self, + step_id: str, + step_identifier: str, + assertions: Optional[List["Assertion"]], + context: Dict[str, Any], + ) -> None: + """Evaluate a request step's declarative assertions after it completes. + + Reuses the frozen conditionData operator vocabulary via + ``_evaluate_structured_condition``. Records a structured list under + ``response__assertions`` and an aggregate boolean under + ``response__assertions_passed``. Unknown operators / missing targets + degrade to a FAILED assertion with a warning; nothing here ever raises. + """ + if not assertions: + return + + results: List[Dict[str, Any]] = [] + all_passed = True + + for index, assertion in enumerate(assertions): + variable = (getattr(assertion, 'variable', '') or '').strip() + operator = (getattr(assertion, 'operator', '') or '').strip() + value = getattr(assertion, 'value', '') + label = getattr(assertion, 'name', None) or f"assertion[{index}]" + passed = False + note = None + + try: + if not variable or not operator: + note = "missing variable or operator" + logger.warning( + f"Step {step_identifier}: assertion '{label}' is missing a " + f"variable or operator; recording as failed (degrade)." + ) + elif operator not in self._KNOWN_ASSERTION_OPERATORS: + note = f"unknown operator '{operator}'" + logger.warning( + f"Step {step_identifier}: assertion '{label}' uses unknown " + f"operator '{operator}'; recording as failed (degrade), run continues." + ) + else: + # Reuse the exact structured-condition evaluator (same vocab, + # same coercion, same missing-target => None handling). + cond = ConditionData(variable=variable, operator=operator, value=value) + passed = bool(self._evaluate_structured_condition(cond, context)) + except Exception as e: # defense-in-depth: never let an assertion crash the run + note = f"evaluation error: {e}" + logger.warning( + f"Step {step_identifier}: assertion '{label}' raised during " + f"evaluation ({e}); recording as failed (degrade)." + ) + passed = False + + all_passed = all_passed and passed + record = { + "name": getattr(assertion, 'name', None), + "variable": variable, + "operator": operator, + "value": value, + "passed": passed, + } + if note: + record["note"] = note + results.append(record) + + log_fn = logger.info if passed else logger.warning + log_fn( + f"Step {step_identifier}: assertion '{label}' " + f"({variable} {operator} {value!r}) => {'PASS' if passed else 'FAIL'}" + + (f" [{note}]" if note else "") + ) + + set_value_in_context(context, f'response_{step_id}_assertions', results) + set_value_in_context(context, f'response_{step_id}_assertions_passed', all_passed) + + if not all_passed: + failed = sum(1 for r in results if not r["passed"]) + logger.warning( + f"Step {step_identifier}: {failed}/{len(results)} assertion(s) FAILED." + ) + + def _evaluate_condition(self, condition_str: Optional[str], context: Dict[str, Any], condition_data: Optional[ConditionData] = None) -> bool: """ Evaluates a condition. Prefers structured data (conditionData) if available and valid, @@ -2557,106 +2785,149 @@ async def _execute_request_step( max_retries = 3 # Retries for connection errors or 5xx server errors base_retry_delay = 0.5 # seconds - for attempt in range(max_retries): - request_start_time = time.monotonic() - try: - async with session.request( - method, - final_url, - headers=final_headers, - json=json_payload, - data=data_payload - # Note: ssl handling is done via connector settings - ) as resp: - # --- Process Response --- - response_status = resp.status - # Convert response headers (CIMultiDict) to a simple dict for context storage - # Handle multiple Set-Cookie headers if needed later, for now just last value. - response_headers_dict = {k: v for k, v in resp.headers.items()} - request_duration_s = time.monotonic() - request_start_time - request_succeeded = True # Mark that we got a response - - # --- Read Response Body --- - response_body = None # Reset for this attempt - try: - resp_content_type = resp.headers.get('Content-Type', '').lower() - if 'application/json' in resp_content_type: - try: response_body = await resp.json(encoding='utf-8') - except (json.JSONDecodeError, UnicodeDecodeError, aiohttp.ContentTypeError) as json_err: - logger.warning(f"Step {step_identifier}: Failed to decode JSON response ({resp.status}) despite Content-Type. Error: {json_err}. Reading as text.") - # Fallback: read as text + # --- ADDITIVE: user-configured per-request retry policy (step.retries) --- + # Mirrors the FlowRunner UI JS engine: an outer retry loop that re-issues + # the WHOLE request on a non-2xx status OR a network/fetch error. count + # defaults to 0 => a single pass, IDENTICAL to prior behavior. delayMs is + # slept between passes. A user-requested stop (self.running == False) is + # NEVER retried. Each pass issues a fresh request (the CLI analogue of a + # fresh AbortController per attempt). This wraps — and is orthogonal to — + # the built-in connection/5xx resilience loop below (max_retries). + retry_cfg = getattr(step, 'retries', None) + user_retry_count = max(0, retry_cfg.count) if retry_cfg else 0 + user_retry_delay_s = (max(0, retry_cfg.delayMs) / 1000.0) if retry_cfg else 0.0 + + for user_attempt in range(user_retry_count + 1): + user_attempts_remaining = user_retry_count - user_attempt + + # Reset per-pass outcome so a prior pass's error/state never leaks. + response_body = None + response_headers_dict = {} + response_status = -1 + error_message = None + request_succeeded = False + + for attempt in range(max_retries): + request_start_time = time.monotonic() + try: + async with session.request( + method, + final_url, + headers=final_headers, + json=json_payload, + data=data_payload + # Note: ssl handling is done via connector settings + ) as resp: + # --- Process Response --- + response_status = resp.status + # Convert response headers (CIMultiDict) to a simple dict for context storage + # Handle multiple Set-Cookie headers if needed later, for now just last value. + response_headers_dict = {k: v for k, v in resp.headers.items()} + request_duration_s = time.monotonic() - request_start_time + request_succeeded = True # Mark that we got a response + + # --- Read Response Body --- + response_body = None # Reset for this attempt + try: + resp_content_type = resp.headers.get('Content-Type', '').lower() + if 'application/json' in resp_content_type: + try: response_body = await resp.json(encoding='utf-8') + except (json.JSONDecodeError, UnicodeDecodeError, aiohttp.ContentTypeError) as json_err: + logger.warning(f"Step {step_identifier}: Failed to decode JSON response ({resp.status}) despite Content-Type. Error: {json_err}. Reading as text.") + # Fallback: read as text + response_body = await resp.text(encoding='utf-8', errors='replace') + elif resp_content_type.startswith('text/'): response_body = await resp.text(encoding='utf-8', errors='replace') - elif resp_content_type.startswith('text/'): - response_body = await resp.text(encoding='utf-8', errors='replace') - else: - # Read non-text types as bytes, store placeholder - raw_bytes = await resp.read() - limit = 100 - if len(raw_bytes) > limit: response_body = f"[Body Binary Data - Type: {resp_content_type}, Size: {len(raw_bytes)} bytes, Starts: {raw_bytes[:limit]!r}...]" - else: response_body = f"[Body Binary Data - Type: {resp_content_type}, Size: {len(raw_bytes)} bytes, Data: {raw_bytes!r}]" - logger.debug(f"Step {step_identifier}: Read {len(raw_bytes)} bytes for Content-Type: {resp_content_type}") - - except aiohttp.ClientPayloadError as payload_err: - logger.error(f"Step {step_identifier}: Payload error reading response body ({resp.status}): {payload_err}") - response_body = f"Error reading response body: {payload_err}" - except Exception as body_err: - logger.error(f"Step {step_identifier}: Generic error reading response body ({resp.status}): {body_err}", exc_info=self.config.debug) - response_body = f"Generic error reading response body: {body_err}" - - - # --- Log Response --- - log_level = logging.WARNING if response_status >= 400 else logging.INFO - logger.log(log_level, f"Step {step_identifier} received: {response_status} {method} {final_url} ({request_duration_s*1000:.2f} ms)") - - if logger.isEnabledFor(logging.DEBUG): - log_body_repr = repr(response_body) - log_body_display = f"{log_body_repr[:250]}{'...' if len(log_body_repr) > 250 else ''}" - log_resp_headers = {k: ('********' if k.lower() == 'set-cookie' and v else v) for k, v in response_headers_dict.items()} - logger.debug(f" Response Headers: {log_resp_headers}") - logger.debug(f" Response Body ({type(response_body).__name__}): {log_body_display}") - - # --- Retry Logic (Retry on 5xx server errors) --- - if response_status >= 500 and attempt < max_retries - 1: - retry_delay = base_retry_delay * (2 ** attempt) # Exponential backoff - logger.warning(f"Step {step_identifier}: Server error {response_status} on attempt {attempt+1}/{max_retries}. Retrying in {retry_delay:.2f}s...") - await asyncio.sleep(retry_delay) - request_succeeded = False # Reset success flag for retry - continue # Go to next attempt - - # If not retrying (success, 4xx, or 5xx on last attempt), break the loop - break + else: + # Read non-text types as bytes, store placeholder + raw_bytes = await resp.read() + limit = 100 + if len(raw_bytes) > limit: response_body = f"[Body Binary Data - Type: {resp_content_type}, Size: {len(raw_bytes)} bytes, Starts: {raw_bytes[:limit]!r}...]" + else: response_body = f"[Body Binary Data - Type: {resp_content_type}, Size: {len(raw_bytes)} bytes, Data: {raw_bytes!r}]" + logger.debug(f"Step {step_identifier}: Read {len(raw_bytes)} bytes for Content-Type: {resp_content_type}") + + except aiohttp.ClientPayloadError as payload_err: + logger.error(f"Step {step_identifier}: Payload error reading response body ({resp.status}): {payload_err}") + response_body = f"Error reading response body: {payload_err}" + except Exception as body_err: + logger.error(f"Step {step_identifier}: Generic error reading response body ({resp.status}): {body_err}", exc_info=self.config.debug) + response_body = f"Generic error reading response body: {body_err}" + + + # --- Log Response --- + log_level = logging.WARNING if response_status >= 400 else logging.INFO + logger.log(log_level, f"Step {step_identifier} received: {response_status} {method} {final_url} ({request_duration_s*1000:.2f} ms)") + + if logger.isEnabledFor(logging.DEBUG): + log_body_repr = repr(response_body) + log_body_display = f"{log_body_repr[:250]}{'...' if len(log_body_repr) > 250 else ''}" + log_resp_headers = {k: ('********' if k.lower() == 'set-cookie' and v else v) for k, v in response_headers_dict.items()} + logger.debug(f" Response Headers: {log_resp_headers}") + logger.debug(f" Response Body ({type(response_body).__name__}): {log_body_display}") + + # --- Retry Logic (Retry on 5xx server errors) --- + if response_status >= 500 and attempt < max_retries - 1: + retry_delay = base_retry_delay * (2 ** attempt) # Exponential backoff + logger.warning(f"Step {step_identifier}: Server error {response_status} on attempt {attempt+1}/{max_retries}. Retrying in {retry_delay:.2f}s...") + await asyncio.sleep(retry_delay) + request_succeeded = False # Reset success flag for retry + continue # Go to next attempt + + # If not retrying (success, 4xx, or 5xx on last attempt), break the loop + break - # --- Handle Connection/Timeout Errors --- - except (aiohttp.ClientConnectionError, aiohttp.ClientConnectorError, asyncio.TimeoutError) as conn_err: - request_duration_s = time.monotonic() - request_start_time - logger.warning(f"Step {step_identifier}: Attempt {attempt+1}/{max_retries} failed: {type(conn_err).__name__}: {conn_err} ({request_duration_s*1000:.2f} ms)") - if attempt < max_retries - 1: - retry_delay = base_retry_delay * (2 ** attempt) - logger.warning(f"Step {step_identifier}: Retrying connection after {retry_delay:.2f}s...") - await asyncio.sleep(retry_delay) - continue # Go to next attempt - else: - # Max retries reached for connection error - error_message = f"Connection/Timeout Error after {max_retries} attempts: {conn_err}" - logger.error(f"Step {step_identifier}: {error_message}") - response_status = 598 # Custom status for connection errors + # --- Handle Connection/Timeout Errors --- + except (aiohttp.ClientConnectionError, aiohttp.ClientConnectorError, asyncio.TimeoutError) as conn_err: + request_duration_s = time.monotonic() - request_start_time + logger.warning(f"Step {step_identifier}: Attempt {attempt+1}/{max_retries} failed: {type(conn_err).__name__}: {conn_err} ({request_duration_s*1000:.2f} ms)") + if attempt < max_retries - 1: + retry_delay = base_retry_delay * (2 ** attempt) + logger.warning(f"Step {step_identifier}: Retrying connection after {retry_delay:.2f}s...") + await asyncio.sleep(retry_delay) + continue # Go to next attempt + else: + # Max retries reached for connection error + error_message = f"Connection/Timeout Error after {max_retries} attempts: {conn_err}" + logger.error(f"Step {step_identifier}: {error_message}") + response_status = 598 # Custom status for connection errors + break # Exit retry loop + + # --- Handle Other Client Errors --- + except aiohttp.ClientError as client_err: + request_duration_s = time.monotonic() - request_start_time + error_message = f"HTTP Client Error: {client_err}" + logger.error(f"Step {step_identifier}: {error_message} ({request_duration_s*1000:.2f} ms)", exc_info=self.config.debug) + response_status = 597 # Custom status for other client errors + break # Exit retry loop (usually not retriable) + + # --- Handle Unexpected Errors --- + except Exception as e: + request_duration_s = time.monotonic() - request_start_time + error_message = f"Unexpected error during request execution: {e}" + logger.error(f"Step {step_identifier}: {error_message} ({request_duration_s*1000:.2f} ms)", exc_info=self.config.debug) + response_status = 596 # Custom code for unexpected errors break # Exit retry loop - # --- Handle Other Client Errors --- - except aiohttp.ClientError as client_err: - request_duration_s = time.monotonic() - request_start_time - error_message = f"HTTP Client Error: {client_err}" - logger.error(f"Step {step_identifier}: {error_message} ({request_duration_s*1000:.2f} ms)", exc_info=self.config.debug) - response_status = 597 # Custom status for other client errors - break # Exit retry loop (usually not retriable) + # --- ADDITIVE: user-retry decision (step.retries) --- + # A pass "failed" if the request never completed (network/timeout/ + # unexpected) OR it completed with a non-2xx status. Retry only while + # user attempts remain AND the user has not requested a stop. + pass_failed = (not request_succeeded) or (response_status < 200) or (response_status >= 300) + if pass_failed and user_attempts_remaining > 0 and getattr(self, 'running', True): + logger.warning( + f"Step {step_identifier}: retry policy re-attempting after " + f"{'status ' + str(response_status) if request_succeeded else (error_message or 'network error')} " + f"(retry {user_attempt + 1}/{user_retry_count})." + ) + if user_retry_delay_s > 0: + await asyncio.sleep(user_retry_delay_s) + # Bail out if a stop landed during the delay — never retry past a stop. + if not getattr(self, 'running', True): + break + continue # next user pass (fresh request issued at loop top) - # --- Handle Unexpected Errors --- - except Exception as e: - request_duration_s = time.monotonic() - request_start_time - error_message = f"Unexpected error during request execution: {e}" - logger.error(f"Step {step_identifier}: {error_message} ({request_duration_s*1000:.2f} ms)", exc_info=self.config.debug) - response_status = 596 # Custom code for unexpected errors - break # Exit retry loop + # Success, no retries left, or user-stop: stop re-attempting. + break # --- Post-Request Processing --- @@ -2702,6 +2973,16 @@ async def _execute_request_step( logger.debug(f"Step {step_identifier}: Skipping extraction due to request execution failure.") + # --- ADDITIVE: declarative assertions (step.assertions) --- + # Evaluate against the response/extracted context once the request has + # completed. This is diagnostic and runs regardless of onFailure (so a + # stop-on-failure step still records what its assertions saw). It never + # raises and never changes the flow-control outcome — it only records + # pass/fail into the context. Skipped when the request never completed. + if request_succeeded: + self._evaluate_assertions(step.id, step_identifier, getattr(step, 'assertions', None), context) + + # Increment metrics only if the request was actually sent and received a response status if request_succeeded: await self.metrics.increment() diff --git a/tests/fixtures/golden_old_flow.json b/tests/fixtures/golden_old_flow.json new file mode 100644 index 0000000..714a70c --- /dev/null +++ b/tests/fixtures/golden_old_flow.json @@ -0,0 +1,94 @@ +{ + "id": "golden-old-flow-1", + "name": "Golden Old Flow (pre-sprint)", + "description": "A real pre-sprint flow used as a cross-app conformance anchor. It MUST parse to an identical execution model with no schemaVersion, with \"1.0\", and with an unknown MINOR such as \"1.5\". Do not add schemaVersion to this base file.", + "headers": { + "Accept": "application/json", + "X-Client": "flowrunner-cli" + }, + "staticVars": { + "basePath": "/api/v1", + "maxItems": 5, + "enabled": true + }, + "steps": [ + { + "id": "login", + "name": "Authenticate", + "type": "request", + "method": "POST", + "url": "{{basePath}}/login", + "headers": { + "Content-Type": "application/json" + }, + "body": { + "user": "##VAR:string:username##", + "pass": "##VAR:string:password##" + }, + "extract": { + "token": "body.data.sessionToken", + "loginStatus": ".status" + }, + "onFailure": "stop" + }, + { + "id": "check-token", + "name": "Token present?", + "type": "condition", + "conditionData": { + "variable": "token", + "operator": "exists", + "value": "" + }, + "then": [ + { + "id": "list-items", + "name": "List items", + "type": "request", + "method": "GET", + "url": "{{basePath}}/items?limit={{maxItems}}", + "headers": { + "Authorization": "Bearer {{token}}" + }, + "extract": { + "items": "body.data.items", + "firstId": "body.data.items[0].id" + }, + "onFailure": "continue" + }, + { + "id": "loop-items", + "name": "Iterate items", + "type": "loop", + "source": "{{items}}", + "loopVariable": "item", + "steps": [ + { + "id": "fetch-item", + "name": "Fetch item detail", + "type": "request", + "method": "GET", + "url": "{{basePath}}/items/{{item.id}}", + "onFailure": "continue" + } + ] + } + ], + "else": [ + { + "id": "no-token-transform", + "name": "Record failure marker", + "type": "transform", + "ops": [ + { + "op": "to_string", + "set": "authFailed", + "args": ["{{loginStatus}}"], + "options": {} + } + ] + } + ] + } + ] +} diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py new file mode 100644 index 0000000..1db5a48 --- /dev/null +++ b/tests/unit/conftest.py @@ -0,0 +1,19 @@ +"""Shared test bootstrapping for the unit suite. + +``flow_runner`` imports ``psutil`` at module load time even though it is not +used in the exercised code paths. The historical test module stubbed it inline; +hoisting the stub into a conftest lets every unit test module import +``flow_runner`` without repeating the shim (and without requiring psutil to be +installed in the test environment). +""" + +import os +import sys +import types + +sys.modules.setdefault("psutil", types.ModuleType("psutil")) + +# Ensure the repo root is importable regardless of pytest's rootdir/invocation. +_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +if _REPO_ROOT not in sys.path: + sys.path.insert(0, _REPO_ROOT) diff --git a/tests/unit/test_assertions.py b/tests/unit/test_assertions.py new file mode 100644 index 0000000..d3e0a55 --- /dev/null +++ b/tests/unit/test_assertions.py @@ -0,0 +1,186 @@ +"""Tests for additive declarative ``step.assertions`` on request steps. + +Assertions reuse the frozen ``conditionData`` operator vocabulary and are +evaluated against the request result (status/headers/body/extracted vars) after +the request completes. Pass/fail is recorded into the execution context; unknown +operators or missing targets degrade to a FAILED assertion with a warning and +never crash the run. +""" + +import logging +from typing import Any, Dict +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from flow_runner import ( + Assertion, + ContainerConfig, + FlowMap, + FlowRunner, + Metrics, + RequestStep, + get_value_from_context, +) + + +@pytest.fixture +def empty_flow() -> FlowMap: + return FlowMap(name="test", steps=[], staticVars={}) + + +def make_runner(config: ContainerConfig, flow: FlowMap) -> FlowRunner: + metrics = Metrics() + metrics.increment = AsyncMock() + metrics.record_flow_duration = AsyncMock() + runner = FlowRunner(config, flow, metrics) + runner.metrics = metrics + runner.running = True + return runner + + +def _resp(status: int, body): + r = AsyncMock() + r.status = status + r.headers = {"Content-Type": "application/json"} + r.json = AsyncMock(return_value=body) + r.text = AsyncMock(return_value="{}") + r.read = AsyncMock(return_value=b"{}") + return r + + +def _session(resp): + session = MagicMock() + cm = AsyncMock() + cm.__aenter__.return_value = resp + cm.__aexit__.return_value = AsyncMock() + session.request.return_value = cm + return session + + +# --- model parsing ---------------------------------------------------------- + +def test_request_step_assertions_parsed(): + step = RequestStep.model_validate({ + "id": "s1", "type": "request", "method": "GET", "url": "/a", + "onFailure": "continue", + "assertions": [ + {"name": "ok", "variable": "response_s1_status", "operator": "equals", "value": "200"}, + ], + }) + assert step.assertions is not None + assert isinstance(step.assertions[0], Assertion) + assert step.assertions[0].operator == "equals" + + +def test_request_step_assertions_absent_defaults_none(): + step = RequestStep.model_validate({ + "id": "s1", "type": "request", "method": "GET", "url": "/a", + "onFailure": "continue", + }) + assert step.assertions is None + + +# --- evaluation: pass / fail recorded -------------------------------------- + +@pytest.mark.asyncio +async def test_assertions_all_pass_recorded(empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session(_resp(200, {"data": {"ok": True, "count": 5}})) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + assertions=[ + Assertion(name="status ok", variable="response_s1_status", operator="equals", value="200"), + Assertion(name="flag true", variable="response_s1_body.data.ok", operator="is_true"), + Assertion(name="count > 3", variable="response_s1_body.data.count", operator="greater_than", value="3"), + ], + ) + ctx: Dict[str, Any] = {} + await runner._execute_request_step(step, session, {}, {}, ctx) + + results = get_value_from_context(ctx, "response_s1_assertions") + assert isinstance(results, list) and len(results) == 3 + assert all(r["passed"] for r in results) + assert get_value_from_context(ctx, "response_s1_assertions_passed") is True + + +@pytest.mark.asyncio +async def test_assertions_failure_recorded(empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session(_resp(500, {"data": {"ok": False}})) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + assertions=[ + Assertion(name="expects 200", variable="response_s1_status", operator="equals", value="200"), + ], + ) + ctx: Dict[str, Any] = {} + await runner._execute_request_step(step, session, {}, {}, ctx) + + results = get_value_from_context(ctx, "response_s1_assertions") + assert len(results) == 1 + assert results[0]["passed"] is False + assert get_value_from_context(ctx, "response_s1_assertions_passed") is False + + +# --- degrade gracefully ----------------------------------------------------- + +@pytest.mark.asyncio +async def test_unknown_operator_degrades_without_crash(empty_flow, caplog): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session(_resp(200, {})) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + assertions=[ + Assertion(name="weird", variable="response_s1_status", operator="frobnicate", value="x"), + ], + ) + ctx: Dict[str, Any] = {} + with caplog.at_level(logging.WARNING): + # Must not raise. + await runner._execute_request_step(step, session, {}, {}, ctx) + + results = get_value_from_context(ctx, "response_s1_assertions") + assert len(results) == 1 + # Unknown operator => failed assertion, flagged, run continues. + assert results[0]["passed"] is False + assert get_value_from_context(ctx, "response_s1_assertions_passed") is False + + +@pytest.mark.asyncio +async def test_unknown_target_missing_variable_degrades(empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session(_resp(200, {})) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + assertions=[ + Assertion(name="missing exists", variable="response_s1_body.nope.deep", operator="exists"), + ], + ) + ctx: Dict[str, Any] = {} + await runner._execute_request_step(step, session, {}, {}, ctx) + results = get_value_from_context(ctx, "response_s1_assertions") + assert results[0]["passed"] is False # missing target => 'exists' is False + + +@pytest.mark.asyncio +async def test_no_assertions_records_nothing(empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session(_resp(200, {})) + + step = RequestStep(id="s1", type="request", method="GET", url="/a", onFailure="continue") + ctx: Dict[str, Any] = {} + await runner._execute_request_step(step, session, {}, {}, ctx) + + from flow_runner import _MISSING + assert get_value_from_context(ctx, "response_s1_assertions") is _MISSING + assert get_value_from_context(ctx, "response_s1_assertions_passed") is _MISSING diff --git a/tests/unit/test_golden_old_flow.py b/tests/unit/test_golden_old_flow.py new file mode 100644 index 0000000..8ebdbce --- /dev/null +++ b/tests/unit/test_golden_old_flow.py @@ -0,0 +1,81 @@ +"""Golden cross-app conformance test for the shared .flow.json contract. + +INVARIANT (mirrors the FlowRunner UI repo's __tests__/goldenOldFlow.test.js): +a real pre-sprint flow MUST parse to an IDENTICAL execution model whether it +carries no ``schemaVersion`` at all, ``"1.0"``, or an unknown MINOR such as +``"1.5"``. Absence of ``schemaVersion`` means ``"1.0"``. This is the guard that +keeps the 24/7 CLI alive when it meets a slightly-newer file: an additive MINOR +bump must never change how an old flow executes, and must never be rejected. + +These tests are written to pass on the *current* parser (before the version gate +lands) and must keep passing after it — old flows are never rejected. +""" + +import copy +import json +import os + +import pytest + +from flow_runner import FlowMap + + +FIXTURE = os.path.join( + os.path.dirname(__file__), "..", "fixtures", "golden_old_flow.json" +) + + +def _load_raw(): + with open(FIXTURE, "r", encoding="utf-8") as f: + return json.load(f) + + +def _execution_model(flowmap: FlowMap) -> dict: + """A normalized dump of the parsed model that reflects execution semantics. + + ``schemaVersion`` is a diagnostic, not part of the execution model, so it is + excluded from the comparison: two files that differ only by an additive + ``schemaVersion`` MINOR must produce byte-identical execution models. + """ + dump = flowmap.model_dump(by_alias=True) + dump.pop("schemaVersion", None) + return dump + + +def test_golden_old_flow_accepted_without_schema_version(): + """A real pre-sprint flow with NO schemaVersion parses successfully.""" + raw = _load_raw() + assert "schemaVersion" not in raw # the base fixture is a genuine old flow + flowmap = FlowMap.model_validate(raw) + assert flowmap.name == "Golden Old Flow (pre-sprint)" + assert len(flowmap.steps) == 2 + # staticVars / extract / conditionData / nested then/else/loop survived intact. + assert flowmap.staticVars["maxItems"] == 5 + login = flowmap.steps[0] + assert login.type == "request" + assert login.extract["token"] == "body.data.sessionToken" + + +@pytest.mark.parametrize("version", [None, "1.0", "1.5"]) +def test_golden_old_flow_parses_identically_across_minor_versions(version): + """Absent / "1.0" / unknown MINOR "1.5" all yield the SAME execution model.""" + baseline = _execution_model(FlowMap.model_validate(_load_raw())) + + raw = _load_raw() + if version is not None: + raw["schemaVersion"] = version + flowmap = FlowMap.model_validate(raw) + + assert _execution_model(flowmap) == baseline, ( + f"schemaVersion={version!r} changed the execution model; additive MINOR " + "bumps must be behavior-preserving for old flows." + ) + + +def test_golden_old_flow_unknown_minor_not_rejected(): + """An unknown MINOR must be tolerated (accepted), never rejected.""" + raw = _load_raw() + raw["schemaVersion"] = "1.99" + # Must not raise. + flowmap = FlowMap.model_validate(raw) + assert flowmap.name == "Golden Old Flow (pre-sprint)" diff --git a/tests/unit/test_request_retries.py b/tests/unit/test_request_retries.py new file mode 100644 index 0000000..c398f2b --- /dev/null +++ b/tests/unit/test_request_retries.py @@ -0,0 +1,203 @@ +"""Tests for the additive per-request ``step.retries={count, delayMs}`` policy. + +Mirrors the FlowRunner UI JS engine (flowRunner.js ``_executeRequestStep``): + +- ``count`` defaults to 0 => single attempt, IDENTICAL to prior behavior. +- A retry fires on a non-2xx HTTP status OR a network/fetch error. +- ``delayMs`` is slept between attempts. +- A user-requested stop (``self.running == False``) is NEVER retried. +- Each attempt issues a fresh request (the CLI's analogue of a fresh + AbortController per attempt in the browser). +""" + +from typing import Any, Dict +from unittest.mock import AsyncMock, MagicMock + +import aiohttp +import asyncio +import pytest + +from flow_runner import ( + ContainerConfig, + FlowMap, + FlowRunner, + Metrics, + RequestStep, + RetryConfig, +) + + +@pytest.fixture +def empty_flow() -> FlowMap: + return FlowMap(name="test", steps=[], staticVars={}) + + +def make_runner(config: ContainerConfig, flow: FlowMap) -> FlowRunner: + metrics = Metrics() + metrics.increment = AsyncMock() + metrics.record_flow_duration = AsyncMock() + runner = FlowRunner(config, flow, metrics) + runner.metrics = metrics + # A request step only ever executes while the runner is actively running; + # the executor loop runs inside `while self.running`. Reflect that here so + # the user-retry policy (which must not fire past a user-stop) is exercised. + runner.running = True + return runner + + +def _resp(status: int): + r = AsyncMock() + r.status = status + r.headers = {"Content-Type": "application/json"} + r.json = AsyncMock(return_value={}) + r.text = AsyncMock(return_value="{}") + r.read = AsyncMock(return_value=b"{}") + return r + + +def _cm(resp): + cm = AsyncMock() + cm.__aenter__.return_value = resp + cm.__aexit__.return_value = AsyncMock() + return cm + + +def _session_from(side_effect): + session = MagicMock() + session.request.side_effect = side_effect + return session + + +# --- retries model on RequestStep ------------------------------------------ + +def test_request_step_retries_field_parsed(): + step = RequestStep.model_validate({ + "id": "s1", "type": "request", "method": "GET", "url": "/a", + "onFailure": "continue", "retries": {"count": 2, "delayMs": 50}, + }) + assert isinstance(step.retries, RetryConfig) + assert step.retries.count == 2 + assert step.retries.delayMs == 50 + + +def test_request_step_retries_absent_defaults_none(): + step = RequestStep.model_validate({ + "id": "s1", "type": "request", "method": "GET", "url": "/a", + "onFailure": "continue", + }) + assert step.retries is None + + +# --- default (count 0) is a single attempt --------------------------------- + +@pytest.mark.asyncio +async def test_no_retries_single_attempt_on_non_2xx(monkeypatch, empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session_from([_cm(_resp(404))]) + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + step = RequestStep(id="s1", type="request", method="GET", url="/a", onFailure="continue") + await runner._execute_request_step(step, session, {}, {}, {}) + assert session.request.call_count == 1 # no user-retry on 4xx by default + + +# --- retry on non-2xx ------------------------------------------------------ + +@pytest.mark.asyncio +async def test_retries_on_non_2xx_then_success(monkeypatch, empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session_from([_cm(_resp(404)), _cm(_resp(200))]) + sleep_mock = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep_mock) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + retries=RetryConfig(count=2, delayMs=25), + ) + ctx: Dict[str, Any] = {} + await runner._execute_request_step(step, session, {}, {}, ctx) + assert session.request.call_count == 2 # 404 then 200 + from flow_runner import get_value_from_context + assert get_value_from_context(ctx, "response_s1_status") == 200 + # delayMs was slept at least once + assert any(call.args and abs(call.args[0] - 0.025) < 1e-9 for call in sleep_mock.await_args_list) + + +@pytest.mark.asyncio +async def test_retries_exhausted_on_persistent_non_2xx(monkeypatch, empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + # Use 404 (a non-2xx the built-in 5xx resilience loop does NOT retry) so + # the attempt count reflects ONLY the user retry policy. + session = _session_from([_cm(_resp(404)), _cm(_resp(404)), _cm(_resp(404))]) + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + retries=RetryConfig(count=2, delayMs=0), + ) + ctx: Dict[str, Any] = {} + await runner._execute_request_step(step, session, {}, {}, ctx) + # 1 initial + 2 retries = 3 total attempts, all 404. + assert session.request.call_count == 3 + from flow_runner import get_value_from_context + assert get_value_from_context(ctx, "response_s1_status") == 404 + + +# --- retry on network error ------------------------------------------------- + +@pytest.mark.asyncio +async def test_retries_on_network_error_then_success(monkeypatch, empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session_from([aiohttp.ClientConnectionError(), _cm(_resp(200))]) + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + retries=RetryConfig(count=3, delayMs=0), + ) + ctx: Dict[str, Any] = {} + await runner._execute_request_step(step, session, {}, {}, ctx) + assert session.request.call_count == 2 + from flow_runner import get_value_from_context + assert get_value_from_context(ctx, "response_s1_status") == 200 + + +# --- user-stop is never retried -------------------------------------------- + +@pytest.mark.asyncio +async def test_user_stop_not_retried(monkeypatch, empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + runner.running = False # simulate a user-requested stop + # 404 (not retried by the built-in 5xx loop) isolates the user-retry path. + session = _session_from([_cm(_resp(404)), _cm(_resp(200))]) + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + retries=RetryConfig(count=5, delayMs=0), + ) + await runner._execute_request_step(step, session, {}, {}, {}) + # Stop signal => no user-retry, single attempt only. + assert session.request.call_count == 1 + + +# --- a 2xx never triggers user-retry --------------------------------------- + +@pytest.mark.asyncio +async def test_success_2xx_no_retry(monkeypatch, empty_flow): + cfg = ContainerConfig(flow_target_url="http://base.com", sim_users=1) + runner = make_runner(cfg, empty_flow) + session = _session_from([_cm(_resp(200)), _cm(_resp(200))]) + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + step = RequestStep( + id="s1", type="request", method="GET", url="/a", onFailure="continue", + retries=RetryConfig(count=3, delayMs=0), + ) + await runner._execute_request_step(step, session, {}, {}, {}) + assert session.request.call_count == 1 diff --git a/tests/unit/test_schema_version_gate.py b/tests/unit/test_schema_version_gate.py new file mode 100644 index 0000000..12b657a --- /dev/null +++ b/tests/unit/test_schema_version_gate.py @@ -0,0 +1,101 @@ +"""Version-gate tests for the additive, OPTIONAL ``schemaVersion`` field. + +Contract (see the FlowRunner UI repo's docs/schema-versioning.md): + +- ``schemaVersion`` is an OPTIONAL top-level string ``"MAJOR.MINOR"``. +- ABSENCE means ``"1.0"``. Absent and ``"1.0"`` are byte-equivalent in meaning. +- Unknown **MINOR** (same MAJOR the CLI supports, e.g. ``"1.5"``) => TOLERATE: + accept and run, degrade gracefully on any unknown construct, warn once. +- Unknown **MAJOR** (``>= 2``, e.g. ``"2.0"``) => REJECT LOUDLY: refuse the flow + with a clear error rather than best-effort mis-executing it. +- A non-string (e.g. integer ``2``) is coerced-and-warned, then gated on its + MAJOR like any other value — never a silent crash. + +The gate must NEVER reject an old flow: the golden conformance suite stays green. +""" + +import logging + +import pytest +from pydantic import ValidationError + +from flow_runner import FlowMap + + +def _base_flow(**extra): + data = { + "name": "gate-flow", + "steps": [ + { + "id": "s1", + "name": "req", + "type": "request", + "method": "GET", + "url": "/ping", + "onFailure": "continue", + } + ], + } + data.update(extra) + return data + + +# --- Accept: absent / current MAJOR / unknown MINOR ------------------------ + +@pytest.mark.parametrize("version", [None, "1.0", "1.1", "1.5", "1.99"]) +def test_accepts_absent_and_known_major(version): + data = _base_flow() + if version is not None: + data["schemaVersion"] = version + flowmap = FlowMap.model_validate(data) # must not raise + assert flowmap.name == "gate-flow" + + +def test_unknown_minor_warns_but_accepts(caplog): + data = _base_flow(schemaVersion="1.7") + with caplog.at_level(logging.WARNING): + FlowMap.model_validate(data) + assert any("schemaVersion" in rec.message for rec in caplog.records), ( + "an unknown MINOR should emit a degrade-gracefully warning" + ) + + +def test_known_minor_1_0_does_not_warn(caplog): + data = _base_flow(schemaVersion="1.0") + with caplog.at_level(logging.WARNING): + FlowMap.model_validate(data) + assert not any("schemaVersion" in rec.message for rec in caplog.records) + + +# --- Reject: unknown MAJOR -------------------------------------------------- + +@pytest.mark.parametrize("version", ["2.0", "2.3", "3.0", "10.0"]) +def test_unknown_major_rejected_loudly(version): + data = _base_flow(schemaVersion=version) + with pytest.raises(ValidationError) as exc: + FlowMap.model_validate(data) + # The error must be attributable to schemaVersion and mention the version. + msg = str(exc.value) + assert "schemaVersion" in msg + assert version in msg + + +# --- Coercion: non-string values never crash silently ----------------------- + +def test_integer_major_coerced_and_gated(caplog): + # Integer 1 should coerce to "1.0"-equivalent and be accepted. + with caplog.at_level(logging.WARNING): + flowmap = FlowMap.model_validate(_base_flow(schemaVersion=1)) + assert flowmap.name == "gate-flow" + + +def test_integer_unknown_major_still_rejected(): + with pytest.raises(ValidationError): + FlowMap.model_validate(_base_flow(schemaVersion=2)) + + +def test_malformed_version_string_does_not_crash(): + # A garbage value must not raise a raw exception type other than a clean + # validation rejection; the parser degrades to a spec'd rejection. + with pytest.raises(ValidationError): + FlowMap.model_validate(_base_flow(schemaVersion="not-a-version"))