Test model CLI

atomkraft/config/config_file.py

@@ -1,8 +1,11 @@
+from pathlib import Path
+from typing import Any
+
 import tomlkit
 
 
 class ConfigFile(object):
-    def __init__(self, path: str):
+    def __init__(self, path: Path):
         self.data = {}
         self.path = path
 
@@ -13,14 +16,28 @@ def __enter__(self):
         return self
 
     def __exit__(self, type, value, traceback):
+        self.fd.truncate(0)
+        tomlkit.dump(self.data, self.fd)
         self.fd.close()
 
-    def query(self, key) -> str:
-        if key not in self.data:
-            raise KeyError
+    def __getitem__(self, key) -> Any:
         return self.data[key]
 
-    def store(self, key, value):
+    def __setitem__(self, key, value):
         self.data[key] = value
-        self.fd.truncate(0)
-        tomlkit.dump(self.data, self.fd)
+
+    def __delitem__(self, key):
+        del self.data[key]
+
+    def get_or_update(self, key: str, value: Any) -> Any:
+        try:
+            return self.data[key]
+        except KeyError:
+            self.data[key] = value
+            return value
+
+    def try_get(self, key: str, default: Any) -> Any:
+        try:
+            return self.data[key]
+        except KeyError:
+            return default

atomkraft/config/model_config.py

@@ -1,4 +1,5 @@
 import os
+from pathlib import Path
 from typing import Optional
 
 from atomkraft.config.config_file import ConfigFile
@@ -8,11 +9,11 @@
 
 
 class ModelConfig(ConfigFile):
-    def __init__(self, path: Optional[str] = None):
+    def __init__(self, path: Optional[Path] = None):
         if not path:
             if "PROJECT_ROOT" in os.environ:
-                root = os.environ["PROJECT_ROOT"]
+                root = Path(os.environ["PROJECT_ROOT"])
             else:
                 root = project_root()
-            path = os.path.join(root, MODEL_CONFIG_PATH)
+            path = root / MODEL_CONFIG_PATH
         super().__init__(path)

atomkraft/model/traces.py

@@ -1,10 +1,9 @@
 from pathlib import Path
-from typing import List
+from typing import List, Optional
 
 from atomkraft.config.atomkraft_config import AtomkraftConfig
 from atomkraft.config.model_config import ModelConfig
 from atomkraft.utils.filesystem import last_modified_file_in
-from modelator.cli.model_config_file import load_config_file
 from modelator.itf import ITF
 from modelator.Model import Model
 from modelator.ModelResult import ModelResult
@@ -14,13 +13,15 @@ def query_configs(key: str) -> str:
     with AtomkraftConfig() as atomkraft_config:
         with ModelConfig() as model_config:
             try:
-                return model_config.query(key) or atomkraft_config.query(key)
+                return model_config[key] or atomkraft_config[key]
             except KeyError:
                 raise FileNotFoundError
 
 
 def generate_traces(
-    model_config_path=None, model_path=None, sample_operators=[]
+    model_config_path: Optional[Path],
+    model_path: Optional[Path] = None,
+    sample_operators=[],
 ) -> ModelResult:
     """
     Call Modelator to get samples of the given model in `model_path`. Return the
@@ -34,21 +35,27 @@ def generate_traces(
     """
     init = "Init"
     next = "Next"
-    if model_config_path:
-        model_config = load_config_file(model_config_path)
-        model_path = model_config["model_path"]
-        init = model_config["init"] or init
-        next = model_config["next"] or next
-        sample_operators = list(set(model_config["examples"] + sample_operators))
+    traces_dir = "traces"
+    with ModelConfig(model_config_path) as model_config:
+        model_path = model_path or Path(model_config["model_path"])
+        init = model_config.try_get("init", init)
+        next = model_config.try_get("next", next)
+        sample_operators = list(
+            set(model_config.try_get("examples", []) + sample_operators)
+        )
+        traces_dir = model_config.try_get("traces_dir", traces_dir)
 
-    if not model_path:
-        raise ValueError("No model path given.")
+        model_config["model_path"] = str(model_path)
+        model_config["init"] = init
+        model_config["next"] = next
+        model_config["examples"] = sample_operators
+        model_config["traces_dir"] = traces_dir
 
-    if not Path(model_path).is_file():
+    if not model_path.is_file():
         raise FileNotFoundError(f"File with model not found: {model_path}")
 
-    model = Model.parse_file(model_path, init, next)
-    return model.sample(examples=sample_operators)
+    model = Model.parse_file(str(model_path), init, next)
+    return model.sample(traces_dir=traces_dir, examples=sample_operators)
 
 
 def last_modified_trace_path() -> str:

atomkraft/reactor/reactor.py

@@ -4,6 +4,7 @@
 from typing import List
 
 from atomkraft.config.atomkraft_config import AtomkraftConfig
