1010from automax .core .capabilities import plugin_tools
1111from automax .core .models import ExecutionContext , PluginResult
1212from automax .plugins .base import BasePlugin , PluginValidationError
13- from automax .plugins .remote_utils import exec_remote , predicate_result_from_remote , quote
13+ from automax .plugins .remote_utils import exec_remote , quote
1414
1515
1616def _as_list (value : Any ) -> list [str ]:
@@ -21,70 +21,6 @@ def _as_list(value: Any) -> list[str]:
2121 return [str (value )]
2222
2323
24- class ToolCheckPlugin (BasePlugin ):
25- name = "os.tool.check"
26- description = "Check whether one executable exists on the remote PATH."
27- required_params = ("name" ,)
28- optional_params = ("path" ,)
29- opens_remote_session = True
30- supports_check_mode = True
31-
32- def diff_preview_reason (self , params : Dict [str , Any ], context : ExecutionContext ) -> str :
33- return "os.tool.check is a read-only remote dependency check"
34-
35- def manual_commands (self , params : Dict [str , Any ], context : ExecutionContext ) -> list [str ]:
36- self .validate (params )
37- path_prefix = f"PATH={ quote (params ['path' ])} :$PATH " if params .get ("path" ) else ""
38- return [f"{ path_prefix } command -v { quote (params ['name' ])} " ]
39-
40- def execute (self , params : Dict [str , Any ], context : ExecutionContext ) -> PluginResult :
41- rc , out , err = exec_remote (context , self .manual_commands (params , context )[0 ])
42- return predicate_result_from_remote (
43- rc = rc ,
44- stdout = out ,
45- stderr = err ,
46- message = f"os.tool.check failed for tool: { params ['name' ]} " ,
47- data_key = "exists" ,
48- data = {"tool" : params ["name" ], "path" : out .strip () if rc == 0 else "" },
49- )
50-
51-
52- class ToolVersionAssertPlugin (BasePlugin ):
53- name = "os.tool.version_check"
54- description = "Check whether a remote tool version output contains or matches the expected value."
55- required_params = ("name" ,)
56- optional_params = ("version_arg" , "contains" , "regex" )
57- opens_remote_session = True
58- supports_check_mode = True
59-
60- def validate (self , params : Dict [str , Any ]) -> None :
61- super ().validate (params )
62- if not params .get ("contains" ) and not params .get ("regex" ):
63- raise PluginValidationError ("os.tool.version_check requires contains or regex" )
64-
65- def diff_preview_reason (self , params : Dict [str , Any ], context : ExecutionContext ) -> str :
66- return "os.tool.version_check is a read-only remote dependency check"
67-
68- def manual_commands (self , params : Dict [str , Any ], context : ExecutionContext ) -> list [str ]:
69- self .validate (params )
70- arg = str (params .get ("version_arg" , "--version" ))
71- command = f"{ quote (params ['name' ])} { arg } 2>&1"
72- if params .get ("contains" ):
73- return [f"{ command } | grep -F -- { quote (params ['contains' ])} " ]
74- return [f"{ command } | grep -E -- { quote (params ['regex' ])} " ]
75-
76- def execute (self , params : Dict [str , Any ], context : ExecutionContext ) -> PluginResult :
77- rc , out , err = exec_remote (context , self .manual_commands (params , context )[0 ])
78- return predicate_result_from_remote (
79- rc = rc ,
80- stdout = out ,
81- stderr = err ,
82- message = f"os.tool.version_check failed for tool: { params ['name' ]} " ,
83- data_key = "matches" ,
84- data = {"tool" : params ["name" ], "version" : out .strip () if rc == 0 else "" },
85- )
86-
87-
8824class CapabilityAssertPlugin (BasePlugin ):
8925 name = "os.capability.check"
9026 description = "Check remote tools, paths and optional shell checks required by a job preflight."
0 commit comments