From 71f53f897188b48034a0881dfa63f3b9bad354bc Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Mon, 28 Sep 2026 16:02:45 +0200 Subject: [PATCH] Accept TOML text in the experiment API Signed-off-by: Ivan Podkidyshev --- doc/api.rst | 8 +++- src/cloudai/api.py | 20 ++++----- src/cloudai/handlers.py | 99 ++++++++++++++++++++++++++++++----------- tests/test_api.py | 99 ++++++++++++++++++++++------------------- 4 files changed, 141 insertions(+), 85 deletions(-) diff --git a/doc/api.rst b/doc/api.rst index f65555d7c..e988c95e1 100644 --- a/doc/api.rst +++ b/doc/api.rst @@ -26,8 +26,12 @@ Pass configuration files as ``pathlib.Path`` objects: experiment = cloudai.api.run_experiment(scenario, system, tests_dir=tests_dir) print(experiment.status, experiment.path) -Test ``path`` references resolve relative to the scenario file. Other relative -paths keep their usual meaning relative to the working directory. +A ``str`` argument for ``scenario`` or ``system`` means TOML content. Use ``Path`` +for a filename. Test ``path`` references resolve relative to a scenario file; +when passing scenario text, use absolute test paths or inline test definitions. +Other relative paths keep their usual meaning relative to the working directory. +Configuration supplied as text is saved in the result directory as ``system.toml`` +and ``scenario.toml``. ``run_experiment`` installs workload prerequisites as needed and returns an ``Experiment`` model containing the saved results. Each invocation gets its own diff --git a/src/cloudai/api.py b/src/cloudai/api.py index 5cfaed099..2344217ff 100644 --- a/src/cloudai/api.py +++ b/src/cloudai/api.py @@ -21,8 +21,8 @@ def run_experiment( - scenario: Path, - system: Path, + scenario: str | Path, + system: str | Path, *, tests_dir: Path | None = None, hook_dir: Path | None = None, @@ -34,9 +34,9 @@ def run_experiment( """ Run a scenario and return its experiment snapshot. - Pass configuration files as Path. Relative test paths resolve relative to - the scenario file. Optional directories have the same meaning as in the CLI. - The call waits for execution to finish. + Pass TOML content as str or configuration files as Path. Relative test paths + require a scenario file. Optional directories have the same meaning as in + the CLI. The call waits for execution to finish. Configuration and setup errors raise exceptions. Workload failures are recorded in the returned experiment's status. This function does not invoke @@ -54,11 +54,11 @@ def run_experiment( ) -def list_experiments(system: Path) -> list[tuple[str, Path]]: +def list_experiments(system: str | Path) -> list[tuple[str, Path]]: """ Return experiment IDs and local result directories under system.output_path. - Pass the system configuration file as Path. A missing results + Pass system TOML as str or a configuration file as Path. A missing results directory returns an empty list. Unreadable or invalid experiment files raise exceptions rather than returning an incomplete catalog. """ @@ -66,15 +66,15 @@ def list_experiments(system: Path) -> list[tuple[str, Path]]: def validate_scenario( - scenario: Path, - system: Path, + scenario: str | Path, + system: str | Path, *, tests_dir: Path | None = None, hook_dir: Path | None = None, single_sbatch: bool = False, ) -> tuple[bool, dict[str, str]]: """ - Validate configuration files without installing or running jobs. + Validate TOML content or config files without installing or running jobs. Return (True, {}) on success, otherwise (False, errors), with configuration names as keys and error messages as values. This checks configuration and diff --git a/src/cloudai/handlers.py b/src/cloudai/handlers.py index 4995b8264..d476c2395 100644 --- a/src/cloudai/handlers.py +++ b/src/cloudai/handlers.py @@ -19,15 +19,19 @@ import logging import tempfile import traceback +from contextlib import contextmanager from pathlib import Path from typing import Optional +import toml + import cloudai.models.output import cloudai.output from cloudai.configurator.env_params import validate_domain_randomization_active from cloudai.core import ( BaseInstaller, CloudAIGymEnv, + ConfigPaths, Installable, MissingTestError, Parser, @@ -288,9 +292,35 @@ def execute_experiment(runner: Runner, tests: list[TestDefinition], enable_cache runner.runner.finish_output(successful) +@contextmanager +def _configuration_files(system: str | Path, scenario: str | Path | None = None): + with tempfile.TemporaryDirectory(prefix="cloudai-config-") as temporary: + system_path = system.expanduser().resolve() if isinstance(system, Path) else Path(temporary) / "system.toml" + if isinstance(system, str): + system_path.write_text(system, encoding="utf-8") + scenario_path = Path(temporary) / "scenario.toml" + if isinstance(scenario, Path): + scenario_path = scenario.expanduser().resolve() + elif isinstance(scenario, str): + try: + data = toml.loads(scenario) + except toml.TomlDecodeError as exc: + raise TestScenarioParsingError(str(exc)) from exc + tests = data.get("Tests", []) + if isinstance(tests, list): + for test in tests: + path = test.get("path") if isinstance(test, dict) else None + if isinstance(path, str) and not Path(path).is_absolute(): + raise TestScenarioParsingError( + "Relative test paths require a scenario file; pass a Path or use absolute test paths." + ) + scenario_path.write_text(scenario, encoding="utf-8") + yield system_path, scenario_path + + def run_experiment( - scenario: Path, - system: Path, + scenario: str | Path, + system: str | Path, *, tests_dir: Path | None = None, hook_dir: Path | None = None, @@ -299,23 +329,36 @@ def run_experiment( single_sbatch: bool = False, enable_cache_without_check: bool = False, ) -> cloudai.models.output.Experiment: - """Run an experiment from configuration files and return its results.""" - parsed_system, tests, parsed_scenario = load_scenario( - scenario.expanduser().resolve(), system.expanduser().resolve(), tests_dir=tests_dir, hook_dir=hook_dir - ) - validate_experiment(parsed_system, parsed_scenario, single_sbatch) - parsed_system.output_path = (output_dir or parsed_system.output_path).expanduser().resolve() - if prepare_output_dir(parsed_system.output_path) is None: - raise OSError(f"Cannot prepare output directory: {parsed_system.output_path}") - timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - result_dir = Path( - tempfile.mkdtemp(prefix=f"{parsed_scenario.name}_{timestamp}_", dir=parsed_system.output_path) - ).resolve() - runner = create_experiment_runner( - parsed_system, parsed_scenario, dry_run=dry_run, single_sbatch=single_sbatch, result_dir=result_dir - ) - execute_experiment(runner, tests, enable_cache_without_check) - return runner.runner.experiment_output.snapshot() + """Run an experiment from configuration files or TOML text.""" + with _configuration_files(system, scenario) as (system_path, scenario_path): + parsed_system, tests, parsed_scenario = load_scenario( + scenario_path, system_path, tests_dir=tests_dir, hook_dir=hook_dir + ) + validate_experiment(parsed_system, parsed_scenario, single_sbatch) + parsed_system.output_path = (output_dir or parsed_system.output_path).expanduser().resolve() + if prepare_output_dir(parsed_system.output_path) is None: + raise OSError(f"Cannot prepare output directory: {parsed_system.output_path}") + timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + result_dir = Path( + tempfile.mkdtemp(prefix=f"{parsed_scenario.name}_{timestamp}_", dir=parsed_system.output_path) + ).resolve() + if isinstance(system, str): + system_path = result_dir / "system.toml" + system_path.write_text(system, encoding="utf-8") + if isinstance(scenario, str): + scenario_path = result_dir / "scenario.toml" + scenario_path.write_text(scenario, encoding="utf-8") + parsed_scenario.config_paths = ConfigPaths( + system_path=system_path, + test_scenario_path=scenario_path, + tests_dir_path=tests_dir.resolve() if tests_dir is not None else None, + ) + runner = create_experiment_runner( + parsed_system, parsed_scenario, dry_run=dry_run, single_sbatch=single_sbatch, result_dir=result_dir + ) + output = runner.runner.experiment_output + execute_experiment(runner, tests, enable_cache_without_check) + return output.snapshot() def get_experiment(exp: str | Path) -> cloudai.models.output.Experiment: @@ -326,9 +369,10 @@ def get_experiment(exp: str | Path) -> cloudai.models.output.Experiment: return cloudai.models.output.Experiment.model_validate_json(path.read_text(encoding="utf-8")) -def list_experiments(system: Path) -> list[tuple[str, Path]]: +def list_experiments(system: str | Path) -> list[tuple[str, Path]]: """Discover experiments beneath the configured results directory.""" - output_dir = Parser.parse_system(system.expanduser()).output_path.expanduser().resolve() + with _configuration_files(system) as (system_path, _): + output_dir = Parser.parse_system(system_path).output_path.expanduser().resolve() if not output_dir.exists(): return [] if not output_dir.is_dir(): @@ -337,8 +381,8 @@ def list_experiments(system: Path) -> list[tuple[str, Path]]: def validate_scenario( - scenario: Path, - system: Path, + scenario: str | Path, + system: str | Path, *, tests_dir: Path | None = None, hook_dir: Path | None = None, @@ -346,10 +390,11 @@ def validate_scenario( ) -> tuple[bool, dict[str, str]]: """Validate configuration and execution constraints without running workloads.""" try: - parsed_system, _, parsed_scenario = load_scenario( - scenario.expanduser().resolve(), system.expanduser().resolve(), tests_dir=tests_dir, hook_dir=hook_dir - ) - validate_experiment(parsed_system, parsed_scenario, single_sbatch) + with _configuration_files(system, scenario) as (system_path, scenario_path): + parsed_system, _, parsed_scenario = load_scenario( + scenario_path, system_path, tests_dir=tests_dir, hook_dir=hook_dir + ) + validate_experiment(parsed_system, parsed_scenario, single_sbatch) except SystemConfigParsingError as exc: return False, {"system": str(exc.__cause__ or exc)} except (TestConfigParsingError, TestScenarioParsingError, MissingTestError, ValueError, OSError) as exc: diff --git a/tests/test_api.py b/tests/test_api.py index 544158c2d..0d2158672 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -36,41 +36,37 @@ @pytest.fixture -def configs(standalone_system: StandaloneSystem, tmp_path: Path) -> tuple[Path, Path]: - system = tmp_path / "system.toml" - system.write_text( - toml.dumps( - { - "name": standalone_system.name, - "scheduler": "standalone", - "install_path": str(standalone_system.install_path), - "output_path": str(standalone_system.output_path), - "monitor_interval": 0, - } - ) +def configs(standalone_system: StandaloneSystem) -> tuple[str, str]: + system = toml.dumps( + { + "name": standalone_system.name, + "scheduler": "standalone", + "install_path": str(standalone_system.install_path), + "output_path": str(standalone_system.output_path), + "monitor_interval": 0, + } ) - scenario = tmp_path / "scenario.toml" - scenario.write_text( - toml.dumps( - { - "name": "api-test", - "Tests": [ - { - "id": "sleep", - "name": "sleep", - "description": "API smoke test", - "test_template_name": "Sleep", - "cmd_args": {"seconds": 0}, - } - ], - } - ) + scenario = toml.dumps( + { + "name": "api-test", + "Tests": [ + { + "id": "sleep", + "name": "sleep", + "description": "API smoke test", + "test_template_name": "Sleep", + "cmd_args": {"seconds": 0}, + } + ], + } ) return scenario, system -@pytest.mark.parametrize("successful", [True, False]) -def test_experiment_api(configs: tuple[Path, Path], tmp_path: Path, monkeypatch: pytest.MonkeyPatch, successful: bool): +@pytest.mark.parametrize("successful,as_text", [(True, False), (True, True), (False, True)]) +def test_experiment_api( + configs: tuple[str, str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch, successful: bool, as_text: bool +): scenario, system = configs monkeypatch.setattr(StandaloneSystem, "is_job_completed", lambda *_: True) monkeypatch.setattr(StandaloneSystem, "is_job_running", lambda *_: False) @@ -94,7 +90,14 @@ def test_experiment_api(configs: tuple[Path, Path], tmp_path: Path, monkeypatch: assert cloudai.api.list_experiments(system) == [] assert cloudai.api.validate_scenario(scenario, system) == (True, {}) assert not (tmp_path / "output").exists() - experiment = cloudai.api.run_experiment(scenario, system) + scenario_input: str | Path = scenario + system_input: str | Path = system + if not as_text: + scenario_input = tmp_path / "scenario.toml" + system_input = tmp_path / "system.toml" + scenario_input.write_text(scenario) + system_input.write_text(system) + experiment = cloudai.api.run_experiment(scenario_input, system_input) assert logging.getLogger().handlers == handlers assert signal.getsignal(signal.SIGINT) == sigint assert experiment.status == ("completed" if successful else "failed") @@ -109,6 +112,9 @@ def test_experiment_api(configs: tuple[Path, Path], tmp_path: Path, monkeypatch: result_dir = Path(experiment.path) assert cloudai.api.get_experiment(result_dir / "experiment.json") == experiment assert cloudai.api.list_experiments(system) == [(experiment.id, result_dir)] + if as_text: + assert (result_dir / "system.toml").read_text() == system + assert (result_dir / "scenario.toml").read_text() == scenario again = cloudai.api.run_experiment(scenario, system) assert again.id != experiment.id assert len(cloudai.api.list_experiments(system)) == 2 @@ -119,29 +125,30 @@ def test_experiment_api(configs: tuple[Path, Path], tmp_path: Path, monkeypatch: cloudai.api.list_experiments(system) -def test_validate_scenario(configs: tuple[Path, Path], tmp_path: Path): +def test_validate_scenario(configs: tuple[str, str], tmp_path: Path): scenario, system = configs - template = toml.loads(scenario.read_text())["Tests"][0] + system_path = tmp_path / "system.toml" + system_path.write_text(system) + template = toml.loads(scenario)["Tests"][0] template.pop("id") (tmp_path / "sleep.toml").write_text(toml.dumps(template)) relative = 'name = "relative"\n[[Tests]]\nid = "sleep"\npath = "sleep.toml"\n' - scenario.write_text(relative) - assert cloudai.api.validate_scenario(scenario, system) == (True, {}) + scenario_path = tmp_path / "scenario.toml" + scenario_path.write_text(relative) + assert cloudai.api.validate_scenario(scenario_path, system_path) == (True, {}) + valid, errors = cloudai.api.validate_scenario(relative, system) + assert not valid and "Relative test paths" in errors["scenario"] for invalid in ("name = [", 'name = "invalid"'): - scenario.write_text(invalid) - valid, errors = cloudai.api.validate_scenario(scenario, system) + valid, errors = cloudai.api.validate_scenario(invalid, system) assert not valid and errors["scenario"] with pytest.raises(cloudai.core.TestScenarioParsingError): - cloudai.api.run_experiment(scenario, system) - scenario.write_text(relative) - original = system.read_text() - system.write_text(original.replace('scheduler = "standalone"', 'scheduler = "bad"')) - valid, errors = cloudai.api.validate_scenario(scenario, system) + cloudai.api.run_experiment(invalid, system) + valid, errors = cloudai.api.validate_scenario( + scenario, system.replace('scheduler = "standalone"', 'scheduler = "bad"') + ) assert not valid and "Unsupported system type" in errors["system"] - system.write_text(original.replace('name = "standalone"', "name = 42")) - valid, errors = cloudai.api.validate_scenario(scenario, system) + valid, errors = cloudai.api.validate_scenario(scenario, system.replace('name = "standalone"', "name = 42")) assert not valid and "name" in errors["system"] - system.write_text(original) valid, errors = cloudai.api.validate_scenario(scenario, tmp_path / "missing.toml") assert not valid and "missing.toml" in errors["system"] valid, errors = cloudai.api.validate_scenario(scenario, system, single_sbatch=True) @@ -152,7 +159,7 @@ def test_validate_scenario(configs: tuple[Path, Path], tmp_path: Path): @pytest.mark.parametrize("failure", ["setup", "submission"]) -def test_failed_experiment_is_saved(configs: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch, failure: str): +def test_failed_experiment_is_saved(configs: tuple[str, str], monkeypatch: pytest.MonkeyPatch, failure: str): scenario, system = configs if failure == "setup": monkeypatch.setattr(StandaloneSystem, "update", Mock(side_effect=RuntimeError("setup failed")))