From 3b685e9dc85827c516f7eb1e4d514ca120058b83 Mon Sep 17 00:00:00 2001 From: DhanashreePetare Date: Wed, 29 Jul 2026 16:47:44 +0530 Subject: [PATCH 1/2] feat: add workflow engine - parser, step context, steps, engine, CLI --- databusclient/cli.py | 40 ++++++ databusclient/workflow/__init__.py | 4 + databusclient/workflow/context.py | 100 +++++++++++++++ databusclient/workflow/engine.py | 110 +++++++++++++++++ databusclient/workflow/parser.py | 188 +++++++++++++++++++++++++++++ databusclient/workflow/steps.py | 150 +++++++++++++++++++++++ poetry.lock | 87 ++++++++++++- pyproject.toml | 1 + tests/test_step_context.py | 75 ++++++++++++ tests/test_workflow_engine.py | 136 +++++++++++++++++++++ tests/test_workflow_parser.py | 180 +++++++++++++++++++++++++++ tests/test_workflow_steps.py | 159 ++++++++++++++++++++++++ 12 files changed, 1228 insertions(+), 2 deletions(-) create mode 100644 databusclient/workflow/__init__.py create mode 100644 databusclient/workflow/context.py create mode 100644 databusclient/workflow/engine.py create mode 100644 databusclient/workflow/parser.py create mode 100644 databusclient/workflow/steps.py create mode 100644 tests/test_step_context.py create mode 100644 tests/test_workflow_engine.py create mode 100644 tests/test_workflow_parser.py create mode 100644 tests/test_workflow_steps.py diff --git a/databusclient/cli.py b/databusclient/cli.py index fab57a9..92989c4 100644 --- a/databusclient/cli.py +++ b/databusclient/cli.py @@ -13,6 +13,9 @@ from databusclient.manifest.replay import ManifestReplayError, replay_manifest, load_manifest from databusclient.manifest.summary import format_summary from databusclient.extensions import webdav +from databusclient.workflow.parser import WorkflowParseError, parse_workflow +from databusclient.workflow.engine import WorkflowEngine, WorkflowExecutionError +from databusclient.workflow.context import StepContext @click.group() @@ -568,5 +571,42 @@ def manifest_summary(manifest_path): except ManifestReplayError as e: raise click.ClickException(str(e)) +@app.group() +def workflow(): + """ + Workflow utilities. + + Run multi-step download/deploy/delete pipelines defined in YAML. + """ + pass + + +@workflow.command("run") +@click.argument("workflow_path", type=click.Path(exists=True, dir_okay=False)) +def workflow_run(workflow_path): + """ + Run a declarative workflow pipeline from a YAML file. + + Executes each step in order, chaining outputs between steps via + ${steps.name.output_files}-style references, and applying each + step's on_error behavior (fail/continue/retry). + """ + try: + parsed = parse_workflow(workflow_path) + except WorkflowParseError as e: + raise click.ClickException(str(e)) + + context = StepContext() + engine = WorkflowEngine(context=context) + + try: + results = engine.run(parsed["steps"]) + except WorkflowExecutionError as e: + raise click.ClickException(str(e)) + + click.echo("Workflow complete.") + for result in results: + click.echo(f" {result.name}: {result.status}") + if __name__ == "__main__": app() diff --git a/databusclient/workflow/__init__.py b/databusclient/workflow/__init__.py new file mode 100644 index 0000000..f560e5f --- /dev/null +++ b/databusclient/workflow/__init__.py @@ -0,0 +1,4 @@ +"""Workflow engine for the Databus Python Client. + +Orchestrates multi-step download/deploy/delete pipelines defined in YAML. +""" \ No newline at end of file diff --git a/databusclient/workflow/context.py b/databusclient/workflow/context.py new file mode 100644 index 0000000..4c0cd6e --- /dev/null +++ b/databusclient/workflow/context.py @@ -0,0 +1,100 @@ +"""StepContext — tracks step outputs and resolves ${steps.name.key} references at runtime. + +WorkflowParser resolves ${VAR_NAME} environment variables at parse time, +but deliberately leaves ${steps.step_name.output_files} tokens untouched, +since those values don't exist until the referenced step has actually run. +StepContext is what resolves them, once the WorkflowEngine has executed +each step in order. +""" + +from __future__ import annotations + +import re +from typing import Any, Dict + +# Matches a single ${steps.step_name.key} token. +_STEP_REF_RE = re.compile(r"\$\{steps\.([^.}]+)\.([^}]+)\}") + + +class StepReferenceError(Exception): + """Raised when a ${steps.name.key} reference cannot be resolved.""" + + +class StepContext: + """Stores per-step outputs and resolves ${steps.name.key} references. + + manifest_context is accepted but unused in Milestone 4 -- it exists as + a seam so Milestone 5 can wire in manifest recording without changing + this class's structure. When None, it has zero effect, matching the + manifest_context=None pattern already used throughout download.py, + deploy.py, and delete.py. + """ + + def __init__(self, manifest_context=None) -> None: + self._outputs: Dict[str, Dict[str, Any]] = {} + self.manifest_context = manifest_context + + def set_output(self, step_name: str, key: str, value: Any) -> None: + """Record an output value produced by a step. + + Args: + step_name: Name of the step that produced this output. + key: Output key, e.g. "output_files". + value: The value to store (e.g. a list of file paths). + """ + self._outputs.setdefault(step_name, {})[key] = value + + def get_output(self, step_name: str, key: str) -> Any: + """Retrieve a previously recorded output value. + + Raises: + StepReferenceError: If the step or key is unknown. + """ + if step_name not in self._outputs: + raise StepReferenceError( + f"Reference to unknown or not-yet-executed step '{step_name}'." + ) + if key not in self._outputs[step_name]: + raise StepReferenceError( + f"Step '{step_name}' has no recorded output '{key}'. " + f"Available outputs: {sorted(self._outputs[step_name].keys())}." + ) + return self._outputs[step_name][key] + + def resolve(self, value: Any) -> Any: + """Recursively resolve ${steps.name.key} references in a value. + + A value that is EXACTLY a single ${steps.name.key} token (nothing + else in the string) resolves to the raw stored value (e.g. a list), + preserving its type. A token embedded inside a larger string is + resolved by inserting str(value) in place, same as environment + variable substitution. + + Args: + value: A string, list, dict, or scalar value from a step config. + + Returns: + The value with all ${steps.*} references resolved. + + Raises: + StepReferenceError: If a referenced step/key is unknown. + """ + if isinstance(value, str): + full_match = _STEP_REF_RE.fullmatch(value) + if full_match: + step_name, key = full_match.group(1), full_match.group(2) + return self.get_output(step_name, key) + + def _replace(match: re.Match) -> str: + step_name, key = match.group(1), match.group(2) + return str(self.get_output(step_name, key)) + + return _STEP_REF_RE.sub(_replace, value) + + if isinstance(value, list): + return [self.resolve(item) for item in value] + + if isinstance(value, dict): + return {k: self.resolve(v) for k, v in value.items()} + + return value \ No newline at end of file diff --git a/databusclient/workflow/engine.py b/databusclient/workflow/engine.py new file mode 100644 index 0000000..d5cec54 --- /dev/null +++ b/databusclient/workflow/engine.py @@ -0,0 +1,110 @@ +"""WorkflowEngine — executes a sequence of parsed workflow steps in order. + +Applies each step's on_error behavior (fail/continue/retry) around a call +to the step's run() method. Retries operate at the whole-step level -- +the engine has no visibility into partial failures inside a step (e.g. +one file out of several failing during a download), since download(), +deploy(), and delete() are called as single atomic operations. +""" + +from __future__ import annotations + +import time +from typing import Any, Dict, List + +from databusclient.workflow.context import StepContext +from databusclient.workflow.steps import STEP_REGISTRY + + +class WorkflowExecutionError(Exception): + """Raised when a workflow step fails and on_error is 'fail' (or defaults to it).""" + + +class StepResult: + """Outcome of running a single step.""" + + def __init__(self, name: str, status: str, error: Exception | None = None, + attempts: int = 1) -> None: + self.name = name + self.status = status # "success", "failed", "skipped_error" + self.error = error + self.attempts = attempts + + +class WorkflowEngine: + """Runs a parsed workflow's steps in order, handling errors per step.""" + + def __init__(self, context: StepContext | None = None) -> None: + self.context = context or StepContext() + self.results: List[StepResult] = [] + + def run(self, steps: List[Dict[str, Any]]) -> List[StepResult]: + """Execute all steps in order. + + Args: + steps: List of validated, environment-substituted step dicts + (as produced by WorkflowParser.parse_workflow). + + Returns: + List of StepResult, one per step actually attempted. + + Raises: + WorkflowExecutionError: If a step with on_error 'fail' (the + default) ultimately fails. + """ + for step_config in steps: + result = self._run_step_with_error_handling(step_config) + self.results.append(result) + if result.status == "failed": + # on_error was 'fail' (or defaulted to it) -- stop the workflow. + raise WorkflowExecutionError( + f"Step '{result.name}' failed: {result.error}" + ) + return self.results + + def _run_step_with_error_handling(self, step_config: Dict[str, Any]) -> StepResult: + name = step_config["name"] + command = step_config["command"] + on_error = step_config.get("on_error", "fail") + + step_class = STEP_REGISTRY.get(command) + if step_class is None: + # Should already be caught by the parser, but defend anyway. + raise WorkflowExecutionError( + f"Step '{name}' has unknown command '{command}'." + ) + step = step_class() + + if on_error == "retry": + return self._run_with_retry(name, step, step_config) + + try: + step.run(step_config, self.context) + return StepResult(name, "success") + except Exception as exc: + if on_error == "continue": + print(f"WARNING: step '{name}' failed and on_error is 'continue': {exc}") + return StepResult(name, "skipped_error", error=exc) + # on_error == "fail" (or missing/defaulted to fail) + return StepResult(name, "failed", error=exc) + + def _run_with_retry(self, name: str, step: Any, step_config: Dict[str, Any]) -> StepResult: + retry_config = step_config["retry"] + max_attempts = retry_config["max_attempts"] + delay_seconds = retry_config["delay_seconds"] + + last_error: Exception | None = None + for attempt in range(1, max_attempts + 1): + try: + step.run(step_config, self.context) + return StepResult(name, "success", attempts=attempt) + except Exception as exc: + last_error = exc + print( + f"WARNING: step '{name}' attempt {attempt}/{max_attempts} " + f"failed: {exc}" + ) + if attempt < max_attempts: + time.sleep(delay_seconds) + + return StepResult(name, "failed", error=last_error, attempts=max_attempts) \ No newline at end of file diff --git a/databusclient/workflow/parser.py b/databusclient/workflow/parser.py new file mode 100644 index 0000000..267dda1 --- /dev/null +++ b/databusclient/workflow/parser.py @@ -0,0 +1,188 @@ +"""WorkflowParser — loads and validates a YAML workflow pipeline file. + +Parses a YAML file describing a sequence of steps (download/deploy/delete), +validates its structure, and substitutes environment variables of the form +${VAR_NAME}. References of the form ${steps.step_name.output_files} are +left untouched here -- those are resolved at runtime by StepContext once +each step has actually run, since their values don't exist yet at parse time. +""" + +from __future__ import annotations + +import os +import re +from typing import Any, Dict, List + +import yaml + +VALID_COMMANDS = {"download", "deploy", "delete"} +VALID_ON_ERROR = {"fail", "continue", "retry"} + +# Matches ${...} tokens. The captured group is everything between the braces. +_TOKEN_RE = re.compile(r"\$\{([^}]+)\}") + + +class WorkflowParseError(Exception): + """Raised when a workflow YAML file is invalid or fails validation.""" + + +class MissingEnvVarError(WorkflowParseError): + """Raised when a workflow references an environment variable that is not set.""" + + +def _load_yaml(path: str) -> Any: + """Load a YAML file using safe_load (never load arbitrary Python objects).""" + try: + with open(path, "r", encoding="utf-8-sig") as f: + return yaml.safe_load(f) + except FileNotFoundError as e: + raise WorkflowParseError(f"Workflow file not found: {path}") from e + except yaml.YAMLError as e: + raise WorkflowParseError(f"Workflow file is not valid YAML: {path}\n{e}") from e + + +def _substitute_value(value: Any, step_name: str) -> Any: + """Recursively substitute ${VAR_NAME} environment variables in a value. + + Tokens of the form ${steps.*} are left untouched -- they are resolved + later, at runtime, by StepContext once earlier steps have produced + their outputs. Only non-"steps."-prefixed tokens are treated as + environment variables here. + + Args: + value: A string, list, dict, or scalar value from the parsed YAML. + step_name: Name of the step this value belongs to (for error messages). + + Returns: + The value with environment variables substituted. + + Raises: + MissingEnvVarError: If a referenced environment variable is not set. + """ + if isinstance(value, str): + def _replace(match: re.Match) -> str: + token = match.group(1) + if token.startswith("steps."): + # Leave step-output references untouched for runtime resolution. + return match.group(0) + env_value = os.environ.get(token) + if env_value is None: + raise MissingEnvVarError( + f"Step '{step_name}' references environment variable " + f"'{token}' which is not set." + ) + return env_value + + return _TOKEN_RE.sub(_replace, value) + + if isinstance(value, list): + return [_substitute_value(item, step_name) for item in value] + + if isinstance(value, dict): + return {k: _substitute_value(v, step_name) for k, v in value.items()} + + return value + + +def _validate_step(step: Any, index: int, seen_names: set) -> Dict[str, Any]: + """Validate the generic structure of a single step. + + Only validates fields common to all step types (name, command, on_error, + retry config). Command-specific required fields (e.g. 'uri' for download) + are validated later, when the step actually executes. + + Args: + step: The raw step dict from the parsed YAML. + index: Position of this step in the steps list (for error messages). + seen_names: Set of step names already seen, for duplicate detection. + + Returns: + The validated step dict (unchanged, just checked). + + Raises: + WorkflowParseError: If the step is structurally invalid. + """ + if not isinstance(step, dict): + raise WorkflowParseError(f"Step at index {index} must be a mapping/object.") + + name = step.get("name") + if not name or not isinstance(name, str): + raise WorkflowParseError(f"Step at index {index} is missing a valid 'name'.") + + if name in seen_names: + raise WorkflowParseError(f"Duplicate step name '{name}'. Step names must be unique.") + seen_names.add(name) + + command = step.get("command") + if command not in VALID_COMMANDS: + raise WorkflowParseError( + f"Step '{name}' has invalid command '{command}'. " + f"Must be one of: {sorted(VALID_COMMANDS)}." + ) + + on_error = step.get("on_error", "fail") + if on_error not in VALID_ON_ERROR: + raise WorkflowParseError( + f"Step '{name}' has invalid on_error '{on_error}'. " + f"Must be one of: {sorted(VALID_ON_ERROR)}." + ) + + if on_error == "retry": + retry_config = step.get("retry") + if not isinstance(retry_config, dict): + raise WorkflowParseError( + f"Step '{name}' has on_error: retry but is missing a 'retry' " + f"configuration block with 'max_attempts' and 'delay_seconds'." + ) + max_attempts = retry_config.get("max_attempts") + if not isinstance(max_attempts, int) or max_attempts < 1: + raise WorkflowParseError( + f"Step '{name}' retry.max_attempts must be a positive integer." + ) + delay_seconds = retry_config.get("delay_seconds") + if not isinstance(delay_seconds, (int, float)) or delay_seconds < 0: + raise WorkflowParseError( + f"Step '{name}' retry.delay_seconds must be a non-negative number." + ) + + return step + + +def parse_workflow(path: str) -> Dict[str, Any]: + """Load, validate, and substitute environment variables in a workflow YAML file. + + Args: + path: Path to the workflow YAML file. + + Returns: + A dict with keys: + "manifest": Optional manifest output path (str or None). + "steps": List of validated, environment-substituted step dicts. + + Raises: + WorkflowParseError: If the file is missing, invalid YAML, or fails + structural validation. + MissingEnvVarError: If a step references an unset environment variable. + """ + raw = _load_yaml(path) + + if not isinstance(raw, dict): + raise WorkflowParseError("Workflow file root must be a mapping/object.") + + steps = raw.get("steps") + if not isinstance(steps, list) or not steps: + raise WorkflowParseError( + "Workflow file must have a non-empty 'steps' list." + ) + + seen_names: set = set() + validated_steps: List[Dict[str, Any]] = [] + for index, step in enumerate(steps): + validated = _validate_step(step, index, seen_names) + substituted = _substitute_value(validated, validated["name"]) + validated_steps.append(substituted) + + return { + "manifest": raw.get("manifest"), + "steps": validated_steps, + } \ No newline at end of file diff --git a/databusclient/workflow/steps.py b/databusclient/workflow/steps.py new file mode 100644 index 0000000..c36c29d --- /dev/null +++ b/databusclient/workflow/steps.py @@ -0,0 +1,150 @@ +"""Step classes — adapt a workflow step config into a call to the existing +download()/deploy()/delete() API functions. + +Each step class resolves any ${steps.name.key} references in its config via +StepContext, calls the existing, unmodified API function, and records its +output back into StepContext so later steps can reference it. + +No new business logic lives here. Steps are thin adapters only. +""" + +from __future__ import annotations +import os +from typing import Any, Dict + +from databusclient.api.delete import delete as api_delete +from databusclient.api.deploy import create_dataset, deploy as api_deploy_call +from databusclient.api.download import download as api_download + +from databusclient.workflow.context import StepContext + + +class StepValidationError(Exception): + """Raised when a step's config is missing a required, command-specific field.""" + + +class DownloadStep: + """Adapts a workflow step to a call to download().""" + + def run(self, step_config: Dict[str, Any], context: StepContext) -> None: + resolved = context.resolve(step_config) + name = resolved["name"] + + uri = resolved.get("uri") + if not uri: + raise StepValidationError(f"Step '{name}': download step requires 'uri'.") + + local_dir = resolved.get("localdir") + if local_dir is None: + local_dir = os.path.join(os.getcwd(), ".workflow", name) + + api_download( + localDir=local_dir, + endpoint=resolved.get("databus"), + databusURIs=[uri], + token=resolved.get("vault_token"), + databus_key=resolved.get("databus_key"), + all_versions=resolved.get("all_versions", False), + compression=resolved.get("convert_to") or resolved.get("compression"), + convert_format=resolved.get("format"), + graph_name=resolved.get("graph_name"), + base_uri=resolved.get("base_uri"), + validate_checksum=resolved.get("validate_checksum", False), + manifest_context=context.manifest_context, + ) + + output_files = self._collect_output_files(local_dir) + context.set_output(name, "output_files", output_files) + context.set_output(name, "output_urls", [uri]) + + @staticmethod + def _collect_output_files(local_dir: str) -> list: + """Walk local_dir and return all file paths produced by the download. + + Always returns a flat list of file paths, even if the download + produced files nested in subdirectories (e.g. a Quad -> Triple + split, which writes multiple files into a subdirectory). + """ + if not os.path.isdir(local_dir): + return [] + return sorted( + os.path.join(root, filename) + for root, _dirs, filenames in os.walk(local_dir) + for filename in filenames + ) + + +class DeployStep: + """Adapts a workflow step to a call to create_dataset() + deploy().""" + + def run(self, step_config: Dict[str, Any], context: StepContext) -> None: + resolved = context.resolve(step_config) + name = resolved["name"] + + required = ["version_id", "title", "abstract", "description", "license", "api_key"] + missing = [f for f in required if not resolved.get(f)] + if missing: + raise StepValidationError( + f"Step '{name}': deploy step is missing required field(s): " + f"{', '.join(missing)}." + ) + + files = resolved.get("files") + if not files: + raise StepValidationError(f"Step '{name}': deploy step requires 'files'.") + if isinstance(files, str): + files = [files] + + dataid = create_dataset( + version_id=resolved["version_id"], + artifact_version_title=resolved["title"], + artifact_version_abstract=resolved["abstract"], + artifact_version_description=resolved["description"], + license_url=resolved["license"], + distributions=files, + ) + api_deploy_call(dataid=dataid, api_key=resolved["api_key"]) + + context.set_output(name, "output_files", files) + context.set_output(name, "version_id", resolved["version_id"]) + + +class DeleteStep: + """Adapts a workflow step to a call to delete(). + + Workflows are meant to run unattended -- a delete step never triggers + the interactive confirmation prompt that the plain `delete` CLI command + uses. force is always effectively True here; dry_run must be set + explicitly in the step config if a preview-only run is wanted. + """ + + def run(self, step_config: Dict[str, Any], context: StepContext) -> None: + resolved = context.resolve(step_config) + name = resolved["name"] + + uris = resolved.get("uris") + if not uris: + raise StepValidationError(f"Step '{name}': delete step requires 'uris'.") + if isinstance(uris, str): + uris = [uris] + + api_key = resolved.get("api_key") + if not api_key: + raise StepValidationError(f"Step '{name}': delete step requires 'api_key'.") + + api_delete( + databusURIs=uris, + databus_key=api_key, + dry_run=resolved.get("dry_run", False), + force=True, + manifest_context=context.manifest_context, + ) + + context.set_output(name, "output_files", []) + + +STEP_REGISTRY = { + "download": DownloadStep, + "deploy": DeployStep, + "delete": DeleteStep, +} \ No newline at end of file diff --git a/poetry.lock b/poetry.lock index e3759ff..88e0a8e 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand. [[package]] name = "black" @@ -329,6 +329,89 @@ pluggy = ">=0.12,<2.0" [package.extras] testing = ["argcomplete", "attrs (>=19.2.0)", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"] +[[package]] +name = "pyyaml" +version = "6.0.3" +description = "YAML parser and emitter for Python" +optional = false +python-versions = ">=3.8" +groups = ["main"] +files = [ + {file = "PyYAML-6.0.3-cp38-cp38-macosx_10_13_x86_64.whl", hash = "sha256:c2514fceb77bc5e7a2f7adfaa1feb2fb311607c9cb518dbc378688ec73d8292f"}, + {file = "PyYAML-6.0.3-cp38-cp38-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9c57bb8c96f6d1808c030b1687b9b5fb476abaa47f0db9c0101f5e9f394e97f4"}, + {file = "PyYAML-6.0.3-cp38-cp38-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:efd7b85f94a6f21e4932043973a7ba2613b059c4a000551892ac9f1d11f5baf3"}, + {file = "PyYAML-6.0.3-cp38-cp38-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:22ba7cfcad58ef3ecddc7ed1db3409af68d023b7f940da23c6c2a1890976eda6"}, + {file = "PyYAML-6.0.3-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:6344df0d5755a2c9a276d4473ae6b90647e216ab4757f8426893b5dd2ac3f369"}, + {file = "PyYAML-6.0.3-cp38-cp38-win32.whl", hash = "sha256:3ff07ec89bae51176c0549bc4c63aa6202991da2d9a6129d7aef7f1407d3f295"}, + {file = "PyYAML-6.0.3-cp38-cp38-win_amd64.whl", hash = "sha256:5cf4e27da7e3fbed4d6c3d8e797387aaad68102272f8f9752883bc32d61cb87b"}, + {file = "pyyaml-6.0.3-cp310-cp310-macosx_10_13_x86_64.whl", hash = "sha256:214ed4befebe12df36bcc8bc2b64b396ca31be9304b8f59e25c11cf94a4c033b"}, + {file = "pyyaml-6.0.3-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:02ea2dfa234451bbb8772601d7b8e426c2bfa197136796224e50e35a78777956"}, + {file = "pyyaml-6.0.3-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b30236e45cf30d2b8e7b3e85881719e98507abed1011bf463a8fa23e9c3e98a8"}, + {file = "pyyaml-6.0.3-cp310-cp310-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:66291b10affd76d76f54fad28e22e51719ef9ba22b29e1d7d03d6777a9174198"}, + {file = "pyyaml-6.0.3-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9c7708761fccb9397fe64bbc0395abcae8c4bf7b0eac081e12b809bf47700d0b"}, + {file = "pyyaml-6.0.3-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:418cf3f2111bc80e0933b2cd8cd04f286338bb88bdc7bc8e6dd775ebde60b5e0"}, + {file = "pyyaml-6.0.3-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:5e0b74767e5f8c593e8c9b5912019159ed0533c70051e9cce3e8b6aa699fcd69"}, + {file = "pyyaml-6.0.3-cp310-cp310-win32.whl", hash = "sha256:28c8d926f98f432f88adc23edf2e6d4921ac26fb084b028c733d01868d19007e"}, + {file = "pyyaml-6.0.3-cp310-cp310-win_amd64.whl", hash = "sha256:bdb2c67c6c1390b63c6ff89f210c8fd09d9a1217a465701eac7316313c915e4c"}, + {file = "pyyaml-6.0.3-cp311-cp311-macosx_10_13_x86_64.whl", hash = "sha256:44edc647873928551a01e7a563d7452ccdebee747728c1080d881d68af7b997e"}, + {file = "pyyaml-6.0.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:652cb6edd41e718550aad172851962662ff2681490a8a711af6a4d288dd96824"}, + {file = "pyyaml-6.0.3-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:10892704fc220243f5305762e276552a0395f7beb4dbf9b14ec8fd43b57f126c"}, + {file = "pyyaml-6.0.3-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:850774a7879607d3a6f50d36d04f00ee69e7fc816450e5f7e58d7f17f1ae5c00"}, + {file = "pyyaml-6.0.3-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b8bb0864c5a28024fac8a632c443c87c5aa6f215c0b126c449ae1a150412f31d"}, + {file = "pyyaml-6.0.3-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:1d37d57ad971609cf3c53ba6a7e365e40660e3be0e5175fa9f2365a379d6095a"}, + {file = "pyyaml-6.0.3-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:37503bfbfc9d2c40b344d06b2199cf0e96e97957ab1c1b546fd4f87e53e5d3e4"}, + {file = "pyyaml-6.0.3-cp311-cp311-win32.whl", hash = "sha256:8098f252adfa6c80ab48096053f512f2321f0b998f98150cea9bd23d83e1467b"}, + {file = "pyyaml-6.0.3-cp311-cp311-win_amd64.whl", hash = "sha256:9f3bfb4965eb874431221a3ff3fdcddc7e74e3b07799e0e84ca4a0f867d449bf"}, + {file = "pyyaml-6.0.3-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:7f047e29dcae44602496db43be01ad42fc6f1cc0d8cd6c83d342306c32270196"}, + {file = "pyyaml-6.0.3-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:fc09d0aa354569bc501d4e787133afc08552722d3ab34836a80547331bb5d4a0"}, + {file = "pyyaml-6.0.3-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9149cad251584d5fb4981be1ecde53a1ca46c891a79788c0df828d2f166bda28"}, + {file = "pyyaml-6.0.3-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:5fdec68f91a0c6739b380c83b951e2c72ac0197ace422360e6d5a959d8d97b2c"}, + {file = "pyyaml-6.0.3-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ba1cc08a7ccde2d2ec775841541641e4548226580ab850948cbfda66a1befcdc"}, + {file = "pyyaml-6.0.3-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:8dc52c23056b9ddd46818a57b78404882310fb473d63f17b07d5c40421e47f8e"}, + {file = "pyyaml-6.0.3-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:41715c910c881bc081f1e8872880d3c650acf13dfa8214bad49ed4cede7c34ea"}, + {file = "pyyaml-6.0.3-cp312-cp312-win32.whl", hash = "sha256:96b533f0e99f6579b3d4d4995707cf36df9100d67e0c8303a0c55b27b5f99bc5"}, + {file = "pyyaml-6.0.3-cp312-cp312-win_amd64.whl", hash = "sha256:5fcd34e47f6e0b794d17de1b4ff496c00986e1c83f7ab2fb8fcfe9616ff7477b"}, + {file = "pyyaml-6.0.3-cp312-cp312-win_arm64.whl", hash = "sha256:64386e5e707d03a7e172c0701abfb7e10f0fb753ee1d773128192742712a98fd"}, + {file = "pyyaml-6.0.3-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:8da9669d359f02c0b91ccc01cac4a67f16afec0dac22c2ad09f46bee0697eba8"}, + {file = "pyyaml-6.0.3-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:2283a07e2c21a2aa78d9c4442724ec1eb15f5e42a723b99cb3d822d48f5f7ad1"}, + {file = "pyyaml-6.0.3-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ee2922902c45ae8ccada2c5b501ab86c36525b883eff4255313a253a3160861c"}, + {file = "pyyaml-6.0.3-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a33284e20b78bd4a18c8c2282d549d10bc8408a2a7ff57653c0cf0b9be0afce5"}, + {file = "pyyaml-6.0.3-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0f29edc409a6392443abf94b9cf89ce99889a1dd5376d94316ae5145dfedd5d6"}, + {file = "pyyaml-6.0.3-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:f7057c9a337546edc7973c0d3ba84ddcdf0daa14533c2065749c9075001090e6"}, + {file = "pyyaml-6.0.3-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:eda16858a3cab07b80edaf74336ece1f986ba330fdb8ee0d6c0d68fe82bc96be"}, + {file = "pyyaml-6.0.3-cp313-cp313-win32.whl", hash = "sha256:d0eae10f8159e8fdad514efdc92d74fd8d682c933a6dd088030f3834bc8e6b26"}, + {file = "pyyaml-6.0.3-cp313-cp313-win_amd64.whl", hash = "sha256:79005a0d97d5ddabfeeea4cf676af11e647e41d81c9a7722a193022accdb6b7c"}, + {file = "pyyaml-6.0.3-cp313-cp313-win_arm64.whl", hash = "sha256:5498cd1645aa724a7c71c8f378eb29ebe23da2fc0d7a08071d89469bf1d2defb"}, + {file = "pyyaml-6.0.3-cp314-cp314-macosx_10_13_x86_64.whl", hash = "sha256:8d1fab6bb153a416f9aeb4b8763bc0f22a5586065f86f7664fc23339fc1c1fac"}, + {file = "pyyaml-6.0.3-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:34d5fcd24b8445fadc33f9cf348c1047101756fd760b4dacb5c3e99755703310"}, + {file = "pyyaml-6.0.3-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:501a031947e3a9025ed4405a168e6ef5ae3126c59f90ce0cd6f2bfc477be31b7"}, + {file = "pyyaml-6.0.3-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:b3bc83488de33889877a0f2543ade9f70c67d66d9ebb4ac959502e12de895788"}, + {file = "pyyaml-6.0.3-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c458b6d084f9b935061bc36216e8a69a7e293a2f1e68bf956dcd9e6cbcd143f5"}, + {file = "pyyaml-6.0.3-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7c6610def4f163542a622a73fb39f534f8c101d690126992300bf3207eab9764"}, + {file = "pyyaml-6.0.3-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:5190d403f121660ce8d1d2c1bb2ef1bd05b5f68533fc5c2ea899bd15f4399b35"}, + {file = "pyyaml-6.0.3-cp314-cp314-win_amd64.whl", hash = "sha256:4a2e8cebe2ff6ab7d1050ecd59c25d4c8bd7e6f400f5f82b96557ac0abafd0ac"}, + {file = "pyyaml-6.0.3-cp314-cp314-win_arm64.whl", hash = "sha256:93dda82c9c22deb0a405ea4dc5f2d0cda384168e466364dec6255b293923b2f3"}, + {file = "pyyaml-6.0.3-cp314-cp314t-macosx_10_13_x86_64.whl", hash = "sha256:02893d100e99e03eda1c8fd5c441d8c60103fd175728e23e431db1b589cf5ab3"}, + {file = "pyyaml-6.0.3-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:c1ff362665ae507275af2853520967820d9124984e0f7466736aea23d8611fba"}, + {file = "pyyaml-6.0.3-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6adc77889b628398debc7b65c073bcb99c4a0237b248cacaf3fe8a557563ef6c"}, + {file = "pyyaml-6.0.3-cp314-cp314t-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a80cb027f6b349846a3bf6d73b5e95e782175e52f22108cfa17876aaeff93702"}, + {file = "pyyaml-6.0.3-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:00c4bdeba853cc34e7dd471f16b4114f4162dc03e6b7afcc2128711f0eca823c"}, + {file = "pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:66e1674c3ef6f541c35191caae2d429b967b99e02040f5ba928632d9a7f0f065"}, + {file = "pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:16249ee61e95f858e83976573de0f5b2893b3677ba71c9dd36b9cf8be9ac6d65"}, + {file = "pyyaml-6.0.3-cp314-cp314t-win_amd64.whl", hash = "sha256:4ad1906908f2f5ae4e5a8ddfce73c320c2a1429ec52eafd27138b7f1cbe341c9"}, + {file = "pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b"}, + {file = "pyyaml-6.0.3-cp39-cp39-macosx_10_13_x86_64.whl", hash = "sha256:b865addae83924361678b652338317d1bd7e79b1f4596f96b96c77a5a34b34da"}, + {file = "pyyaml-6.0.3-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:c3355370a2c156cffb25e876646f149d5d68f5e0a3ce86a5084dd0b64a994917"}, + {file = "pyyaml-6.0.3-cp39-cp39-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3c5677e12444c15717b902a5798264fa7909e41153cdf9ef7ad571b704a63dd9"}, + {file = "pyyaml-6.0.3-cp39-cp39-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:5ed875a24292240029e4483f9d4a4b8a1ae08843b9c54f43fcc11e404532a8a5"}, + {file = "pyyaml-6.0.3-cp39-cp39-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0150219816b6a1fa26fb4699fb7daa9caf09eb1999f3b70fb6e786805e80375a"}, + {file = "pyyaml-6.0.3-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:fa160448684b4e94d80416c0fa4aac48967a969efe22931448d853ada8baf926"}, + {file = "pyyaml-6.0.3-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:27c0abcb4a5dac13684a37f76e701e054692a9b2d3064b70f5e4eb54810553d7"}, + {file = "pyyaml-6.0.3-cp39-cp39-win32.whl", hash = "sha256:1ebe39cb5fc479422b83de611d14e2c0d3bb2a18bbcb01f229ab3cfbd8fee7a0"}, + {file = "pyyaml-6.0.3-cp39-cp39-win_amd64.whl", hash = "sha256:2e71d11abed7344e42a8849600193d15b6def118602c4c176f748e4583246007"}, + {file = "pyyaml-6.0.3.tar.gz", hash = "sha256:d76623373421df22fb4cf8817020cbb7ef15c725b9d5e45f17e189bfc384190f"}, +] + [[package]] name = "rdflib" version = "7.5.0" @@ -466,4 +549,4 @@ zstd = ["backports-zstd (>=1.0.0) ; python_version < \"3.14\""] [metadata] lock-version = "2.1" python-versions = "^3.11" -content-hash = "f625db7ea6714ebf87336efecaef03ec2dc4f6f7838c3239432828cd6649ff96" +content-hash = "b738c415f513b772068e55993bbaf06c4b8db37a77f5303e21b9498636b7c91b" diff --git a/pyproject.toml b/pyproject.toml index e1485ae..d1cdc47 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ requests = "^2.28.1" tqdm = "^4.42.1" SPARQLWrapper = "^2.0.0" rdflib = "^7.2.1" +pyyaml = "^6.0.3" [tool.poetry.group.dev.dependencies] black = "^22.6.0" diff --git a/tests/test_step_context.py b/tests/test_step_context.py new file mode 100644 index 0000000..a90f62e --- /dev/null +++ b/tests/test_step_context.py @@ -0,0 +1,75 @@ +"""Tests for StepContext (Milestone 4).""" + +import pytest + +from databusclient.workflow.context import StepContext, StepReferenceError + + +def test_set_and_get_output(): + ctx = StepContext() + ctx.set_output("fetch", "output_files", ["/data/a.ttl", "/data/b.ttl"]) + assert ctx.get_output("fetch", "output_files") == ["/data/a.ttl", "/data/b.ttl"] + + +def test_get_output_unknown_step_raises(): + ctx = StepContext() + with pytest.raises(StepReferenceError, match="unknown or not-yet-executed"): + ctx.get_output("nope", "output_files") + + +def test_get_output_unknown_key_raises(): + ctx = StepContext() + ctx.set_output("fetch", "output_files", ["/data/a.ttl"]) + with pytest.raises(StepReferenceError, match="no recorded output"): + ctx.get_output("fetch", "some_other_key") + + +def test_resolve_exact_token_preserves_list_type(): + """A value that IS exactly one ${steps.x.y} token resolves to the raw list.""" + ctx = StepContext() + ctx.set_output("fetch", "output_files", ["/data/a.ttl", "/data/b.ttl"]) + resolved = ctx.resolve("${steps.fetch.output_files}") + assert resolved == ["/data/a.ttl", "/data/b.ttl"] + assert isinstance(resolved, list) + + +def test_resolve_embedded_token_in_string(): + ctx = StepContext() + ctx.set_output("fetch", "version", "2024.01") + resolved = ctx.resolve("Deployed version ${steps.fetch.version}") + assert resolved == "Deployed version 2024.01" + + +def test_resolve_nested_dict_and_list(): + ctx = StepContext() + ctx.set_output("fetch", "output_files", ["/data/a.ttl"]) + resolved = ctx.resolve({ + "files": "${steps.fetch.output_files}", + "meta": {"note": "from ${steps.fetch.output_files}"}, + }) + assert resolved["files"] == ["/data/a.ttl"] + assert resolved["meta"]["note"] == "from ['/data/a.ttl']" + + +def test_resolve_plain_value_passthrough(): + ctx = StepContext() + assert ctx.resolve("no tokens here") == "no tokens here" + assert ctx.resolve(42) == 42 + assert ctx.resolve(None) is None + + +def test_resolve_unresolvable_reference_raises(): + ctx = StepContext() + with pytest.raises(StepReferenceError): + ctx.resolve("${steps.never_ran.output_files}") + + +def test_manifest_context_defaults_to_none(): + ctx = StepContext() + assert ctx.manifest_context is None + + +def test_manifest_context_stored_when_provided(): + sentinel = object() + ctx = StepContext(manifest_context=sentinel) + assert ctx.manifest_context is sentinel \ No newline at end of file diff --git a/tests/test_workflow_engine.py b/tests/test_workflow_engine.py new file mode 100644 index 0000000..71005a4 --- /dev/null +++ b/tests/test_workflow_engine.py @@ -0,0 +1,136 @@ +"""Tests for WorkflowEngine (Milestone 4).""" + +import pytest + +from databusclient.workflow.engine import WorkflowEngine, WorkflowExecutionError + +def test_runs_steps_in_order(monkeypatch): + order = [] + + class OrderedStep: + def run(self, step_config, context): + order.append(step_config["name"]) + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", OrderedStep) + monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", OrderedStep) + + engine = WorkflowEngine() + engine.run([ + {"name": "a", "command": "download"}, + {"name": "b", "command": "deploy"}, + ]) + assert order == ["a", "b"] + + +def test_step_failure_with_default_fail_raises(monkeypatch): + class FailingStep: + def run(self, step_config, context): + raise RuntimeError("boom") + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FailingStep) + + engine = WorkflowEngine() + with pytest.raises(WorkflowExecutionError, match="boom"): + engine.run([{"name": "a", "command": "download"}]) + + +def test_step_failure_with_continue_does_not_raise(monkeypatch): + class FailingStep: + def run(self, step_config, context): + raise RuntimeError("boom") + + class OKStep: + def run(self, step_config, context): + pass + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FailingStep) + monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", OKStep) + + engine = WorkflowEngine() + results = engine.run([ + {"name": "a", "command": "download", "on_error": "continue"}, + {"name": "b", "command": "deploy"}, + ]) + assert results[0].status == "skipped_error" + assert results[1].status == "success" + + +def test_retry_succeeds_on_second_attempt(monkeypatch): + attempts = {"count": 0} + + class FlakyStep: + def run(self, step_config, context): + attempts["count"] += 1 + if attempts["count"] < 2: + raise RuntimeError("transient failure") + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FlakyStep) + + engine = WorkflowEngine() + results = engine.run([{ + "name": "a", "command": "download", "on_error": "retry", + "retry": {"max_attempts": 3, "delay_seconds": 0}, + }]) + assert results[0].status == "success" + assert results[0].attempts == 2 + assert attempts["count"] == 2 + + +def test_retry_exhausts_attempts_and_fails(monkeypatch): + class AlwaysFailsStep: + def run(self, step_config, context): + raise RuntimeError("permanent failure") + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", AlwaysFailsStep) + + engine = WorkflowEngine() + with pytest.raises(WorkflowExecutionError, match="permanent failure"): + engine.run([{ + "name": "a", "command": "download", "on_error": "retry", + "retry": {"max_attempts": 2, "delay_seconds": 0}, + }]) + + +def test_step_chaining_end_to_end(monkeypatch): + """A download step's output is available to a deploy step via StepContext.""" + class FetchStep: + def run(self, step_config, context): + context.set_output(step_config["name"], "output_files", ["/data/a.ttl"]) + + captured = {} + + class PublishStep: + def run(self, step_config, context): + resolved = context.resolve(step_config) + captured["files"] = resolved["files"] + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FetchStep) + monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", PublishStep) + + engine = WorkflowEngine() + engine.run([ + {"name": "fetch", "command": "download"}, + {"name": "publish", "command": "deploy", "files": "${steps.fetch.output_files}"}, + ]) + assert captured["files"] == ["/data/a.ttl"] + +def test_unknown_step_reference_surfaces_as_workflow_execution_error(monkeypatch): + class PublishStep: + def run(self, step_config, context): + context.resolve(step_config) # will raise, since "nonexistent" never ran + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", PublishStep) + + engine = WorkflowEngine() + with pytest.raises(WorkflowExecutionError, match="unknown or not-yet-executed"): + engine.run([ + {"name": "publish", "command": "deploy", + "files": "${steps.nonexistent.output_files}"}, + ]) \ No newline at end of file diff --git a/tests/test_workflow_parser.py b/tests/test_workflow_parser.py new file mode 100644 index 0000000..b82bab7 --- /dev/null +++ b/tests/test_workflow_parser.py @@ -0,0 +1,180 @@ +"""Tests for WorkflowParser (Milestone 4).""" + +import os +import tempfile + +import pytest +import yaml + +from databusclient.workflow.parser import ( + MissingEnvVarError, + WorkflowParseError, + parse_workflow, +) + + +def _write_yaml(content: dict) -> str: + fd, path = tempfile.mkstemp(suffix=".yml") + with os.fdopen(fd, "w", encoding="utf-8") as f: + yaml.safe_dump(content, f) + return path + + +def test_parses_minimal_valid_workflow(): + path = _write_yaml({ + "steps": [ + {"name": "fetch", "command": "download", "uri": "https://example.org/x"}, + ] + }) + result = parse_workflow(path) + assert result["manifest"] is None + assert len(result["steps"]) == 1 + assert result["steps"][0]["name"] == "fetch" + + +def test_missing_steps_key_raises(): + path = _write_yaml({"manifest": "run.json"}) + with pytest.raises(WorkflowParseError, match="steps"): + parse_workflow(path) + + +def test_empty_steps_list_raises(): + path = _write_yaml({"steps": []}) + with pytest.raises(WorkflowParseError, match="non-empty"): + parse_workflow(path) + + +def test_step_missing_name_raises(): + path = _write_yaml({"steps": [{"command": "download", "uri": "x"}]}) + with pytest.raises(WorkflowParseError, match="name"): + parse_workflow(path) + + +def test_duplicate_step_names_raise(): + path = _write_yaml({ + "steps": [ + {"name": "a", "command": "download", "uri": "x"}, + {"name": "a", "command": "delete", "uris": ["x"]}, + ] + }) + with pytest.raises(WorkflowParseError, match="Duplicate step name"): + parse_workflow(path) + + +def test_invalid_command_raises(): + path = _write_yaml({"steps": [{"name": "a", "command": "bogus"}]}) + with pytest.raises(WorkflowParseError, match="invalid command"): + parse_workflow(path) + + +def test_invalid_on_error_raises(): + path = _write_yaml({ + "steps": [{"name": "a", "command": "download", "uri": "x", "on_error": "maybe"}] + }) + with pytest.raises(WorkflowParseError, match="invalid on_error"): + parse_workflow(path) + + +def test_retry_without_config_raises(): + path = _write_yaml({ + "steps": [{"name": "a", "command": "download", "uri": "x", "on_error": "retry"}] + }) + with pytest.raises(WorkflowParseError, match="retry"): + parse_workflow(path) + + +def test_retry_with_invalid_max_attempts_raises(): + path = _write_yaml({ + "steps": [{ + "name": "a", "command": "download", "uri": "x", "on_error": "retry", + "retry": {"max_attempts": 0, "delay_seconds": 5}, + }] + }) + with pytest.raises(WorkflowParseError, match="max_attempts"): + parse_workflow(path) + + +def test_valid_retry_config_passes(monkeypatch): + path = _write_yaml({ + "steps": [{ + "name": "a", "command": "download", "uri": "x", "on_error": "retry", + "retry": {"max_attempts": 3, "delay_seconds": 5}, + }] + }) + result = parse_workflow(path) + assert result["steps"][0]["retry"]["max_attempts"] == 3 + + +def test_env_var_substitution(monkeypatch): + monkeypatch.setenv("MY_API_KEY", "secret123") + path = _write_yaml({ + "steps": [{"name": "a", "command": "deploy", "api_key": "${MY_API_KEY}"}] + }) + result = parse_workflow(path) + assert result["steps"][0]["api_key"] == "secret123" + + +def test_missing_env_var_raises(monkeypatch): + monkeypatch.delenv("DOES_NOT_EXIST_VAR", raising=False) + path = _write_yaml({ + "steps": [{"name": "a", "command": "deploy", "api_key": "${DOES_NOT_EXIST_VAR}"}] + }) + with pytest.raises(MissingEnvVarError, match="DOES_NOT_EXIST_VAR"): + parse_workflow(path) + + +def test_steps_reference_left_untouched(): + """${steps.x.output_files} must NOT be treated as a missing env var.""" + path = _write_yaml({ + "steps": [ + {"name": "fetch", "command": "download", "uri": "x"}, + {"name": "publish", "command": "deploy", "files": "${steps.fetch.output_files}"}, + ] + }) + result = parse_workflow(path) + assert result["steps"][1]["files"] == "${steps.fetch.output_files}" + + +def test_multiple_tokens_in_same_string(monkeypatch): + monkeypatch.setenv("ACCOUNT", "myaccount") + monkeypatch.setenv("GROUP", "mygroup") + path = _write_yaml({ + "steps": [{ + "name": "a", "command": "deploy", + "version_id": "https://databus.dbpedia.org/${ACCOUNT}/${GROUP}/art/1.0", + }] + }) + result = parse_workflow(path) + assert result["steps"][0]["version_id"] == "https://databus.dbpedia.org/myaccount/mygroup/art/1.0" + + +def test_nonexistent_file_raises(): + with pytest.raises(WorkflowParseError, match="not found"): + parse_workflow("does-not-exist.yml") + + +def test_invalid_yaml_raises(): + fd, path = tempfile.mkstemp(suffix=".yml") + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write("steps: [unclosed") + with pytest.raises(WorkflowParseError, match="not valid YAML"): + parse_workflow(path) + +def test_bare_dollar_var_without_braces_passes_through_unchanged(): + """$VAR (no braces) is not a substitution token -- left as literal text.""" + path = _write_yaml({ + "steps": [{"name": "a", "command": "download", "uri": "$HOME/data"}] + }) + result = parse_workflow(path) + assert result["steps"][0]["uri"] == "$HOME/data" + + +def test_empty_braces_pass_through_unchanged(): + """${} has no characters between the braces, so it doesn't match the + substitution pattern at all (which requires at least one character) -- + it passes through as literal text, same as a bare $VAR without braces.""" + path = _write_yaml({ + "steps": [{"name": "a", "command": "download", "uri": "${}/data"}] + }) + result = parse_workflow(path) + assert result["steps"][0]["uri"] == "${}/data" \ No newline at end of file diff --git a/tests/test_workflow_steps.py b/tests/test_workflow_steps.py new file mode 100644 index 0000000..8d94359 --- /dev/null +++ b/tests/test_workflow_steps.py @@ -0,0 +1,159 @@ +"""Tests for step classes (Milestone 4). No live Databus calls -- the +underlying api_download/api_deploy_call/api_delete functions are mocked.""" + +import os +import pytest + +from databusclient.workflow.context import StepContext +from databusclient.workflow.steps import ( + DeleteStep, + DeployStep, + DownloadStep, + StepValidationError, +) + + +def test_download_step_requires_uri(): + ctx = StepContext() + step = DownloadStep() + with pytest.raises(StepValidationError, match="requires 'uri'"): + step.run({"name": "fetch", "command": "download"}, ctx) + + +def test_download_step_calls_api_download_and_collects_files(monkeypatch, tmp_path): + captured = {} + + def fake_download(**kwargs): + captured.update(kwargs) + local_dir = kwargs["localDir"] + os.makedirs(local_dir, exist_ok=True) + with open(os.path.join(local_dir, "a.ttl"), "w") as f: + f.write("data") + + monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download) + + ctx = StepContext() + step = DownloadStep() + step.run( + {"name": "fetch", "command": "download", "uri": "https://example.org/x", + "localdir": str(tmp_path)}, + ctx, + ) + + assert captured["databusURIs"] == ["https://example.org/x"] + output = ctx.get_output("fetch", "output_files") + assert len(output) == 1 + assert output[0].endswith("a.ttl") + + +def test_download_step_collects_files_from_subdirectory(monkeypatch, tmp_path): + """Simulates a Quad -> Triple split producing files in a subdirectory.""" + def fake_download(**kwargs): + local_dir = kwargs["localDir"] + sub = os.path.join(local_dir, "split") + os.makedirs(sub, exist_ok=True) + with open(os.path.join(sub, "graph1.nt"), "w") as f: + f.write("data") + with open(os.path.join(sub, "graph2.nt"), "w") as f: + f.write("data") + + monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download) + + ctx = StepContext() + step = DownloadStep() + step.run( + {"name": "fetch", "command": "download", "uri": "x", "localdir": str(tmp_path)}, + ctx, + ) + output = ctx.get_output("fetch", "output_files") + assert len(output) == 2 + assert all(isinstance(p, str) for p in output) + + +def test_deploy_step_requires_fields(): + ctx = StepContext() + step = DeployStep() + with pytest.raises(StepValidationError, match="missing required field"): + step.run({"name": "publish", "command": "deploy"}, ctx) + + +def test_deploy_step_resolves_step_reference_and_calls_deploy(monkeypatch): + captured = {} + + def fake_create_dataset(**kwargs): + captured["create_dataset_kwargs"] = kwargs + return {"@graph": [{"@id": "fake"}]} + + def fake_deploy(dataid, api_key): + captured["api_key"] = api_key + + monkeypatch.setattr("databusclient.workflow.steps.create_dataset", fake_create_dataset) + monkeypatch.setattr("databusclient.workflow.steps.api_deploy_call", fake_deploy) + + ctx = StepContext() + ctx.set_output("fetch", "output_files", ["/data/a.ttl"]) + + step = DeployStep() + step.run({ + "name": "publish", "command": "deploy", + "version_id": "https://databus.dbpedia.org/a/b/c/1.0", + "title": "T", "abstract": "A", "description": "D", + "license": "https://license.example.org", "api_key": "key123", + "files": "${steps.fetch.output_files}", + }, ctx) + + assert captured["create_dataset_kwargs"]["distributions"] == ["/data/a.ttl"] + assert captured["api_key"] == "key123" + + +def test_delete_step_always_forces_no_prompt(monkeypatch): + captured = {} + + def fake_delete(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr("databusclient.workflow.steps.api_delete", fake_delete) + + ctx = StepContext() + step = DeleteStep() + step.run({ + "name": "cleanup", "command": "delete", + "uris": ["https://databus.dbpedia.org/a/b/c/old"], + "api_key": "key123", + }, ctx) + + assert captured["force"] is True + assert captured["dry_run"] is False + + +def test_delete_step_requires_uris(): + ctx = StepContext() + step = DeleteStep() + with pytest.raises(StepValidationError, match="requires 'uris'"): + step.run({"name": "cleanup", "command": "delete", "api_key": "k"}, ctx) + + +def test_delete_step_requires_api_key(): + ctx = StepContext() + step = DeleteStep() + with pytest.raises(StepValidationError, match="requires 'api_key'"): + step.run({"name": "cleanup", "command": "delete", "uris": ["x"]}, ctx) + +def test_download_step_records_output_urls(monkeypatch, tmp_path): + def fake_download(**kwargs): + local_dir = kwargs["localDir"] + os.makedirs(local_dir, exist_ok=True) + with open(os.path.join(local_dir, "a.ttl"), "w") as f: + f.write("data") + + monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download) + + ctx = StepContext() + step = DownloadStep() + step.run( + {"name": "fetch", "command": "download", "uri": "https://example.org/data/a.ttl", + "localdir": str(tmp_path)}, + ctx, + ) + + assert ctx.get_output("fetch", "output_urls") == ["https://example.org/data/a.ttl"] \ No newline at end of file From 08b6768fcc9be1d86c6fdabd36e3a234d40bca05 Mon Sep 17 00:00:00 2001 From: DhanashreePetare Date: Thu, 30 Jul 2026 16:17:41 +0530 Subject: [PATCH 2/2] docs: add example workflow YAML files --- .gitignore | 2 +- examples/workflows/download-delete.yml | 11 +++++++++++ examples/workflows/download-deploy.yml | 16 ++++++++++++++++ examples/workflows/full-pipeline.yml | 23 +++++++++++++++++++++++ 4 files changed, 51 insertions(+), 1 deletion(-) create mode 100644 examples/workflows/download-delete.yml create mode 100644 examples/workflows/download-deploy.yml create mode 100644 examples/workflows/full-pipeline.yml diff --git a/.gitignore b/.gitignore index 1e3c183..0401074 100644 --- a/.gitignore +++ b/.gitignore @@ -169,4 +169,4 @@ cython_debug/ # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. .idea/ - +workflow-output/ \ No newline at end of file diff --git a/examples/workflows/download-delete.yml b/examples/workflows/download-delete.yml new file mode 100644 index 0000000..ccf6ae1 --- /dev/null +++ b/examples/workflows/download-delete.yml @@ -0,0 +1,11 @@ +steps: + - name: fetch_dataset + command: download + uri: https://raw.githubusercontent.com/dbpedia/databus/master/server/app/api/swagger.yml + localdir: ./workflow-output/download-delete + + - name: cleanup_old + command: delete + uris: + - https://databus.dbpedia.org/DhanashreeP/test-group/deleteme/1.0 + api_key: ${DATABUS_API_KEY} \ No newline at end of file diff --git a/examples/workflows/download-deploy.yml b/examples/workflows/download-deploy.yml new file mode 100644 index 0000000..55fb32d --- /dev/null +++ b/examples/workflows/download-deploy.yml @@ -0,0 +1,16 @@ +steps: + - name: fetch_dataset + command: download + uri: https://raw.githubusercontent.com/dbpedia/databus/master/server/app/api/swagger.yml + localdir: ./workflow-output/download-deploy + + - name: publish_dataset + command: deploy + version_id: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-demo/2.0 + title: "Workflow Demo - Download and Deploy" + abstract: "throwaway, testing workflow deploy chaining" + description: "throwaway version testing deploy chaining via output_urls" + license: https://creativecommons.org/licenses/by-sa/3.0/ + api_key: ${DATABUS_API_KEY} + files: ${steps.fetch_dataset.output_urls} + on_error: fail \ No newline at end of file diff --git a/examples/workflows/full-pipeline.yml b/examples/workflows/full-pipeline.yml new file mode 100644 index 0000000..24a570c --- /dev/null +++ b/examples/workflows/full-pipeline.yml @@ -0,0 +1,23 @@ +steps: + - name: fetch_dataset + command: download + uri: https://raw.githubusercontent.com/dbpedia/databus/master/server/app/api/swagger.yml + localdir: ./workflow-output/full-pipeline + + - name: publish_dataset + command: deploy + version_id: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-full/1.0 + title: "Workflow Demo - Full Pipeline" + abstract: "throwaway, testing full download-deploy-delete pipeline" + description: "throwaway version testing all three commands chained together" + license: https://creativecommons.org/licenses/by-sa/3.0/ + api_key: 53a899fe-1c7f-4426-8cbf-3a7c9fc15405 + files: ${steps.fetch_dataset.output_urls} + on_error: fail + + - name: cleanup_previous_version + command: delete + uris: + - https://databus.dbpedia.org/DhanashreeP/test-group/workflow-full/0.9 + api_key: 53a899fe-1c7f-4426-8cbf-3a7c9fc15405 + on_error: continue \ No newline at end of file