diff --git a/databusclient/workflow/steps.py b/databusclient/workflow/steps.py index 49d6061..2b673e6 100644 --- a/databusclient/workflow/steps.py +++ b/databusclient/workflow/steps.py @@ -86,6 +86,11 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None: ) new_entries = capture_context.files[files_before:] + failed_entries = [e for e in new_entries if e.get("status") == "failed"] + if failed_entries: + raise RuntimeError( + f"Step '{name}': download failed for {len(failed_entries)} file(s)." + ) resolved_urls = [e["url"] for e in new_entries if e.get("status") == "success"] output_files = self._collect_output_files(local_dir) diff --git a/tests/test_workflow_download_failures.py b/tests/test_workflow_download_failures.py new file mode 100644 index 0000000..1728f80 --- /dev/null +++ b/tests/test_workflow_download_failures.py @@ -0,0 +1,54 @@ +"""A recorded download failure must reach workflow error handling.""" + +from unittest.mock import Mock + +import pytest +import requests + +from databusclient.manifest.context import ManifestContext +from databusclient.workflow.engine import WorkflowEngine, WorkflowExecutionError + + +@pytest.mark.parametrize("with_manifest", [False, True]) +@pytest.mark.parametrize("policy", ["fail", "continue", "retry"]) +def test_download_404_obeys_workflow_error_policy( + monkeypatch, tmp_path, with_manifest, policy +): + missing = requests.Response() + missing.status_code = 404 + success = requests.Response() + success.status_code = 200 + success.headers["Content-Length"] = "4" + success.iter_content = lambda chunk_size: iter([b"data"]) + head = requests.Response() + head.status_code = 200 + monkeypatch.setattr("requests.head", lambda *args, **kwargs: head) + get = Mock(side_effect=[missing, success]) + monkeypatch.setattr("requests.get", get) + + manifest = ManifestContext(command="workflow") if with_manifest else None + engine = WorkflowEngine(manifest_context=manifest) + step = { + "name": "fetch", + "command": "download", + "uri": "https://example.org/account/group/artifact/version/data.ttl", + "localdir": str(tmp_path), + "on_error": policy, + } + if policy == "retry": + step["retry"] = {"max_attempts": 2, "delay_seconds": 0} + + if policy == "fail": + with pytest.raises(WorkflowExecutionError, match="download failed"): + engine.run([step]) + else: + result = engine.run([step])[0] + if policy == "continue": + assert result.status == "skipped_error" + assert get.call_count == 1 + else: + assert result.status == "success" + assert result.attempts == 2 + assert get.call_count == 2 + output = tmp_path / "account/group/artifact/version/data.ttl" + assert output.read_bytes() == b"data"