diff --git a/databusclient/api/download.py b/databusclient/api/download.py index caac28f..d4a5dfd 100644 --- a/databusclient/api/download.py +++ b/databusclient/api/download.py @@ -517,6 +517,12 @@ def _download_file( except requests.exceptions.HTTPError as e: if response.status_code == 404: print(f"WARNING: Skipping file {url} because it was not found (404).") + if manifest_context is not None: + manifest_context.record_file( + url=url, + status="failed", + error_message="404 Not Found", + ) return else: raise e diff --git a/databusclient/cli.py b/databusclient/cli.py index 92989c4..ca3e71f 100644 --- a/databusclient/cli.py +++ b/databusclient/cli.py @@ -583,30 +583,67 @@ def workflow(): @workflow.command("run") @click.argument("workflow_path", type=click.Path(exists=True, dir_okay=False)) -def workflow_run(workflow_path): +@click.option( + "--manifest", + "manifest_path", + default=None, + help="Write a unified JSON-LD manifest of the entire workflow run to PATH.", +) +def workflow_run(workflow_path, manifest_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). + step's on_error behavior (fail/continue/retry). Prints a console + summary after every run. Use --manifest to also write a unified + JSON-LD manifest covering every step. """ try: parsed = parse_workflow(workflow_path) except WorkflowParseError as e: raise click.ClickException(str(e)) - context = StepContext() - engine = WorkflowEngine(context=context) + # CLI flag takes priority; falls back to the YAML file's own + # top-level 'manifest:' key if --manifest was not given on the + # command line. + if manifest_path is None: + manifest_path = parsed.get("manifest") + # A workflow-level manifest is always built internally (for the + # automatic console summary), even when no manifest path is set. + # It's only written to disk when a path is provided (via --manifest + # or the YAML file's own 'manifest:' key). + manifest_ctx = ManifestContext(command="workflow") + + step_context = StepContext() + engine = WorkflowEngine(context=step_context, manifest_context=manifest_ctx) + + workflow_error = None try: - results = engine.run(parsed["steps"]) + engine.run(parsed["steps"]) except WorkflowExecutionError as e: - raise click.ClickException(str(e)) + workflow_error = e + finally: + click.echo("Workflow complete." if workflow_error is None else "Workflow failed.") + for result in engine.results: + click.echo(f" {result.name}: {result.status}") + + click.echo("") + click.echo(format_summary(ManifestWriter.build_manifest_dict(manifest_ctx))) + + if manifest_path: + try: + actual_path = ManifestWriter.write(manifest_ctx, manifest_path) + click.echo(f"\nManifest written to {actual_path}") + except (OSError, IOError) as e: + click.echo( + f"WARNING: Manifest could not be written to {manifest_path}: {e}", + err=True, + ) - click.echo("Workflow complete.") - for result in results: - click.echo(f" {result.name}: {result.status}") + if workflow_error is not None: + raise click.ClickException(str(workflow_error)) if __name__ == "__main__": app() diff --git a/databusclient/manifest/context.py b/databusclient/manifest/context.py index 790f909..2e6cf32 100644 --- a/databusclient/manifest/context.py +++ b/databusclient/manifest/context.py @@ -156,4 +156,25 @@ def summary(self) -> dict: "succeeded": succeeded, "failed": failed, "total_bytes": total_bytes, - } \ No newline at end of file + } + + def merge_from(self, other: "ManifestContext", step_name: Optional[str] = None) -> None: + """Merge another context's recorded files into this one. + + Used by the workflow engine: each step records into its own + temporary ManifestContext (so per-step failures/successes stay + isolated), then that context's entries are merged into the + workflow-level master context here, tagged with which step + produced them. + + Args: + other: The ManifestContext to merge entries from. + step_name: If given, tags each merged file entry with + "step": step_name, so a multi-step workflow manifest + remains traceable to which step produced which file. + """ + for entry in other.files: + merged_entry = dict(entry) + if step_name is not None: + merged_entry["step"] = step_name + self.files.append(merged_entry) \ No newline at end of file diff --git a/databusclient/manifest/replay.py b/databusclient/manifest/replay.py index 0bb135c..f7cede7 100644 --- a/databusclient/manifest/replay.py +++ b/databusclient/manifest/replay.py @@ -294,6 +294,12 @@ def replay_manifest( if not command: raise ManifestReplayError("Manifest missing required field dbus:command.") + if command not in ("download", "delete", "deploy"): + raise ManifestReplayError( + f"Replay for command '{command}' is not implemented yet. " + "Currently supported: download, delete, deploy." + ) + replay_params = _validate_replay_params(manifest.get("dbus:replayParams")) if command == "download": @@ -305,9 +311,4 @@ def replay_manifest( return _replay_delete(replay_params, overrides, confirm_fn) if command == "deploy": - return _replay_deploy(replay_params, overrides) - - raise ManifestReplayError( - f"Replay for command '{command}' is not implemented yet. " - "Currently supported: download, delete, deploy." - ) \ No newline at end of file + return _replay_deploy(replay_params, overrides) \ No newline at end of file diff --git a/databusclient/manifest/summary.py b/databusclient/manifest/summary.py index 914ebfe..a8bd6e0 100644 --- a/databusclient/manifest/summary.py +++ b/databusclient/manifest/summary.py @@ -39,7 +39,7 @@ def format_summary(manifest: Dict[str, Any]) -> str: Args: manifest: A manifest dict, as produced by loading a manifest - JSON-LD file (e.g. via replay._load_manifest). + JSON-LD file (e.g. via replay.load_manifest). Returns: A formatted multi-line string ready to print to the console. @@ -81,4 +81,18 @@ def format_summary(manifest: Dict[str, Any]) -> str: else: lines.append(f"Error : {error_message}") + failed_files = [ + f for f in manifest.get("dataid:distribution", {}).get("dataid:file", []) + if f.get("dbus:status") == "failed" + ] + if failed_files: + lines.append("") + lines.append("Failures:") + for f in failed_files: + step = f.get("dbus:stepName") + url = f.get("dcat:downloadURL", "unknown") + error_message = f.get("dbus:errorMessage", "no error message recorded") + prefix = f" [{step}] " if step else " " + lines.append(f"{prefix}{url}: {error_message}") + return "\n".join(lines) \ No newline at end of file diff --git a/databusclient/manifest/writer.py b/databusclient/manifest/writer.py index 5e606aa..3c0fd59 100644 --- a/databusclient/manifest/writer.py +++ b/databusclient/manifest/writer.py @@ -30,31 +30,14 @@ class ManifestWriter: """Serializes a ManifestContext to a JSON-LD manifest file.""" @staticmethod - def write(context: ManifestContext, path: str) -> str: - """Write the manifest to a JSON-LD file at the given path. - - Creates parent directories if they do not exist. - If a file already exists at `path`, auto-suffixes with _1, _2, etc. - and prints a warning rather than silently overwriting. - On failure, raises OSError — callers should catch and warn. - - Args: - context: The completed ManifestContext to serialize. - path: File path to write the manifest to. - - Raises: - OSError: If the file cannot be written, or if path is a directory. + def build_manifest_dict(context: ManifestContext) -> dict: + """Build the JSON-LD manifest dict from a context, without writing + to disk. Extracted from write() so callers (like the workflow + engine's automatic console summary) can get the dict without + needing a file path. """ - if path.endswith(("/", "\\")) or os.path.isdir(path): - stripped = path.rstrip("/\\") - raise OSError( - f"--manifest path '{path}' is a directory, not a file. " - f"Please provide a full file path, e.g. '{stripped}/manifest.jsonld'." - ) - summary = context.summary() - # Build file entries using DataID vocabulary file_entries = [] for f in context.files: entry: dict = { @@ -79,6 +62,8 @@ def write(context: ManifestContext, path: str) -> str: entry["dbus:errorTraceback"] = f["error_traceback"] if f.get("retry_count"): entry["dbus:retryCount"] = f["retry_count"] + if f.get("step"): + entry["dbus:stepName"] = f["step"] file_entries.append(entry) manifest = { @@ -121,6 +106,33 @@ def write(context: ManifestContext, path: str) -> str: "dbus:errorTraceback": context.operation_error["error_traceback"], } + return manifest + + @staticmethod + def write(context: ManifestContext, path: str) -> str: + """Write the manifest to a JSON-LD file at the given path. + + Creates parent directories if they do not exist. + If a file already exists at `path`, auto-suffixes with _1, _2, etc. + and prints a warning rather than silently overwriting. + On failure, raises OSError — callers should catch and warn. + + Args: + context: The completed ManifestContext to serialize. + path: File path to write the manifest to. + + Raises: + OSError: If the file cannot be written, or if path is a directory. + """ + if path.endswith(("/", "\\")) or os.path.isdir(path): + stripped = path.rstrip("/\\") + raise OSError( + f"--manifest path '{path}' is a directory, not a file. " + f"Please provide a full file path, e.g. '{stripped}/manifest.jsonld'." + ) + + manifest = ManifestWriter.build_manifest_dict(context) + parent = os.path.dirname(os.path.abspath(path)) if parent: os.makedirs(parent, exist_ok=True) diff --git a/databusclient/workflow/engine.py b/databusclient/workflow/engine.py index d5cec54..8a15e9e 100644 --- a/databusclient/workflow/engine.py +++ b/databusclient/workflow/engine.py @@ -10,8 +10,9 @@ from __future__ import annotations import time -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional +from databusclient.manifest.context import ManifestContext from databusclient.workflow.context import StepContext from databusclient.workflow.steps import STEP_REGISTRY @@ -32,10 +33,25 @@ def __init__(self, name: str, status: str, error: Exception | None = None, class WorkflowEngine: - """Runs a parsed workflow's steps in order, handling errors per step.""" - - def __init__(self, context: StepContext | None = None) -> None: + """Runs a parsed workflow's steps in order, handling errors per step. + + If manifest_context is given, one unified manifest is built for the + entire workflow run: each step gets its own temporary ManifestContext + (isolating its recorded files/errors), which is merged into + manifest_context afterward, tagged with the step's name. This lets a + single workflow manifest remain traceable to which step produced or + failed on which file, without touching download.py/deploy.py/delete.py + at all -- those already accept manifest_context=None as a no-op, and + here they simply receive a real (temporary, per-step) one instead. + """ + + def __init__( + self, + context: StepContext | None = None, + manifest_context: Optional[ManifestContext] = None, + ) -> None: self.context = context or StepContext() + self.manifest_context = manifest_context self.results: List[StepResult] = [] def run(self, steps: List[Dict[str, Any]]) -> List[StepResult]: @@ -78,25 +94,75 @@ def _run_step_with_error_handling(self, step_config: Dict[str, Any]) -> StepResu if on_error == "retry": return self._run_with_retry(name, step, step_config) + step_manifest_ctx = self._start_step_manifest(command) try: step.run(step_config, self.context) + self._finish_step_manifest(step_manifest_ctx, name) return StepResult(name, "success") except Exception as exc: + self._finish_step_manifest(step_manifest_ctx, name, error=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 _start_step_manifest(self, command: str) -> Optional[ManifestContext]: + """If a workflow-level manifest is active, give this step its own + temporary ManifestContext to record into. Returns None if no + workflow manifest was requested -- in that case self.context's + manifest_context is left as whatever it already was (e.g. a step's + own throwaway context, like DownloadStep uses for output_urls). + """ + if self.manifest_context is None: + return None + step_ctx = ManifestContext(command=command) + self.context.manifest_context = step_ctx + return step_ctx + + def _finish_step_manifest( + self, + step_manifest_ctx: Optional[ManifestContext], + step_name: str, + error: Optional[Exception] = None, + ) -> None: + """Merge a completed step's temporary manifest entries into the + workflow-level master manifest, tagged with the step name. If the + step failed, also record a synthetic entry so the failure is + visible in the manifest even if the step recorded no per-file + entries before failing. The synthetic entry is tagged with the + same "step" field merge_from() uses, so format_summary()'s + [stepname] prefix mechanism works consistently for BOTH per-file + failures (merged from a step's own context) and whole-step + failures (no file-level detail available at all) -- previously + only the merged case was tagged, so whole-step failures (like an + auth error before any file work happens) showed up without the + [stepname] prefix, relying on the step name being embedded in a + fake url string instead. + """ + if self.manifest_context is None or step_manifest_ctx is None: + return + self.manifest_context.merge_from(step_manifest_ctx, step_name=step_name) + if error is not None: + self.manifest_context.record_file( + url="(no file-level detail -- step failed before producing one)", + status="failed", + error_message=str(error), + ) + self.manifest_context.files[-1]["step"] = step_name + 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"] + command = step_config["command"] last_error: Exception | None = None for attempt in range(1, max_attempts + 1): + step_manifest_ctx = self._start_step_manifest(command) try: step.run(step_config, self.context) + self._finish_step_manifest(step_manifest_ctx, name) return StepResult(name, "success", attempts=attempt) except Exception as exc: last_error = exc @@ -104,6 +170,8 @@ def _run_with_retry(self, name: str, step: Any, step_config: Dict[str, Any]) -> f"WARNING: step '{name}' attempt {attempt}/{max_attempts} " f"failed: {exc}" ) + if attempt == max_attempts: + self._finish_step_manifest(step_manifest_ctx, name, error=exc) if attempt < max_attempts: time.sleep(delay_seconds) diff --git a/databusclient/workflow/steps.py b/databusclient/workflow/steps.py index df35065..de51135 100644 --- a/databusclient/workflow/steps.py +++ b/databusclient/workflow/steps.py @@ -153,6 +153,14 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None: context.set_output(name, "output_files", output_files) context.set_output(name, "version_id", resolved["version_id"]) + # deploy()/deploy_from_metadata() do not accept manifest_context + # (unlike download()/delete()) -- manifest recording for deploy is + # always done manually by the caller. This mirrors exactly what + # cli.py's own `deploy` command does after a successful deploy. + if context.manifest_context is not None: + for url in output_files: + context.manifest_context.record_file(url=url, status="success") + def _run_classic_mode(self, resolved: Dict[str, Any], name: str) -> list: files = resolved.get("files") if not files: diff --git a/examples/workflows/README.md b/examples/workflows/README.md index 5ee0901..ccc7c8f 100644 --- a/examples/workflows/README.md +++ b/examples/workflows/README.md @@ -1,17 +1,28 @@ # Example Workflows -Three example workflow pipelines, each runnable directly, though the deploy/delete steps use paths under a specific Databus account -- swap in your own account/version paths before running them yourself. All three use real, existing Databus data as their download source. +Eight example workflow pipelines, each runnable directly (though the deploy/delete steps use paths under a specific Databus account -- swap in your own account/version paths before running them yourself). All use real, existing Databus data as their download source. ```bash export DATABUS_API_KEY=your-key-here databusclient workflow run download-deploy.yml ``` +## Basic examples + - **`download-deploy.yml`** - downloads a real Databus dataset, then redeploys it exactly as downloaded (classic deploy mode, using `${steps.name.output_urls}` - the actual, redirect-resolved source URL, not the local file). - **`download-delete.yml`** - downloads a real Databus dataset, then deletes that same version, demonstrating a realistic archive-then-delete workflow. -- **`full-pipeline.yml`** - chains all three commands together: download a real Databus dataset, deploy it, then delete that same deployed version, demonstrating a complete download-deploy-cleanup pipeline. +- **`full-pipeline.yml`** - chains all three commands together: download a real Databus dataset, deploy it, then delete that same deployed version. -All three set `api_key: ${DATABUS_API_KEY}` - set that environment variable before running, rather than writing a real key into the file. +## The five proposal use cases (Milestone 5) -See the main [README's Workflow section](../../README.md#cli-workflow) for the full YAML format, step chaining, error handling, and WebDAV deploy mode documentation. +- **`reproducible-research-download.yml`** - downloads a dataset with checksum validation and a saved manifest, so the exact same download can be verified or reproduced later. +- **`nightly-publishing-pipeline.yml`** - download, deploy, then clean up an old version, meant to run unattended (e.g. via cron), with a unified manifest for the whole run. +- **`batch-deployment-with-retry.yml`** - deploys multiple versions in one run, with `on_error: retry` configured on each deploy step to handle transient failures automatically. +- **`ci-cd-integration.yml`** - a workflow with no interactive prompts anywhere, safe to call as a step in a CI/CD pipeline such as GitHub Actions. +- **`failure-debugging.yml`** - intentionally fails a deploy step (invalid API key) to demonstrate what a failed workflow's console output and manifest look like. + +## Manifests +Workflows can write a unified manifest covering every step in two ways: pass `--manifest path.jsonld` on the command line, or set a top-level `manifest:` key inside the YAML file itself (the command-line flag takes priority if both are given). Several of the examples above use the YAML key. Every manifest file entry that came from a workflow step is tagged with `dbus:stepName`, so a multi-step run stays traceable to which step produced or failed on which file. + +See the main [README's Workflow section](../../README.md#cli-workflow) for the full YAML format, step chaining, error handling, and WebDAV deploy mode documentation. diff --git a/examples/workflows/batch-deployment-with-retry.yml b/examples/workflows/batch-deployment-with-retry.yml new file mode 100644 index 0000000..4ad0ca5 --- /dev/null +++ b/examples/workflows/batch-deployment-with-retry.yml @@ -0,0 +1,42 @@ +manifest: ./manifests/batch-deploy.jsonld +steps: + - name: fetch_source + command: download + uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0 + localdir: ./workflow-output/batch + + - name: deploy_batch_1 + command: deploy + version_id: https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-1/1.0 + title: "Batch Deploy Demo 1" + abstract: "Batch deployment with retry example" + description: "First of several deploys in a batch, demonstrating retry on transient failure" + license: https://creativecommons.org/licenses/by-sa/3.0/ + api_key: ${DATABUS_API_KEY} + files: ${steps.fetch_source.output_urls} + on_error: retry + retry: + max_attempts: 3 + delay_seconds: 5 + + - name: deploy_batch_2 + command: deploy + version_id: https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-2/1.0 + title: "Batch Deploy Demo 2" + abstract: "Batch deployment with retry example" + description: "Second of several deploys in a batch, demonstrating retry on transient failure" + license: https://creativecommons.org/licenses/by-sa/3.0/ + api_key: ${DATABUS_API_KEY} + files: ${steps.fetch_source.output_urls} + on_error: retry + retry: + max_attempts: 3 + delay_seconds: 5 + + - name: cleanup_batch + command: delete + uris: + - https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-1/1.0 + - https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-2/1.0 + api_key: ${DATABUS_API_KEY} + on_error: continue diff --git a/examples/workflows/ci-cd-integration.yml b/examples/workflows/ci-cd-integration.yml new file mode 100644 index 0000000..b92a6c9 --- /dev/null +++ b/examples/workflows/ci-cd-integration.yml @@ -0,0 +1,24 @@ +manifest: ./manifests/ci-run.jsonld +steps: + - name: fetch_build_data + command: download + uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0 + localdir: ./workflow-output/ci + + - name: publish_build_artifact + command: deploy + version_id: https://databus.dbpedia.org/DhanashreeP/test-group/ci-demo/1.0 + title: "CI/CD Integration Demo" + abstract: "Example of a workflow suitable for GitHub Actions" + description: "No prompts, no interactive input -- safe to run in an unattended CI job" + license: https://creativecommons.org/licenses/by-sa/3.0/ + api_key: ${DATABUS_API_KEY} + files: ${steps.fetch_build_data.output_urls} + on_error: fail + + - name: cleanup_ci_artifact + command: delete + uris: + - https://databus.dbpedia.org/DhanashreeP/test-group/ci-demo/1.0 + api_key: ${DATABUS_API_KEY} + on_error: continue diff --git a/examples/workflows/failure-debugging.yml b/examples/workflows/failure-debugging.yml new file mode 100644 index 0000000..2eb589d --- /dev/null +++ b/examples/workflows/failure-debugging.yml @@ -0,0 +1,17 @@ +manifest: ./manifests/failure-debug.jsonld +steps: + - name: fetch_source + command: download + uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0 + localdir: ./workflow-output/failure-demo + + - name: deploy_with_bad_key + command: deploy + version_id: https://databus.dbpedia.org/DhanashreeP/test-group/failure-demo/1.0 + title: "Failure Debugging Demo" + abstract: "Intentionally fails to demonstrate manifest/console error output" + description: "Deploys with an invalid API key on purpose" + license: https://creativecommons.org/licenses/by-sa/3.0/ + api_key: THIS-KEY-IS-INTENTIONALLY-INVALID + files: ${steps.fetch_source.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 index 4ec618c..016e726 100644 --- a/examples/workflows/full-pipeline.yml +++ b/examples/workflows/full-pipeline.yml @@ -1,7 +1,7 @@ steps: - name: fetch_dataset command: download - uri: https://databus.dev.dbpedia.link/fhofer/gsoc26/test-data/2.0 + uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0 localdir: ./workflow-output/full-pipeline - name: publish_dataset diff --git a/examples/workflows/nightly-publishing-pipeline.yml b/examples/workflows/nightly-publishing-pipeline.yml new file mode 100644 index 0000000..659451c --- /dev/null +++ b/examples/workflows/nightly-publishing-pipeline.yml @@ -0,0 +1,24 @@ +manifest: ./manifests/nightly-publish.jsonld +steps: + - name: fetch_latest_data + command: download + uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0 + localdir: ./workflow-output/nightly + + - name: publish_processed_dataset + command: deploy + version_id: https://databus.dbpedia.org/DhanashreeP/test-group/nightly-demo/1.0 + title: "Nightly Publishing Demo" + abstract: "Automated nightly publishing pipeline example" + description: "Demonstrates download -> deploy -> cleanup running unattended" + license: https://creativecommons.org/licenses/by-sa/3.0/ + api_key: ${DATABUS_API_KEY} + files: ${steps.fetch_latest_data.output_urls} + on_error: fail + + - name: cleanup_previous_version + command: delete + uris: + - https://databus.dbpedia.org/DhanashreeP/test-group/nightly-demo/1.0 + api_key: ${DATABUS_API_KEY} + on_error: continue \ No newline at end of file diff --git a/examples/workflows/reproducible-research-download.yml b/examples/workflows/reproducible-research-download.yml new file mode 100644 index 0000000..a1cf37b --- /dev/null +++ b/examples/workflows/reproducible-research-download.yml @@ -0,0 +1,7 @@ +manifest: ./manifests/research-download.jsonld +steps: + - name: fetch_research_data + command: download + uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0 + localdir: ./workflow-output/research-download + validate_checksum: true \ No newline at end of file diff --git a/tests/test_download.py b/tests/test_download.py index 94d8813..17dcefa 100644 --- a/tests/test_download.py +++ b/tests/test_download.py @@ -35,3 +35,29 @@ def test_with_query(): ) def test_with_collection(): api_download("tmp", DEFAULT_ENDPOINT, [TEST_COLLECTION]) + +def test_404_records_failed_manifest_entry(monkeypatch): + from databusclient.manifest.context import ManifestContext + import databusclient.api.download as dl + + class FakeHeadResp: + status_code = 200 + headers = {} + + class FakeGetResp: + status_code = 404 + headers = {"content-length": "0"} + + def raise_for_status(self): + import requests + raise requests.exceptions.HTTPError(response=self) + + monkeypatch.setattr("requests.head", lambda *a, **k: FakeHeadResp()) + monkeypatch.setattr("requests.get", lambda *a, **k: FakeGetResp()) + + ctx = ManifestContext(command="download") + dl._download_file("https://example.org/missing.ttl", localDir=".", manifest_context=ctx) + + assert len(ctx.files) == 1 + assert ctx.files[0]["status"] == "failed" + assert ctx.files[0]["error_message"] == "404 Not Found" \ No newline at end of file diff --git a/tests/test_manifest_replay.py b/tests/test_manifest_replay.py index 59ed13d..6dc776a 100644 --- a/tests/test_manifest_replay.py +++ b/tests/test_manifest_replay.py @@ -445,4 +445,18 @@ def test_replay_deploy_missing_deploy_mode_raises(tmp_path): _write_manifest(path, manifest) with pytest.raises(ManifestReplayError, match="predate"): - replay_manifest(str(path), overrides={"api_key": "dummy-key"}) \ No newline at end of file + replay_manifest(str(path), overrides={"api_key": "dummy-key"}) + +def test_replay_workflow_manifest_gives_clean_not_implemented_error(tmp_path): + """Workflow manifests have no replayParams (workflows don't call + record_params()). Confirm this gives the standard 'not implemented' + message, not a confusing validation error about a missing field.""" + manifest = { + "@type": "dbus:OperationManifest", + "dbus:command": "workflow", + } + path = tmp_path / "workflow-manifest.jsonld" + _write_manifest(path, manifest) + + with pytest.raises(ManifestReplayError, match="not implemented"): + replay_manifest(str(path)) \ No newline at end of file diff --git a/tests/test_manifest_summary.py b/tests/test_manifest_summary.py index 9407449..16c6302 100644 --- a/tests/test_manifest_summary.py +++ b/tests/test_manifest_summary.py @@ -114,4 +114,72 @@ def test_summary_no_error_line_when_no_operation_error(): "dbus:executionResult": {"dbus:succeeded": 1, "dbus:failed": 0, "dbus:totalBytes": 100}, } output = format_summary(manifest) - assert "Error" not in output \ No newline at end of file + assert "Error" not in output + +def test_summary_ignores_step_name_field_gracefully(): + """dbus:stepName is a per-file field, not surfaced in the top-level + summary -- confirms it doesn't break formatting.""" + manifest = { + "dbus:command": "workflow", + "dcterms:issued": {"@value": "2024-03-24T10:00:00Z"}, + "dbus:executionResult": {"dbus:succeeded": 1, "dbus:failed": 0, "dbus:totalBytes": 0}, + "dataid:distribution": {"dataid:file": [{"dbus:stepName": "fetch"}]}, + } + output = format_summary(manifest) + assert "Command : workflow" in output + +def test_summary_lists_failed_file_details_with_step_name(): + manifest = { + "dbus:command": "workflow", + "dcterms:issued": {"@value": "2024-03-24T10:00:00Z"}, + "dbus:executionResult": {"dbus:succeeded": 1, "dbus:failed": 1, "dbus:totalBytes": 0}, + "dataid:distribution": { + "dataid:file": [ + {"dcat:downloadURL": "https://a.org/x", "dbus:status": "success"}, + { + "dcat:downloadURL": "step:deploy_with_bad_key", + "dbus:status": "failed", + "dbus:stepName": "deploy_with_bad_key", + "dbus:errorMessage": "Authentication failed.", + }, + ] + }, + } + output = format_summary(manifest) + assert "Failures:" in output + assert "[deploy_with_bad_key] step:deploy_with_bad_key: Authentication failed." in output + + +def test_summary_lists_failed_file_details_without_step_name(): + """Single-command manifests (not workflows) have no dbus:stepName -- + confirm the failed-file line still renders cleanly without it.""" + manifest = { + "dbus:command": "download", + "dcterms:issued": {"@value": "2024-03-24T10:00:00Z"}, + "dbus:executionResult": {"dbus:succeeded": 0, "dbus:failed": 1, "dbus:totalBytes": 0}, + "dataid:distribution": { + "dataid:file": [ + { + "dcat:downloadURL": "https://a.org/missing.ttl", + "dbus:status": "failed", + "dbus:errorMessage": "404 Not Found", + }, + ] + }, + } + output = format_summary(manifest) + assert "Failures:" in output + assert "https://a.org/missing.ttl: 404 Not Found" in output + + +def test_summary_no_failed_files_section_when_all_succeeded(): + manifest = { + "dbus:command": "download", + "dcterms:issued": {"@value": "2024-03-24T10:00:00Z"}, + "dbus:executionResult": {"dbus:succeeded": 1, "dbus:failed": 0, "dbus:totalBytes": 100}, + "dataid:distribution": { + "dataid:file": [{"dcat:downloadURL": "https://a.org/x", "dbus:status": "success"}] + }, + } + output = format_summary(manifest) + assert "Failures:" not in output \ No newline at end of file diff --git a/tests/test_workflow_engine.py b/tests/test_workflow_engine.py index 71005a4..f086368 100644 --- a/tests/test_workflow_engine.py +++ b/tests/test_workflow_engine.py @@ -133,4 +133,90 @@ def run(self, step_config, context): engine.run([ {"name": "publish", "command": "deploy", "files": "${steps.nonexistent.output_files}"}, - ]) \ No newline at end of file + ]) + +def test_workflow_manifest_records_step_names(monkeypatch, tmp_path): + from databusclient.manifest.context import ManifestContext + + class FetchStep: + def run(self, step_config, context): + context.manifest_context.record_file(url="https://a.org/x", status="success") + + class PublishStep: + def run(self, step_config, context): + context.manifest_context.record_file(url="https://a.org/y", status="success") + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FetchStep) + monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", PublishStep) + + manifest_ctx = ManifestContext(command="workflow") + engine = WorkflowEngine(manifest_context=manifest_ctx) + engine.run([ + {"name": "fetch", "command": "download"}, + {"name": "publish", "command": "deploy"}, + ]) + + steps_seen = {f["url"]: f.get("step") for f in manifest_ctx.files} + assert steps_seen["https://a.org/x"] == "fetch" + assert steps_seen["https://a.org/y"] == "publish" + + +def test_workflow_manifest_records_failed_step(monkeypatch): + from databusclient.manifest.context import ManifestContext + + 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) + + manifest_ctx = ManifestContext(command="workflow") + engine = WorkflowEngine(manifest_context=manifest_ctx) + + with pytest.raises(WorkflowExecutionError): + engine.run([{"name": "a", "command": "download"}]) + + failed_entries = [f for f in manifest_ctx.files if f["status"] == "failed"] + assert len(failed_entries) == 1 + assert failed_entries[0]["step"] == "a" + assert "boom" in failed_entries[0]["error_message"] + + +def test_workflow_without_manifest_context_still_works(monkeypatch): + """No manifest_context given -- workflow still runs normally, no crash.""" + class OKStep: + def run(self, step_config, context): + pass + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", OKStep) + + engine = WorkflowEngine() + results = engine.run([{"name": "a", "command": "download"}]) + assert results[0].status == "success" + +def test_workflow_manifest_whole_step_failure_gets_step_tag(monkeypatch): + """Whole-step failures (no file-level work happened) must be tagged + with 'step' the same way merged per-file failures are, so + format_summary()'s [stepname] prefix works for both cases.""" + from databusclient.manifest.context import ManifestContext + + class FailingStep: + def run(self, step_config, context): + raise RuntimeError("auth failed") + + from databusclient.workflow import steps as steps_module + monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", FailingStep) + + manifest_ctx = ManifestContext(command="workflow") + engine = WorkflowEngine(manifest_context=manifest_ctx) + + with pytest.raises(WorkflowExecutionError): + engine.run([{"name": "deploy_with_bad_key", "command": "deploy"}]) + + failed = [f for f in manifest_ctx.files if f["status"] == "failed"] + assert len(failed) == 1 + assert failed[0]["step"] == "deploy_with_bad_key" + assert "auth failed" in failed[0]["error_message"] \ No newline at end of file diff --git a/tests/test_workflow_steps.py b/tests/test_workflow_steps.py index d89c0e5..acba21d 100644 --- a/tests/test_workflow_steps.py +++ b/tests/test_workflow_steps.py @@ -370,4 +370,54 @@ def fake_deploy(dataid, api_key): }, ctx) assert captured["kwargs"]["distributions"] == ["https://example.org/data.ttl"] - assert ctx.get_output("publish", "output_files") == ["https://example.org/data.ttl"] \ No newline at end of file + assert ctx.get_output("publish", "output_files") == ["https://example.org/data.ttl"] + +def test_deploy_step_records_to_manifest_context_on_success(monkeypatch): + from databusclient.manifest.context import ManifestContext + + def fake_create_dataset(**kwargs): + return {"@graph": [{"@id": "fake"}]} + + def fake_deploy(dataid, api_key): + pass + + monkeypatch.setattr("databusclient.workflow.steps.create_dataset", fake_create_dataset) + monkeypatch.setattr("databusclient.workflow.steps.api_deploy_call", fake_deploy) + + manifest_ctx = ManifestContext(command="download") + ctx = StepContext(manifest_context=manifest_ctx) + 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": ["https://example.org/data.ttl"], + }, ctx) + + assert len(manifest_ctx.files) == 1 + assert manifest_ctx.files[0]["url"] == "https://example.org/data.ttl" + assert manifest_ctx.files[0]["status"] == "success" + + +def test_deploy_step_does_nothing_when_no_manifest_context(monkeypatch): + """No manifest_context set -- must not crash, just skip recording.""" + def fake_create_dataset(**kwargs): + return {"@graph": [{"@id": "fake"}]} + + def fake_deploy(dataid, api_key): + pass + + monkeypatch.setattr("databusclient.workflow.steps.create_dataset", fake_create_dataset) + monkeypatch.setattr("databusclient.workflow.steps.api_deploy_call", fake_deploy) + + ctx = StepContext() + 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": ["https://example.org/data.ttl"], + }, ctx) + # No assertion needed beyond "did not raise" \ No newline at end of file