本文整理汇总了Python中timedops.TimedSubprocess类的典型用法代码示例。如果您正苦于以下问题：Python TimedSubprocess类的具体用法？Python TimedSubprocess怎么用？Python TimedSubprocess使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了TimedSubprocess类的11个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: CheckSELinux
class CheckSELinux(Question):
def __init__ (self, troubleshooter):
Question.__init__ (self, troubleshooter, "Check SELinux contexts")
troubleshooter.new_page (Gtk.Label (), self)
def display (self):
self.answers = {}
#answers = self.troubleshooter.answers
RESTORECON = "/sbin/restorecon"
if not os.access (RESTORECON, os.X_OK):
return False
try:
import selinux
except ImportError:
return False
if not selinux.is_selinux_enabled():
return False
paths = ["/etc/cups/", "/usr/lib/cups/", "/usr/share/cups/"]
null = open ("/dev/null", "r+")
parent = self.troubleshooter.get_window ()
contexts = {}
new_environ = os.environ.copy()
new_environ['LC_ALL'] = "C"
restorecon_args = [RESTORECON, "-nvR"].extend(paths)
try:
# Run restorecon -nvR
self.op = TimedSubprocess (parent=parent,
args=restorecon_args,
close_fds=True,
env=new_environ,
stdin=null,
stdout=subprocess.PIPE,
stderr=null)
(restorecon_stdout, restorecon_stderr, result) = self.op.run ()
except:
# Problem executing command.
return False
for line in restorecon_stdout:
l = shlex.split (line)
if (len (l) < 1):
continue
contexts[l[2]] = l[4]
self.answers['selinux_contexts'] = contexts
return False
def collect_answer (self):
return self.answers
def cancel_operation (self):
self.op.cancel ()
开发者ID:daniel-dressler，项目名称:system-config-printer，代码行数:53，代码来源:CheckSELinux.py
示例2: VerifyPackages
class VerifyPackages(Question):
def __init__ (self, troubleshooter):
Question.__init__ (self, troubleshooter, "Verify packages")
troubleshooter.new_page (Gtk.Label (), self)
def display (self):
self.answers = {}
packages_verification = {}
package_manager="/bin/rpm"
if not os.access (package_manager, os.X_OK):
return False
packages = ["cups",
"foomatic",
"gutenprint",
"hpijs",
"hplip",
"system-config-printer"]
null = open ("/dev/null", "r+")
parent = self.troubleshooter.get_window ()
new_environ = os.environ.copy()
new_environ['LC_ALL'] = "C"
for package in packages:
verification_args = [package_manager, "-V", package]
try:
self.op = TimedSubprocess (parent=parent,
args=verification_args,
close_fds=True,
env=new_environ,
stdin=null,
stdout=subprocess.PIPE,
stderr=null)
(verif_stdout, verif_stderr, result) = self.op.run ()
except:
# Problem executing command.
return False
packages_verification[package] = verif_stdout[:-1]
self.answers['packages_verification'] = packages_verification
return False
def collect_answer (self):
return self.answers
def cancel_operation (self):
self.op.cancel ()
开发者ID:daniel-dressler，项目名称:system-config-printer，代码行数:49，代码来源:VerifyPackages.py
示例3: display
def display (self):
self.answers = {}
packages_verification = {}
package_manager="/bin/rpm"
if not os.access (package_manager, os.X_OK):
return False
packages = ["cups",
"foomatic",
"gutenprint",
"hpijs",
"hplip",
"system-config-printer"]
null = file ("/dev/null", "r+")
parent = self.troubleshooter.get_window ()
for package in packages:
verification_args = "LC_ALL=C " + package_manager + " -V " + package
try:
self.op = TimedSubprocess (parent=parent,
args=verification_args,
close_fds=True,
shell=True,
stdin=null,
stdout=subprocess.PIPE,
stderr=null)
(verif_stdout, verif_stderr, result) = self.op.run ()
except:
# Problem executing command.
return False
packages_verification[package] = verif_stdout[:-1]
self.answers['packages_verification'] = packages_verification
return False
开发者ID:Alberto-Beralix，项目名称:Beralix，代码行数:35，代码来源:VerifyPackages.py
示例4: display
def display (self):
self.answers = {}
#answers = self.troubleshooter.answers
RESTORECON = "/sbin/restorecon"
if not os.access (RESTORECON, os.X_OK):
return False
try:
import selinux
except ImportError:
return False
if not selinux.is_selinux_enabled():
return False
paths = ["/etc/cups/", "/usr/lib/cups/", "/usr/share/cups/"]
parent = self.troubleshooter.get_window ()
contexts = {}
new_environ = os.environ.copy()
new_environ['LC_ALL'] = "C"
restorecon_args = [RESTORECON, "-nvR"].extend(paths)
try:
# Run restorecon -nvR
self.op = TimedSubprocess (parent=parent,
args=restorecon_args,
close_fds=True,
env=new_environ,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL)
(restorecon_stdout, restorecon_stderr, result) = self.op.run ()
except:
# Problem executing command.
return False
for line in restorecon_stdout:
l = shlex.split (line)
if (len (l) < 1):
continue
contexts[l[2]] = l[4]
self.answers['selinux_contexts'] = contexts
return False
开发者ID:Distrotech，项目名称:system-config-printer，代码行数:41，代码来源:CheckSELinux.py
示例5: display
def display (self):
self.answers = {}
#answers = self.troubleshooter.answers
RESTORECON = "/sbin/restorecon"
if not os.access (RESTORECON, os.X_OK):
return False
try:
import selinux
except ImportError:
return False
if not selinux.is_selinux_enabled():
return False
paths = "/etc/cups/ /usr/lib/cups/ /usr/share/cups/"
null = file ("/dev/null", "r+")
parent = self.troubleshooter.get_window ()
contexts = {}
restorecon_args = "LC_ALL=C " + RESTORECON + " -nvR " + paths
try:
# Run restorecon -nvR
self.op = TimedSubprocess (parent=parent,
args=restorecon_args,
close_fds=True,
shell=True,
stdin=null,
stdout=subprocess.PIPE,
stderr=null)
(restorecon_stdout, restorecon_stderr, result) = self.op.run ()
except:
# Problem executing command.
return False
for line in restorecon_stdout:
l = shlex.split (line)
if (len (l) < 1):
continue
contexts[l[2]] = l[4]
self.answers['selinux_contexts'] = contexts
return False
开发者ID:Alberto-Beralix，项目名称:Beralix，代码行数:40，代码来源:CheckSELinux.py
示例6: display
def display (self):
self.answers = {}
packages_verification = {}
package_manager="/bin/rpm"
if not os.access (package_manager, os.X_OK):
return False
packages = ["cups",
"foomatic",
"gutenprint",
"hpijs",
"hplip",
"system-config-printer"]
parent = self.troubleshooter.get_window ()
new_environ = os.environ.copy()
new_environ['LC_ALL'] = "C"
for package in packages:
verification_args = [package_manager, "-V", package]
try:
self.op = TimedSubprocess (parent=parent,
args=verification_args,
close_fds=True,
env=new_environ,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL)
(verif_stdout, verif_stderr, result) = self.op.run ()
except:
# Problem executing command.
return False
packages_verification[package] = verif_stdout[:-1]
self.answers['packages_verification'] = packages_verification
return False
开发者ID:Distrotech，项目名称:system-config-printer，代码行数:37，代码来源:VerifyPackages.py
示例7: display
#.........这里部分代码省略.........
'http',
'https']):
if answers.get ('cups_device_uri_scheme') == 'https':
encryption = cups.HTTP_ENCRYPT_REQUIRED
else:
encryption = cups.HTTP_ENCRYPT_IF_REQUESTED
try:
self.op = TimedOperation (cups.Connection,
kwargs={"host": server_name,
"port": server_port,
"encryption": encryption},
parent=parent)
c = self.op.run ()
ipp_connect = True
except RuntimeError:
ipp_connect = False
self.answers['remote_server_connect_ipp'] = ipp_connect
if ipp_connect:
try:
self.op = TimedOperation (c.getPrinters, parent=parent)
self.op.run ()
cups_server = True
except:
cups_server = False
self.answers['remote_server_cups'] = cups_server
if cups_server:
cups_printer_dict = answers.get ('cups_printer_dict', {})
uri = cups_printer_dict.get ('device-uri', None)
if uri:
try:
self.op = TimedOperation (c.getPrinterAttributes,
kwargs={"uri": uri},
parent=parent)
attr = self.op.run ()
self.answers['remote_cups_queue_attributes'] = attr
except:
pass
if try_connect:
# Try to see if we can connect using smbc.
context = None
try:
context = smbc.Context ()
name = self.answers['remote_server_try_connect']
self.op = TimedOperation (context.opendir,
args=("smb://%s/" % name,),
parent=parent)
dir = self.op.run ()
self.op = TimedOperation (dir.getdents, parent=parent)
shares = self.op.run ()
self.answers['remote_server_smb'] = True
self.answers['remote_server_smb_shares'] = shares
except NameError:
# No smbc support
pass
except RuntimeError as e:
(e, s) = e.args
self.answers['remote_server_smb_shares'] = (e, s)
if context != None and answers.has_key ('cups_printer_dict'):
uri = answers['cups_printer_dict'].get ('device-uri', '')
u = smburi.SMBURI (uri)
(group, host, share, user, password) = u.separate ()
accessible = False
try:
self.op = TimedOperation (context.open,
args=("smb://%s/%s" % (host,
share),
os.O_RDWR,
0777),
parent=parent)
f = self.op.run ()
accessible = True
except RuntimeError as e:
(e, s) = e.args
accessible = (e, s)
self.answers['remote_server_smb_share_anon_access'] = accessible
# Try traceroute if we haven't already.
if (try_connect and
not answers.has_key ('remote_server_traceroute')):
try:
self.op = TimedSubprocess (parent=parent, close_fds=True,
args=['traceroute', '-w', '1',
server_name],
stdin=file("/dev/null"),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
self.answers['remote_server_traceroute'] = self.op.run ()
except:
# Problem executing command.
pass
return False
开发者ID:incrazyboyy，项目名称:RPiTC-X，代码行数:101，代码来源:CheckNetworkServerSanity.py
示例8: display
def display (self):
self.answers = {}
answers = self.troubleshooter.answers
if not answers['cups_queue_listed']:
return False
parent = self.troubleshooter.get_window ()
name = answers['cups_queue']
tmpf = None
try:
cups.setServer ('')
self.op = TimedOperation (cups.Connection, parent=parent)
c = self.op.run ()
self.op = TimedOperation (c.getPPD, args=(name,), parent=parent)
tmpf = self.op.run ()
except RuntimeError:
return False
except cups.IPPError:
return False
self.install_button.hide ()
title = None
text = None
try:
ppd = cups.PPD (tmpf)
self.answers['cups_printer_ppd_valid'] = True
def options (options_list):
o = {}
for option in options_list:
o[option.keyword] = option.defchoice
return o
defaults = {}
for group in ppd.optionGroups:
g = options (group.options)
for subgroup in group.subgroups:
g[subgroup.name] = options (subgroup.options)
defaults[group.name] = g
self.answers['cups_printer_ppd_defaults'] = defaults
except RuntimeError:
title = _("Invalid PPD File")
self.answers['cups_printer_ppd_valid'] = False
try:
self.op = TimedSubprocess (parent=parent,
args=['cupstestppd', '-rvv', tmpf],
close_fds=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
result = self.op.run ()
self.answers['cupstestppd_output'] = result
text = _("The PPD file for printer '%s' does not conform "
"to the specification. "
"Possible reason follows:") % name
text += '\n' + reduce (lambda x, y: x + '\n' + y, result[0])
except OSError:
# Perhaps cupstestppd is not in the path.
text = _("There is a problem with the PPD file for "
"printer '%s'.") % name
if tmpf:
os.unlink (tmpf)
if title == None and not answers['cups_printer_remote']:
(pkgs, exes) = cupshelpers.missingPackagesAndExecutables (ppd)
self.answers['missing_pkgs_and_exes'] = (pkgs, exes)
if len (pkgs) > 0 or len (exes) > 0:
title = _("Missing Printer Driver")
if len (pkgs) > 0:
try:
self.packagekit = installpackage.PackageKit ()
except:
pkgs = []
if len (pkgs) > 0:
self.package = pkgs[0]
text = _("Printer '%s' requires the %s package but it "
"is not currently installed.") % (name,
self.package)
self.install_button.show ()
else:
text = _("Printer '%s' requires the '%s' program but it "
"is not currently installed.") % (name,
(exes + pkgs)[0])
if title != None:
self.label.set_markup ('<span weight="bold" size="larger">' +
title + '</span>\n\n' + text)
return title != None
开发者ID:Distrotech，项目名称:system-config-printer，代码行数:92，代码来源:CheckPPDSanity.py
示例9: CheckPPDSanity
class CheckPPDSanity(Question):
def __init__ (self, troubleshooter):
Question.__init__ (self, troubleshooter, "Check PPD sanity")
vbox = Gtk.VBox ()
vbox.set_border_width (12)
vbox.set_spacing (12)
self.label = Gtk.Label ()
self.label.set_line_wrap (True)
self.label.set_use_markup (True)
self.label.set_alignment (0, 0)
vbox.pack_start (self.label, False, False, 0)
box = Gtk.HButtonBox ()
box.set_layout (Gtk.ButtonBoxStyle.START)
self.install_button = Gtk.Button.new_with_label (_("Install"))
box.add (self.install_button)
# Although we want this hidden initially,
# troubleshooter.new_page will call show_all() on the widget
# we give it. We'll need to hide this button in the display()
# callback instead.
vbox.pack_start (box, False, False, 0)
troubleshooter.new_page (vbox, self)
def display (self):
self.answers = {}
answers = self.troubleshooter.answers
if not answers['cups_queue_listed']:
return False
parent = self.troubleshooter.get_window ()
name = answers['cups_queue']
tmpf = None
try:
cups.setServer ('')
self.op = TimedOperation (cups.Connection, parent=parent)
c = self.op.run ()
self.op = TimedOperation (c.getPPD, args=(name,), parent=parent)
tmpf = self.op.run ()
except RuntimeError:
return False
except cups.IPPError:
return False
self.install_button.hide ()
title = None
text = None
try:
ppd = cups.PPD (tmpf)
self.answers['cups_printer_ppd_valid'] = True
def options (options_list):
o = {}
for option in options_list:
o[option.keyword] = option.defchoice
return o
defaults = {}
for group in ppd.optionGroups:
g = options (group.options)
for subgroup in group.subgroups:
g[subgroup.name] = options (subgroup.options)
defaults[group.name] = g
self.answers['cups_printer_ppd_defaults'] = defaults
except RuntimeError:
title = _("Invalid PPD File")
self.answers['cups_printer_ppd_valid'] = False
try:
self.op = TimedSubprocess (parent=parent,
args=['cupstestppd', '-rvv', tmpf],
close_fds=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
result = self.op.run ()
self.answers['cupstestppd_output'] = result
text = _("The PPD file for printer '%s' does not conform "
"to the specification. "
"Possible reason follows:") % name
text += '\n' + reduce (lambda x, y: x + '\n' + y, result[0])
except OSError:
# Perhaps cupstestppd is not in the path.
text = _("There is a problem with the PPD file for "
"printer '%s'.") % name
if tmpf:
os.unlink (tmpf)
if title == None and not answers['cups_printer_remote']:
(pkgs, exes) = cupshelpers.missingPackagesAndExecutables (ppd)
self.answers['missing_pkgs_and_exes'] = (pkgs, exes)
if len (pkgs) > 0 or len (exes) > 0:
title = _("Missing Printer Driver")
if len (pkgs) > 0:
try:
self.packagekit = installpackage.PackageKit ()
except:
pkgs = []
#.........这里部分代码省略.........
开发者ID:Distrotech，项目名称:system-config-printer，代码行数:101，代码来源:CheckPPDSanity.py
示例10: display
def display (self):
# Collect information useful for the various checks.
self.answers = {}
answers = self.troubleshooter.answers
if not answers['cups_queue_listed']:
return False
name = answers['cups_queue']
parent = self.troubleshooter.get_window ()
# Find out if this is a printer or a class.
try:
cups.setServer ('')
c = TimedOperation (cups.Connection, parent=parent).run ()
printers = TimedOperation (c.getPrinters, parent=parent).run ()
if printers.has_key (name):
self.answers['is_cups_class'] = False
queue = printers[name]
self.answers['cups_printer_dict'] = queue
else:
self.answers['is_cups_class'] = True
classes = TimedOperation (c.getClasses, parent=parent).run ()
queue = classes[name]
self.answers['cups_class_dict'] = queue
attrs = TimedOperation (c.getPrinterAttributes, (name,),
parent=parent).run ()
self.answers['local_cups_queue_attributes'] = attrs
except:
pass
if self.answers.has_key ('cups_printer_dict'):
cups_printer_dict = self.answers['cups_printer_dict']
uri = cups_printer_dict['device-uri']
(scheme, rest) = urllib.splittype (uri)
self.answers['cups_device_uri_scheme'] = scheme
if scheme in ["ipp", "http", "https"]:
(hostport, rest) = urllib.splithost (rest)
(host, port) = urllib.splitnport (hostport, defport=631)
self.answers['remote_server_name'] = host
self.answers['remote_server_port'] = port
elif scheme == "smb":
u = smburi.SMBURI (uri)
(group, host, share, user, password) = u.separate ()
new_environ = os.environ.copy()
new_environ['LC_ALL'] = "C"
if group:
args = ["nmblookup", "-W", group, host]
else:
args = ["nmblookup", host]
try:
p = TimedSubprocess (parent=parent,
timeout=5000,
args=args,
env=new_environ,
close_fds=True,
stdin=file("/dev/null"),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
result = p.run ()
self.answers['nmblookup_output'] = result
for line in result[0]:
if line.startswith ("querying"):
continue
spc = line.find (' ')
if (spc != -1 and
not line[spc:].startswith (" failed ")):
# Remember the IP address.
self.answers['remote_server_name'] = line[:spc]
break
except OSError:
# Problem executing command.
pass
elif scheme == "hp":
new_environ = os.environ.copy()
new_environ['LC_ALL'] = "C"
new_environ['DISPLAY'] = ""
try:
p = TimedSubprocess (parent=parent,
timeout=3000,
args=["hp-info", "-d" + uri],
close_fds=True,
env=new_environ,
stdin=file("/dev/null"),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
self.answers['hplip_output'] = p.run ()
except OSError:
# Problem executing command.
pass
r = cups_printer_dict['printer-type'] & cups.CUPS_PRINTER_REMOTE
self.answers['cups_printer_remote'] = (r != 0)
return False
开发者ID:thnguyn2，项目名称:ECE_527_MP，代码行数:97，代码来源:CheckPrinterSanity.py
示例11: display
def display (self):
self.answers = {}
answers = self.troubleshooter.answers
if answers['cups_queue_listed']:
if answers['is_cups_class']:
return False
cups_printer_dict = answers['cups_printer_dict']
device_uri = cups_printer_dict['device-uri']
elif answers.get ('cups_device_listed', False):
device_uri = answers['cups_device_uri']
else:
return False
(scheme, rest) = urllib.splittype (device_uri)
if scheme not in ['hp', 'hpfax', 'usb', 'hal']:
return False
LSUSB = "/sbin/lsusb"
if not os.access (LSUSB, os.X_OK):
return False
GETFACL = "/usr/bin/getfacl"
if not os.access (GETFACL, os.X_OK):
return False
new_environ = os.environ.copy()
new_environ['LC_ALL'] = "C"
# Run lsusb
parent = self.troubleshooter.get_window ()
try:
self.op = TimedSubprocess (parent=parent,
args=[LSUSB, "-v"],
close_fds=True,
env=new_environ,
stdin=file("/dev/null"),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
(lsusb_stdout, lsusb_stderr, result) = self.op.run ()
except:
# Problem executing command.
return False
# Now parse it.
dev_by_id = {}
this_dev = None
for line in lsusb_stdout:
if (this_dev != None and
((line.find ("bInterfaceClass") != -1 and
line.find ("7 Printer") != -1) or
(line.find ("bInterfaceSubClass") != -1 and
line.find ("1 Printer") != -1))):
mfr = dev_by_id.get (this_mfr_id, {})
mdl = mfr.get (this_mdl_id, [])
mdl.append (this_dev)
mfr[this_mdl_id] = mdl
dev_by_id[this_mfr_id] = mfr
this_dev = None
continue
separators = [ ('Bus ', 3),
(' Device ', 3),
(': ID ', 4),
(':', 4),
(' ', -1)]
fields = []
i = 0
p = line
while i < len (separators):
(sep, length) = separators[i]
if not p.startswith (sep):
break
start = len (sep)
if length == -1:
end = len (p)
fields.append (p[start:])
else:
end = start + length
fields.append (p[start:end])
p = p[end:]
i += 1
if i < len (separators):
continue
if not scheme.startswith ('hp') and fields[2] != '03f0':
# Skip non-HP printers if we know we're using HPLIP.
continue
this_dev = { 'bus': fields[0],
'dev': fields[1],
'name': fields[4],
'full': line }
this_mfr_id = fields[2]
this_mdl_id = fields[3]
infos = {}
paths = []
#.........这里部分代码省略.........
开发者ID:incrazyboyy，项目名称:RPiTC-X，代码行数:101，代码来源:CheckUSBPermissions.py
注：本文中的timedops.TimedSubprocess类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论