Move ext_command_thread from gPodderLib to util

We want to get rid of gPodderLib in the end, so move
the utility function out of gPodderLib and correct
its uses accordingly. Also add a unittest for it.
This commit is contained in:
Thomas Perl 2009-04-01 12:53:13 +02:00
parent e3a8795a3e
commit aa4eb0b4d0
4 changed files with 38 additions and 38 deletions

View File

@ -449,7 +449,7 @@ class DownloadTask(object):
os.environ["GPODDER_EPISODE_PUBDATE"]=str(int(self.__episode.pubDate))
os.environ["GPODDER_EPISODE_LINK"]=self.__episode.link or ''
os.environ["GPODDER_EPISODE_DESC"]=self.__episode.description or ''
threading.Thread(target=gl.ext_command_thread, args=(gl.config.cmd_download_complete,)).start()
util.run_external_command(gl.config.cmd_download_complete)
except DownloadCancelledException:
log('Download has been cancelled/paused: %s', self, sender=self)
if self.status == DownloadTask.CANCELLED:

View File

@ -847,8 +847,7 @@ class gPodder(GladeWidget, dbus.service.Object):
elif last_download_count > 0:
log('All downloads have finished.', sender=self)
if gl.config.cmd_all_downloads_complete:
Thread(target=gl.ext_command_thread, args=( \
gl.config.cmd_all_downloads_complete)).start()
util.run_external_command(gl.config.cmd_all_downloads_complete)
self.gPodder.set_title(' - '.join(title))

View File

@ -389,40 +389,6 @@ class gPodderLib(object):
return ( False, command_line[0] )
return ( True, command_line[0] )
def ext_command_thread(self, command_line):
"""
This is the function that will be called in a separate
thread that will call an external command (specified by
command_line). In case of problem (i.e. the command has
not been found or there has been another error), we will
call the notification function with two arguments - the
first being the error message and the second being the
title to be used for the error message.
"""
log("(ExtCommandThread) Excuting command Line [%s]", command_line)
p = subprocess.Popen(command_line, shell=True, stdout=sys.stdout, stderr=sys.stderr)
result = p.wait()
# FIXME: NOTIFICATION
def notification(message, title):
log('Message: %s (%s)', title, message, sender=self)
if result == 127:
title = _('User command not found')
message = _('The user command [%s] was not found.\nPlease check your user command settings in the preferences dialog.' % command_line)
notification(message, title)
elif result == 126:
title = _('User command permission denied')
message = _('Permission denied when trying to execute user command [%s].\nPlease check that you have permissions to execute this command.' % command_line)
notification(message, title)
elif result > 0 :
title = _('User command returned an error')
message = _('The user command [%s] returned an error code of [%d]' % (command_line,result))
notification(message, title)
log("(ExtCommandThread) Finished command line [%s] result [%d]",command_line,result)
class HistoryStore( types.ListType):

View File

@ -1105,7 +1105,7 @@ def find_mount_point(directory):
...
ValueError: Directory names should be of type str.
>>> from minimock import mock
>>> from minimock import mock, restore
>>> mocked_mntpoints = ('/', '/home', '/media/usbdisk', '/media/cdrom')
>>> mock('os.path.ismount', returns_func=lambda x: x in mocked_mntpoints)
>>>
@ -1142,6 +1142,7 @@ def find_mount_point(directory):
Called os.path.ismount('/media/usbdisk/blubb')
Called os.path.ismount('/media/usbdisk')
'/media/usbdisk'
>>> restore()
"""
if isinstance(directory, unicode):
# We do not accept unicode strings, because they could fail when
@ -1257,3 +1258,37 @@ def relpath(p1, p2):
return os.path.join(*p)
def run_external_command(command_line):
"""
This is the function that will be called in a separate
thread that will call an external command (specified by
command_line). In case of problem (i.e. the command has
not been found or there has been another error), we will
call the notification function with two arguments - the
first being the error message and the second being the
title to be used for the error message.
>>> from minimock import mock, Mock, restore
>>> mock('subprocess.Popen', returns=Mock('subprocess.Popen'))
>>> run_external_command('testprogramm')
Called subprocess.Popen('testprogramm', shell=True)
Called subprocess.Popen.wait()
>>> restore()
"""
def open_process(command_line):
log('Running external command: %s', command_line)
p = subprocess.Popen(command_line, shell=True)
result = p.wait()
if result == 127:
log('Command not found: %s', command_line)
elif result == 126:
log('Command permission denied: %s', command_line)
elif result > 0:
log('Command returned an error (%d): %s', result, command_line)
else:
log('Command finished successfully: %s', command_line)
threading.Thread(target=open_process, args=(command_line,)).start()