[API] Accept TOML text as configuration inputs

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

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,27 +54,27 @@ 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.
     """
     return cloudai.handlers.list_experiments(system)
 
 
 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

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,19 +381,20 @@ 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 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:

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")))