16
16
import json
17
17
import os
18
18
import tempfile
19
+ from typing import Dict
20
+ from typing import List
21
+ from typing import Optional
22
+ from typing import Union
19
23
20
24
21
25
import testinfra
@@ -87,7 +91,7 @@ def get_ansible_host(config, inventory, host, ssh_config=None, ssh_identity_file
87
91
port = config .get ("defaults" , "remote_port" , fallback = None )
88
92
if "ansible_port" in hostvars :
89
93
port = hostvars .get ("ansible_port" )
90
- kwargs = {}
94
+ kwargs : Dict [ str , Union [ str , bool ]] = {}
91
95
if hostvars .get ("ansible_become" , False ):
92
96
kwargs ["sudo" ] = True
93
97
kwargs ["sudo_user" ] = hostvars .get ("ansible_become_user" )
@@ -143,7 +147,7 @@ def is_empty_inventory(inventory):
143
147
144
148
145
149
class AnsibleRunner :
146
- _runners = {}
150
+ _runners : Dict [ Optional [ str ], "AnsibleRunner" ] = {}
147
151
_known_options = {
148
152
# Boolean arguments.
149
153
"become" : {
@@ -250,8 +254,8 @@ def options_to_cli(self, options):
250
254
args = {"become" : False , "check" : True }
251
255
args .update (options )
252
256
253
- cli = []
254
- cli_args = []
257
+ cli : List [ str ] = []
258
+ cli_args : List [ str ] = []
255
259
if verbose :
256
260
cli .append ("-" + "v" * verbose )
257
261
for arg_name , value in args .items ():
@@ -262,6 +266,7 @@ def options_to_cli(self, options):
262
266
if value :
263
267
cli .append (opt_cli )
264
268
elif opt_type == "string" :
269
+ assert isinstance (value , str )
265
270
cli .append (opt_cli + " %s" )
266
271
cli_args .append (value )
267
272
elif opt_type == "json" :
@@ -273,7 +278,7 @@ def options_to_cli(self, options):
273
278
return " " .join (cli ), cli_args
274
279
275
280
def run_module (self , host , module_name , module_args , ** options ):
276
- cmd , args = "ansible --tree %s" , [None ]
281
+ cmd , args = "ansible --tree %s" , []
277
282
if self .inventory_file :
278
283
cmd += " -i %s"
279
284
args += [self .inventory_file ]
@@ -289,7 +294,7 @@ def run_module(self, host, module_name, module_args, **options):
289
294
cmd += " %s"
290
295
args += [host ]
291
296
with tempfile .TemporaryDirectory () as d :
292
- args [ 0 ] = d
297
+ args . insert ( 0 , d )
293
298
out = local .run_expect ([0 , 2 , 8 ], cmd , * args )
294
299
files = os .listdir (d )
295
300
if not files and "skipped" in out .stdout .lower ():
0 commit comments