1# SPDX-License-Identifier: GPL-2.0 2 3import json as _json 4import subprocess 5 6class cmd: 7 def __init__(self, comm, shell=True, fail=True, ns=None, background=False): 8 if ns: 9 if isinstance(ns, NetNS): 10 ns = ns.name 11 comm = f'ip netns exec {ns} ' + comm 12 13 self.stdout = None 14 self.stderr = None 15 self.ret = None 16 17 self.comm = comm 18 self.proc = subprocess.Popen(comm, shell=shell, stdout=subprocess.PIPE, 19 stderr=subprocess.PIPE) 20 if not background: 21 self.process(terminate=False, fail=fail) 22 23 def process(self, terminate=True, fail=None): 24 if terminate: 25 self.proc.terminate() 26 stdout, stderr = self.proc.communicate() 27 self.stdout = stdout.decode("utf-8") 28 self.stderr = stderr.decode("utf-8") 29 self.proc.stdout.close() 30 self.proc.stderr.close() 31 self.ret = self.proc.returncode 32 33 if self.proc.returncode != 0 and fail: 34 if len(stderr) > 0 and stderr[-1] == "\n": 35 stderr = stderr[:-1] 36 raise Exception("Command failed: %s\n%s" % (self.proc.args, stderr)) 37 38 39def ip(args, json=None, ns=None): 40 cmd_str = "ip " 41 if json: 42 cmd_str += '-j ' 43 cmd_str += args 44 cmd_obj = cmd(cmd_str, ns=ns) 45 if json: 46 return _json.loads(cmd_obj.stdout) 47 return cmd_obj 48