+from atomkraft.utils.project import get_absolute_project_path
 from caseconverter import snakecase
 
 from . import constants, utils
@@ -36,13 +37,13 @@ def check_reactor(trace: PathLike, reactor=None) -> bool:
         return all_trace_actions.issubset(v.step_functions)
 
 
-def get_reactor() -> PathLike:
+def get_reactor() -> Path:
     """
     returns the path to the current reactor from the internal config
     """
     with AtomkraftConfig() as config:
         try:
-            return config.query(constants.REACTOR_CONFIG_KEY)
+            return get_absolute_project_path(config[constants.REACTOR_CONFIG_KEY])
         except KeyError:
             raise RuntimeError(
                 "Could not find default reactor; have you ran `atomkraft reactor`?"
@@ -75,7 +76,7 @@ def generate_reactor(
         f.write(actions_stub)
 
     with AtomkraftConfig() as config:
-        config.store(constants.REACTOR_CONFIG_KEY, str(stub_file_path))
+        config[constants.REACTOR_CONFIG_KEY] = str(stub_file_path)
 
     return stub_file_path
 

atomkraft/test/__init__.py

@@ -1,7 +1,10 @@
+from pathlib import Path
 from typing import List, Optional
 
 import typer
+from atomkraft.config.atomkraft_config import AtomkraftConfig
 
+from .model import MODEL_CONFIG_KEY, test_model
 from .trace import test_trace
 
 app = typer.Typer(rich_markup_mode="rich", add_completion=False)
@@ -10,6 +13,9 @@
 def FileOption(help, default):
     return typer.Option(
         None,
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
         show_default=False,
         help=f"{help} [grey30][default: set via [bold cyan]atomkraft {default}[/bold cyan]][/grey30]",
     )
@@ -18,6 +24,9 @@ def FileOption(help, default):
 def RequiredFileOption(help, default):
     return typer.Option(
         ...,
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
         show_default=False,
         help=f"{help} [grey30][default: set via [bold cyan]atomkraft {default}[/bold cyan]][/grey30]",
     )
@@ -27,47 +36,50 @@ def RequiredFileOption(help, default):
 def trace(
     # currently, require the trace to be present.
     # later, there will be an option to pick up the last one from the model
-    verbose: bool = typer.Option(
-        False, "--verbose", "-v", help="Output logging on console"
-    ),
-    trace: typer.FileText = RequiredFileOption("trace to execute", "model"),
-    reactor: typer.FileText = FileOption("reactor to interpret the trace", "reactor"),
+    trace: Path = RequiredFileOption("trace to execute", "model"),
+    reactor: Optional[Path] = FileOption("reactor to interpret the trace", "reactor"),
     keypath: str = typer.Option(
         "action",
         show_default=True,
         help="Path to key used as step name, extracted from ITF states",
     ),
+    verbose: bool = typer.Option(
+        False, "--verbose", "-v", help="Output logging on console"
+    ),
 ):
     """
     Test blockchain by running one trace
     """
 
-    test_trace(
-        trace.name, reactor if reactor is None else reactor.name, keypath, verbose
-    )
+    test_trace(trace, reactor, keypath, verbose)
 
 
 @app.command()
 def model(
-    model: typer.FileText = FileOption("model used to generate traces", "model"),
-    config: typer.FileText = FileOption("model configuration", "model"),
-    reactor: typer.FileText = FileOption("reactor to interpret the traces", "reactor"),
-    test: Optional[List[str]] = typer.Option(
+    model: Optional[Path] = FileOption("model used to generate traces", "model"),
+    config: Optional[Path] = FileOption("model configuration", "model"),
+    test: List[str] = typer.Option(
         None,
         show_default=False,
         help="model operator(s) describing test traces. multiple can be given either comma-separated, or via separate --test options",
     ),
+    reactor: Optional[Path] = FileOption("reactor to interpret the traces", "reactor"),
+    keypath: str = typer.Option(
+        "action",
+        show_default=True,
+        help="Path to key used as step name, extracted from ITF states",
+    ),
+    verbose: bool = typer.Option(
+        False, "--verbose", "-v", help="Output logging on console"
+    ),
 ):
     """
     Test blockchain by running multiple traces generated from a model
     """
-    tests = [ts.split(",") for ts in test]
-    tests = [t.strip() for ts in tests for t in ts]
-    print(
-        f"""
-    Model: {model}
-    Config: {config}
-    Reactor: {reactor}
-    Tests: {tests}
-    """
-    )
+    tests = [t.strip() for ts in test for t in ts.split(",")]
+
+    test_model(model, tests, reactor, keypath, verbose)
+
+    if model:
+        with AtomkraftConfig() as c:
+            c[MODEL_CONFIG_KEY] = str(model)