File indexing completed on 2026-04-27 07:28:14
0001 """Tests for experimental stacks and layers."""
0002
0003 from dataclasses import dataclass, field
0004 from typing import Dict, Any, List
0005 from aid2e.utilities.workflows import (
0006 AnaLayer,
0007 ExperimentStack,
0008 StackLayer
0009 )
0010
0011
0012 def _make_ana_layer_payload() -> Dict[str, Any]:
0013 """Build a dictionary of AnaLayer details, inputs, outputs, arguments for test."""
0014 return {
0015 "ana_details": {
0016 "name": "test_ana",
0017 "command": "python run_test_ana.py",
0018 "rule": "{command} {arguments} -I {inputs} -O {outputs}",
0019 },
0020 "ana_input": [
0021 "ana_input.root",
0022 ],
0023 "ana_output": [
0024 "ana_output.root",
0025 ],
0026 "ana_args": [
0027 "-P 22",
0028 ],
0029 }
0030
0031
0032 def _make_experiment_stack_payload() -> Dict[str, Any]:
0033 """Build a dictionary of stack details, inputs, outputs for test"""
0034 return {
0035 "sim_details" : {
0036 "name" : "sim",
0037 "command" : "dosim",
0038 "rule" : '{command} {arguments} -I {inputs} -O {outputs}'
0039 },
0040 "sim_input" : [
0041 "sim_input.root",
0042 ],
0043 "sim_output" : [
0044 "sim_output.root",
0045 ],
0046 "sim_args" : [
0047 "--physicsList QBert.yaml",
0048 ]
0049 }
0050
0051
0052 def test_ana_layer():
0053 """Validate AnaLayer functionality"""
0054 payload = _make_ana_layer_payload()
0055 analayer = AnaLayer()
0056 analayer.name = payload["ana_details"]["name"]
0057 analayer.command = payload["ana_details"]["command"]
0058 analayer.rule = payload["ana_details"]["rule"]
0059 assert analayer.name == payload["ana_details"]["name"]
0060 assert analayer.command == payload["ana_details"]["command"]
0061 assert analayer.rule == payload["ana_details"]["rule"]
0062
0063 command = analayer.make_command(
0064 payload["ana_input"],
0065 payload["ana_output"],
0066 payload["ana_args"]
0067 )
0068 assert command == "python run_test_ana.py -P 22 -I ana_input.root -O ana_output.root"
0069
0070
0071 def test_experiment_stack():
0072 """Validate ExperimentStack functionality"""
0073 payload = _make_experiment_stack_payload()
0074 class MySimLayer(StackLayer):
0075 name = payload["sim_details"]["name"]
0076 command = payload["sim_details"]["command"]
0077 rule = payload["sim_details"]["rule"]
0078 def _make_input_arg(self, inputs: List[str]) -> str:
0079 return ' '.join(inputs)
0080 def _make_output_arg(self, outputs: List[str]) -> str:
0081 return ' '.join(outputs)
0082
0083 @dataclass
0084 class MyExperimentStack(ExperimentStack):
0085 sim: MySimLayer = field(default_factory = MySimLayer)
0086
0087 mystack = MyExperimentStack()
0088 assert isinstance(mystack, MyExperimentStack)
0089 assert isinstance(mystack[payload["sim_details"]["name"]], MySimLayer)
0090
0091 command = mystack[payload["sim_details"]["name"]].make_command(
0092 payload["sim_input"],
0093 payload["sim_output"],
0094 payload["sim_args"]
0095 )
0096 assert command == "dosim --physicsList QBert.yaml -I sim_input.root -O sim_output.root"
