50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
import shlex
|
|
import subprocess as sp
|
|
import typing
|
|
from dataclasses import dataclass
|
|
|
|
from jinja2 import Template
|
|
|
|
from qbit_ci.template import path_exists
|
|
from qbit_ci.template import template_env
|
|
|
|
StateMap = typing.Mapping[str, typing.Any]
|
|
StepCondition = typing.Callable[[StateMap], bool]
|
|
|
|
def expr_to_step_condition(expr: str) -> StepCondition:
|
|
def inner(state_map: StateMap) -> bool:
|
|
templ: Template = template_env.from_string(expr)
|
|
result = templ.render(state_map)
|
|
return result.lower() == "true"
|
|
|
|
return inner
|
|
|
|
|
|
class GenericStep(typing.Protocol):
|
|
name: str
|
|
|
|
def should_invoke(self, state_map: StateMap) -> bool:
|
|
...
|
|
|
|
def invoke(self, state_map: StateMap) -> int:
|
|
...
|
|
|
|
@dataclass
|
|
class CommandStep:
|
|
name: str
|
|
commands: typing.Sequence[str]
|
|
conditions: typing.Sequence[StepCondition]
|
|
|
|
def should_invoke(self, state_map: StateMap) -> bool:
|
|
return all(cond(state_map) for cond in self.conditions)
|
|
|
|
|
|
def invoke(self, state_map: StateMap) -> int:
|
|
cmd_fmt = " && ".join(self.commands)
|
|
cmd_template: Template = template_env.from_string(cmd_fmt)
|
|
cmd = cmd_template.render(state_map)
|
|
resp = sp.Popen(shlex.split(cmd, posix=True))
|
|
return resp.wait()
|
|
|
|
__all__ = ("expr_to_step_condition", "GenericStep", "CommandStep", "StepCondition")
|