diff --git a/docs/ert/reference/workflows/complete_workflows.rst b/docs/ert/reference/workflows/complete_workflows.rst index 61d8c551392..76fb816c333 100644 --- a/docs/ert/reference/workflows/complete_workflows.rst +++ b/docs/ert/reference/workflows/complete_workflows.rst @@ -122,6 +122,14 @@ The :code:`status` field is one of :code:`success`, :code:`failed` or jobs that were stopped because the workflow was cancelled are logged at :code:`INFO` level. +Workflows hooked in with :code:`HOOK_WORKFLOW` are in addition recorded +alongside the experiment they belong to, in +:code:`/experiments//workflow_events.jsonl`. That file +holds one JSON object per job invocation and exists so the output of a +workflow can be shown again later; it is not meant to be read directly. +Output from hooks that ran before the experiment is created, such as +:code:`PRE_EXPERIMENT`, is held back and written once the storage is created. + .. _runpath-file-workflows: Locating the realisations: diff --git a/src/ert/run_models/event.py b/src/ert/run_models/event.py index cfd66c8a502..73ec333593b 100644 --- a/src/ert/run_models/event.py +++ b/src/ert/run_models/event.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +from datetime import datetime from pathlib import Path from typing import Annotated, Any, Literal from uuid import UUID @@ -20,6 +21,7 @@ StartEvent, WarningEvent, ) +from ert.workflow_runner import WorkflowJobStatus logger = logging.getLogger(__name__) @@ -93,6 +95,23 @@ def write_as_csv(self, output_path: Path | None) -> None: self.data.to_csv("Report", output_path / str(self.run_id)) +class WorkflowEvent(BaseModel, extra="forbid"): + """The output of a single workflow job invocation.""" + + event_type: Literal["WorkflowEvent"] = "WorkflowEvent" + run_id: UUID + hook: str + workflow_name: str + job_name: str # The name of the workflow job that produced this event + job_index: int # The index of the job in the workflow command list + arguments: list[str] + stdout: str + stderr: str + status: WorkflowJobStatus + timestamp: datetime + iteration: int | None = None + + class RunPathCreationEvent(BaseModel, extra="forbid"): pass @@ -129,6 +148,7 @@ class RunPathCreatedEvent(RunPathCreationEvent): | SnapshotUpdateEvent | StartEvent | WarningEvent + | WorkflowEvent | EnsembleEvaluationWarning | StartingTotalRunPathCreationEvent | FinishedTotalRunPathCreationEvent diff --git a/src/ert/run_models/run_model.py b/src/ert/run_models/run_model.py index 03bf35219c3..cbb41e2a7b3 100644 --- a/src/ert/run_models/run_model.py +++ b/src/ert/run_models/run_model.py @@ -39,11 +39,13 @@ ConfigValidationError, DesignMatrix, HookedWorkflowFixtures, + HookRuntime, ModelConfig, ParameterConfig, PostSimulationFixtures, PreSimulationFixtures, QueueConfig, + Workflow, create_workflow_fixtures_from_hooked, ) from ert.config.queue_config import KnownQueueOptionsAdapter @@ -75,7 +77,7 @@ from ert.trace import tracer from ert.utils import log_duration from ert.warnings import PostExperimentWarning, capture_specific_warning -from ert.workflow_runner import WorkflowRunner +from ert.workflow_runner import WorkflowJobStatus, WorkflowRunner from ._create_run_path import create_run_path from .event import ( @@ -84,6 +86,7 @@ SnapshotUpdateEvent, StartEvent, StatusEvents, + WorkflowEvent, ) if TYPE_CHECKING: @@ -181,6 +184,8 @@ class RunModel(RunModelConfig, ABC): _start_iteration: int = PrivateAttr(default=0) _max_parallelism_violation: ParallelismViolation = ParallelismViolation() _workflow_runner: WorkflowRunner | None = PrivateAttr(default=None) + _workflow_run_id: uuid.UUID = PrivateAttr(default_factory=uuid.uuid4) + _pending_workflow_events: list[WorkflowEvent] = PrivateAttr(default_factory=list) def __init__( self, @@ -381,6 +386,8 @@ def handle_captured_event(message: Warning | str) -> None: self.send_event(WarningEvent(msg=str(message))) start_timestamp = datetime.datetime.now(tz=datetime.UTC) + self._workflow_run_id = uuid.uuid4() + self._pending_workflow_events = [] try: # ruff: ignore[too-many-statements-in-try-clause] self.send_event(StartEvent(timestamp=start_timestamp)) with ( @@ -832,24 +839,110 @@ def run_workflows( self, fixtures: HookedWorkflowFixtures, ) -> None: - for workflow in self.hooked_workflows[fixtures.hook]: - workflow_runner = WorkflowRunner( - workflow=workflow, - fixtures=create_workflow_fixtures_from_hooked(fixtures), - hook=str(fixtures.hook), - ) - self._workflow_runner = workflow_runner - try: + ensemble = getattr(fixtures, "ensemble", None) + experiment = ensemble.experiment if ensemble is not None else None + iteration = ensemble.iteration if ensemble is not None else None + try: + for workflow in self.hooked_workflows[fixtures.hook]: if self._end_event.is_set(): - workflow_runner.cancel() - raise UserCancelled("Experiment cancelled by user during workflows") + # Cancel all remaining workflows + self._send_cancelled_workflow_events( + workflow=workflow, + hook=fixtures.hook, + iteration=iteration, + ) + continue - workflow_runner.run_blocking() - finally: - self._workflow_runner = None + workflow_runner = WorkflowRunner( + workflow=workflow, + fixtures=create_workflow_fixtures_from_hooked(fixtures), + hook=str(fixtures.hook), + ) + self._workflow_runner = workflow_runner + try: + workflow_runner.run_blocking() + finally: + self._workflow_runner = None + self._send_workflow_events( + workflow_runner=workflow_runner, + hook=fixtures.hook, + workflow_name=workflow.name, + iteration=iteration, + ) + finally: + self._persist_workflow_events_to_storage(experiment) - if self._end_event.is_set(): - raise UserCancelled("Experiment cancelled by user during workflows") + if self._end_event.is_set(): + raise UserCancelled("Experiment cancelled by user during workflows") + + def _send_workflow_events( + self, + workflow_runner: WorkflowRunner, + hook: HookRuntime, + workflow_name: str, + iteration: int | None, + ) -> None: + events = [ + WorkflowEvent( + run_id=self._workflow_run_id, + hook=str(hook), + workflow_name=workflow_name, + job_name=result.name, + job_index=result.index, + arguments=result.arguments, + stdout=result.stdout, + stderr=result.stderr, + status=result.status, + timestamp=result.timestamp, + iteration=iteration, + ) + for result in workflow_runner.workflow_job_results() + ] + for event in events: + self.send_event(event) + self._pending_workflow_events.extend(events) + + def _send_cancelled_workflow_events( + self, + workflow: Workflow, + hook: HookRuntime, + iteration: int | None, + ) -> None: + # Report jobs not started due to cancellation + now = datetime.datetime.now(tz=datetime.UTC) + events = [ + WorkflowEvent( + run_id=self._workflow_run_id, + hook=str(hook), + workflow_name=workflow.name, + job_name=job.name, + job_index=index, + arguments=[str(arg) for arg in args], + stdout="", + stderr="", + status=WorkflowJobStatus.CANCELLED, + timestamp=now, + iteration=iteration, + ) + for index, (job, args) in enumerate(workflow) + ] + for event in events: + self.send_event(event) + self._pending_workflow_events.extend(events) + + def _persist_workflow_events_to_storage( + self, experiment: Experiment | None + ) -> None: + # Hold back output until storage is created + if experiment is None or not self._pending_workflow_events: + return + try: + experiment.append_workflow_events( + event.model_dump_json() for event in self._pending_workflow_events + ) + except Exception: + logger.exception("Failed to persist workflow events to storage") + self._pending_workflow_events = [] def _evaluate_and_postprocess( self, diff --git a/src/ert/storage/local_experiment.py b/src/ert/storage/local_experiment.py index f2fb9e11e69..9ad0ba7bb84 100644 --- a/src/ert/storage/local_experiment.py +++ b/src/ert/storage/local_experiment.py @@ -5,7 +5,7 @@ import json import logging import shutil -from collections.abc import Generator +from collections.abc import Generator, Iterable from datetime import UTC, datetime from enum import StrEnum, auto from functools import cached_property @@ -916,6 +916,23 @@ def export_everest_opt_results_to_csv(self) -> Path: def status_snapshot_path(self, iteration: int) -> Path: return self._path / "status" / f"iteration-{iteration}.json" + @property + def workflow_events_path(self) -> Path: + """The events describing the workflows run for this experiment. + + One JSON-serialized workflow log event per line, used to repopulate + the workflow view when an experiment is opened again later. + """ + return self._path / "workflow_events.jsonl" + + @require_write + def append_workflow_events(self, lines: Iterable[str]) -> None: + """Append already JSON-serialized workflow log events, one per line.""" + events_path = self.workflow_events_path + events_path.parent.mkdir(parents=True, exist_ok=True) + with events_path.open("a", encoding="utf-8") as fout: + fout.writelines(f"{line}\n" for line in lines) + def write_status_snapshot(self, iteration: int, data: bytes) -> None: snapshot_path = self.status_snapshot_path(iteration) snapshot_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/tests/ert/ui_tests/cli/test_cli.py b/tests/ert/ui_tests/cli/test_cli.py index 9a631538be4..d45676472df 100644 --- a/tests/ert/ui_tests/cli/test_cli.py +++ b/tests/ert/ui_tests/cli/test_cli.py @@ -30,6 +30,7 @@ ES_MDA_MODE, TEST_RUN_MODE, ) +from ert.run_models.event import WorkflowEvent from ert.sample_prior import sample_prior from ert.scheduler.driver import Driver from ert.scheduler.job import Job @@ -511,6 +512,49 @@ def test_that_stop_on_fail_workflow_jobs_stop_ert( run_cli(TEST_RUN_MODE, "--disable-monitoring", "poly.ert") +@pytest.mark.usefixtures("copy_poly_case") +def test_that_workflow_output_is_written_to_experiment_in_storage(): + Path("print_job").write_text("EXECUTABLE print_script.sh\n", encoding="utf-8") + Path("print_script.sh").write_text( + dedent( + """\ + #!/bin/bash + echo hello from the workflow + echo problem from the workflow >&2 + """ + ), + encoding="utf-8", + ) + Path("print_script.sh").chmod(os.stat("print_script.sh").st_mode | 0o111) + Path("print_workflow").write_text("printjob\n", encoding="utf-8") + + with Path("poly.ert").open(mode="a", encoding="utf-8") as fh: + fh.write( + dedent( + """ + LOAD_WORKFLOW_JOB print_job printjob + LOAD_WORKFLOW print_workflow wfprint + HOOK_WORKFLOW wfprint PRE_SIMULATION + """ + ) + ) + + run_cli(TEST_RUN_MODE, "--disable-monitoring", "poly.ert") + + with open_storage("storage", "r") as storage: + (experiment,) = storage.experiments + (line,) = experiment.workflow_events_path.read_text( + encoding="utf-8" + ).splitlines() + event = WorkflowEvent.model_validate_json(line) + assert event.hook == "PRE_SIMULATION" + assert event.workflow_name == "wfprint" + assert event.job_name == "printjob" + assert event.job_index == 0 + assert event.stdout == "hello from the workflow\n" + assert event.stderr == "problem from the workflow\n" + + @pytest.fixture(name="mock_cli_run") def fixture_mock_cli_run(monkeypatch): end_event = Mock() diff --git a/tests/ert/unit_tests/run_models/test_base_run_model.py b/tests/ert/unit_tests/run_models/test_base_run_model.py index 65f5e020281..2a6b70c6a02 100644 --- a/tests/ert/unit_tests/run_models/test_base_run_model.py +++ b/tests/ert/unit_tests/run_models/test_base_run_model.py @@ -2,6 +2,7 @@ import logging import math import os +import stat import uuid import warnings from pathlib import Path @@ -16,14 +17,20 @@ from ert.config import ( CircleShapeConfig, ErtConfig, + ExecutableWorkflow, + HookRuntime, ModelConfig, ObservationType, + PreSimulationFixtures, + PreUpdateFixtures, QueueConfig, QueueSystem, ShapeRegistry, + Workflow, ) from ert.config.parsing import ObservationDict from ert.config.queue_config import LsfQueueOptions +from ert.config.workflow_fixtures import PostExperimentFixtures, PreExperimentFixtures from ert.ensemble_evaluator import EndEvent, EvaluatorServerConfig, StartEvent from ert.ensemble_evaluator.evaluator import ParallelismViolation from ert.ensemble_evaluator.event import FullSnapshotEvent @@ -35,11 +42,13 @@ from ert.mode_definitions import TEST_RUN_MODE from ert.plugins import ErtRuntimePlugins from ert.run_models import create_model +from ert.run_models.event import WorkflowEvent from ert.run_models.run_model import ( RunModel, UserCancelled, ) from ert.warnings import PostExperimentWarning +from ert.workflow_runner import WorkflowJobStatus, WorkflowRunner @pytest.fixture(autouse=True) @@ -798,7 +807,7 @@ def test_that_status_snapshot_is_written_only_when_iteration_is_finalized(use_tm brm._iter_snapshot[0] = EnsembleSnapshot.from_nested_dict( {"reals": {"0": {"status": "Pending"}, "1": {"status": "Pending"}}} ) - experiment = brm._storage.create_experiment(name="exp") + experiment = brm._storage.create_experiment(name="experiment") brm.forward_event_from_ee( EESnapshotUpdate(snapshot={"reals": {"0": {"status": "Finished"}}}), @@ -916,3 +925,337 @@ def test_that_ert_config_logs_insensitive_information_about_observations_and_sha f"Count of shapes in ShapeRegistry: {{'{CircleShapeConfig.__name__}': 1}}" in caplog.text ) + + +def _drain(status_queue): + events = [] + while not status_queue.empty(): + events.append(status_queue.get()) + return events + + +def _persisted_workflow_events(experiment): + """The workflow events persisted to storage for an experiment, oldest first.""" + if not experiment.workflow_events_path.exists(): + return [] + return [ + WorkflowEvent.model_validate_json(line) + for line in experiment.workflow_events_path.read_text( + encoding="utf-8" + ).splitlines() + ] + + +def _printing_workflow(tmp_path, name, script, *, stop_on_fail=False): + executable = tmp_path / f"{name}.py" + executable.write_text(f"#!/usr/bin/env python\n{script}\n", encoding="utf-8") + executable.chmod(executable.stat().st_mode | stat.S_IEXEC) + return Workflow( + src_file=str(tmp_path / f"{name}_workflow"), + cmd_list=[ + ( + ExecutableWorkflow( + name=name.upper(), + executable=str(executable), + stop_on_fail=stop_on_fail, + ), + [], + ) + ], + ) + + +def test_that_run_workflows_sends_workflow_event_per_job(tmp_path, use_tmpdir): + workflow = _printing_workflow(tmp_path, "hello", 'print("hello from workflow")') + workflow.cmd_list.append(workflow.cmd_list[0]) + status_queue = SimpleQueue() + brm = create_run_model( + hooked_workflows={HookRuntime.PRE_EXPERIMENT: [workflow]}, + status_queue=status_queue, + ) + + brm.run_workflows(fixtures=PreExperimentFixtures(random_seed=1)) + + events = _drain(status_queue) + assert [(e.job_name, e.job_index, e.stdout) for e in events] == [ + ("HELLO", 0, "hello from workflow\n"), + ("HELLO", 1, "hello from workflow\n"), + ] + assert all(e.hook == "PRE_EXPERIMENT" for e in events) + assert all(e.run_id == brm._workflow_run_id for e in events) + assert all(e.status is WorkflowJobStatus.SUCCESS for e in events) + + +def test_that_workflow_event_is_sent_and_persisted_when_stop_on_fail_aborts_workflow( + tmp_path, use_tmpdir +): + workflow = _printing_workflow( + tmp_path, + "failing", + 'import sys\nprint("printed before failing")\nsys.exit(1)', + stop_on_fail=True, + ) + status_queue = SimpleQueue() + brm = create_run_model( + hooked_workflows={HookRuntime.PRE_SIMULATION: [workflow]}, + status_queue=status_queue, + ) + experiment = brm._storage.create_experiment(name="experiment") + ensemble = brm._storage.create_ensemble( + experiment, ensemble_size=1, name="ensemble" + ) + + with pytest.raises(RuntimeError, match="failed with error"): + brm.run_workflows( + fixtures=PreSimulationFixtures( + random_seed=1, + reports_dir="", + run_paths=MagicMock(), + storage=brm._storage, + ensemble=ensemble, + ) + ) + + (event,) = _drain(status_queue) + assert event.status is WorkflowJobStatus.FAILED + assert event.stdout == "printed before failing\n" + + assert [e.stdout for e in _persisted_workflow_events(experiment)] == [ + "printed before failing\n" + ] + + +def test_that_workflow_events_from_update_hook_carry_iteration(tmp_path, use_tmpdir): + workflow = _printing_workflow(tmp_path, "hello", 'print("hello")') + status_queue = SimpleQueue() + brm = create_run_model( + hooked_workflows={HookRuntime.PRE_UPDATE: [workflow]}, + status_queue=status_queue, + ) + ensemble = MagicMock() + ensemble.iteration = 2 + + brm.run_workflows( + fixtures=PreUpdateFixtures( + random_seed=1, + reports_dir="", + run_paths=MagicMock(), + storage=MagicMock(), + ensemble=ensemble, + es_settings=MagicMock(), + observation_settings=MagicMock(), + ) + ) + + (event,) = _drain(status_queue) + assert event.iteration == 2 + assert event.hook == "PRE_UPDATE" + + +def test_that_workflow_output_is_appended_to_experiment_in_storage( + tmp_path, use_tmpdir +): + workflow = _printing_workflow(tmp_path, "hello", 'print("hello from workflow")') + brm = create_run_model( + hooked_workflows={HookRuntime.PRE_SIMULATION: [workflow]}, + status_queue=SimpleQueue(), + ) + experiment = brm._storage.create_experiment(name="experiment") + ensemble = brm._storage.create_ensemble( + experiment, ensemble_size=1, name="ensemble" + ) + + brm.run_workflows( + fixtures=PreSimulationFixtures( + random_seed=1, + reports_dir="", + run_paths=MagicMock(), + storage=brm._storage, + ensemble=ensemble, + ) + ) + + (event,) = _persisted_workflow_events(experiment) + assert event.hook == "PRE_SIMULATION" + assert event.workflow_name == "hello_workflow" + assert event.job_name == "HELLO" + assert event.job_index == 0 + assert event.stdout == "hello from workflow\n" + + +def test_that_pre_experiment_output_is_persisted_once_experiment_exists( + tmp_path, use_tmpdir +): + startup = _printing_workflow(tmp_path, "startup", 'print("before the experiment")') + later = _printing_workflow(tmp_path, "later", 'print("after the experiment")') + brm = create_run_model( + hooked_workflows={ + HookRuntime.PRE_EXPERIMENT: [startup], + HookRuntime.PRE_SIMULATION: [later], + }, + status_queue=SimpleQueue(), + ) + + brm.run_workflows(fixtures=PreExperimentFixtures(random_seed=1)) + + experiment = brm._storage.create_experiment(name="experiment") + assert not experiment.workflow_events_path.exists() + + ensemble = brm._storage.create_ensemble( + experiment, ensemble_size=1, name="ensemble" + ) + brm.run_workflows( + fixtures=PreSimulationFixtures( + random_seed=1, + reports_dir="", + run_paths=MagicMock(), + storage=brm._storage, + ensemble=ensemble, + ) + ) + + assert [e.stdout for e in _persisted_workflow_events(experiment)] == [ + "before the experiment\n", + "after the experiment\n", + ] + + +def test_that_failure_to_persist_workflow_events_does_not_stop_experiment( + tmp_path, use_tmpdir, caplog +): + workflow = _printing_workflow(tmp_path, "hello", 'print("hello from workflow")') + status_queue = SimpleQueue() + brm = create_run_model( + hooked_workflows={HookRuntime.PRE_SIMULATION: [workflow]}, + status_queue=status_queue, + ) + ensemble = MagicMock() + ensemble.iteration = 0 + ensemble.experiment.append_workflow_events.side_effect = OSError("disk on fire") + + with caplog.at_level(logging.ERROR): + brm.run_workflows( + fixtures=PreSimulationFixtures( + random_seed=1, + reports_dir="", + run_paths=MagicMock(), + storage=MagicMock(), + ensemble=ensemble, + ) + ) + + assert "Failed to persist workflow events to storage" in caplog.text + assert _drain(status_queue), "the event should still be sent" + + +def test_that_starting_experiment_discards_workflow_output_from_previous_experiment( + use_tmpdir, +): + brm = create_run_model() + brm._status_queue = SimpleQueue() + brm._pending_workflow_events = [MagicMock()] + previous_log_id = brm._workflow_run_id + + brm.start_simulations_thread( + EvaluatorServerConfig(use_token=False), rerun_failed_realizations=True + ) + + assert brm._pending_workflow_events == [] + assert brm._workflow_run_id != previous_log_id + + +def test_that_workflow_output_is_persisted_when_user_cancels_experiment( + tmp_path, use_tmpdir +): + startup = _printing_workflow(tmp_path, "startup", 'print("before the experiment")') + later = _printing_workflow(tmp_path, "later", 'print("never runs")') + brm = create_run_model( + hooked_workflows={ + HookRuntime.PRE_EXPERIMENT: [startup], + HookRuntime.PRE_SIMULATION: [later], + }, + status_queue=SimpleQueue(), + ) + brm.run_workflows(fixtures=PreExperimentFixtures(random_seed=1)) + + experiment = brm._storage.create_experiment(name="experiment") + ensemble = brm._storage.create_ensemble( + experiment, ensemble_size=1, name="ensemble" + ) + brm._end_event.set() + + with pytest.raises(UserCancelled): + brm.run_workflows( + fixtures=PreSimulationFixtures( + random_seed=1, + reports_dir="", + run_paths=MagicMock(), + storage=brm._storage, + ensemble=ensemble, + ) + ) + + startup_event, skipped_event = _persisted_workflow_events(experiment) + assert startup_event.stdout == "before the experiment\n" + assert startup_event.status is not WorkflowJobStatus.CANCELLED + assert skipped_event.workflow_name == "later_workflow" + assert skipped_event.status is WorkflowJobStatus.CANCELLED + assert not skipped_event.stdout + + +def test_that_workflows_hooked_after_cancelled_workflow_still_appear_as_cancelled( + tmp_path, use_tmpdir +): + """Regression test: when several workflows are hooked to the same + runtime and cancellation happens while the first one is running, the + workflows that come after it in the hook's list must still be reported + (as cancelled) rather than silently disappearing from the workflow events. + """ + first = _printing_workflow(tmp_path, "first", 'print("first workflow")') + second = _printing_workflow(tmp_path, "second", 'print("second workflow")') + third = _printing_workflow(tmp_path, "third", 'print("third workflow")') + status_queue = SimpleQueue() + brm = create_run_model( + hooked_workflows={ + HookRuntime.POST_EXPERIMENT: [first, second, third], + }, + status_queue=status_queue, + ) + + # Simulate cancellation happening while the first workflow is running, + # i.e. before the second and third ones get their turn. + real_run_blocking = WorkflowRunner.run_blocking + + def _run_blocking_then_cancel(self): + brm._end_event.set() + return real_run_blocking(self) + + experiment = brm._storage.create_experiment(name="experiment") + ensemble = brm._storage.create_ensemble( + experiment, ensemble_size=1, name="ensemble" + ) + + with ( + patch.object( + WorkflowRunner, "run_blocking", _run_blocking_then_cancel, create=False + ), + pytest.raises(UserCancelled), + ): + brm.run_workflows( + fixtures=PostExperimentFixtures( + random_seed=1, storage=brm._storage, ensemble=ensemble + ) + ) + + events = _drain(status_queue) + assert [e.workflow_name for e in events] == [ + "first_workflow", + "second_workflow", + "third_workflow", + ] + first_event, second_event, third_event = events + + assert first_event.status is WorkflowJobStatus.SUCCESS + + for skipped_event in (second_event, third_event): + assert skipped_event.status is WorkflowJobStatus.CANCELLED diff --git a/tests/ert/unit_tests/run_models/test_status_events_serialization.py b/tests/ert/unit_tests/run_models/test_status_events_serialization.py index 74b709a718a..a9f569360f1 100644 --- a/tests/ert/unit_tests/run_models/test_status_events_serialization.py +++ b/tests/ert/unit_tests/run_models/test_status_events_serialization.py @@ -20,10 +20,12 @@ RunModelUpdateBeginEvent, RunModelUpdateEndEvent, SnapshotUpdateEvent, + WorkflowEvent, load_status_snapshot_event, status_event_from_json, status_event_to_json, ) +from ert.workflow_runner import WorkflowJobStatus from tests.ert.utils import SnapshotBuilder METADATA = EnsembleSnapshotMetadata( @@ -185,6 +187,22 @@ ), id="RunModelUpdateEndEvent", ), + pytest.param( + WorkflowEvent( + run_id=uuid.uuid1(), + hook="PRE_UPDATE", + workflow_name="my_workflow", + job_name="MY_JOB", + job_index=0, + arguments=["a", "b"], + stdout="some output\n", + stderr="", + status=WorkflowJobStatus.SUCCESS, + timestamp=dt(2020, 1, 1, tzinfo=UTC), + iteration=1, + ), + id="WorkflowEvent", + ), ], ) def test_status_event_serialization(event): diff --git a/tests/ert/unit_tests/storage/test_workflow_events.py b/tests/ert/unit_tests/storage/test_workflow_events.py new file mode 100644 index 00000000000..23a17e5467d --- /dev/null +++ b/tests/ert/unit_tests/storage/test_workflow_events.py @@ -0,0 +1,38 @@ +import json + +from ert.storage import open_storage + + +def test_that_appended_workflow_events_accumulate_as_one_line_each(tmp_path): + with open_storage(tmp_path, mode="w") as storage: + experiment = storage.create_experiment(name="experiment") + + experiment.append_workflow_events(['{"job": "first"}']) + experiment.append_workflow_events(['{"job": "second"}', '{"job": "third"}']) + + lines = experiment.workflow_events_path.read_text(encoding="utf-8").splitlines() + assert [json.loads(line)["job"] for line in lines] == [ + "first", + "second", + "third", + ] + + +def test_that_workflow_events_live_next_to_experiment_they_belong_to(tmp_path): + with open_storage(tmp_path, mode="w") as storage: + first = storage.create_experiment(name="first") + second = storage.create_experiment(name="second") + + first.append_workflow_events(['{"job": "belongs to the first"}']) + second.append_workflow_events(['{"job": "belongs to the second"}']) + + assert first.workflow_events_path.parent == first._path + assert first.workflow_events_path != second.workflow_events_path + assert ( + first.workflow_events_path.read_text(encoding="utf-8") + == '{"job": "belongs to the first"}\n' + ) + assert ( + second.workflow_events_path.read_text(encoding="utf-8") + == '{"job": "belongs to the second"}\n' + )