Differential D27

aligned to 80 column
AbandonedPublic
Actions

Authored by garretraziel on Mar 20 2014, 11:04 AM.

Details

Reviewers
kparal
Maniphest Tasks
T96: Wrap lines at 80th column + other pylint and pyflakes fixes
Summary

I have wrapped all lines to 80 column.

Test Plan

run py.test --functional

Diff Detail

Branch
feature/T96-wraplines
Lint
No Linters Available
Unit
No Unit Test Coverage
kparal requested changes to this revision.Mar 20 2014, 12:50 PM

Incorrectly wrapped files (lines longer than 80 characters):
koji_directive.py
python_directive.py
koji_utils.py
test_python_directive.py

Please fix it and then either update it here, or close this patch and update D25 instead (that would be better I think).

garretraziel abandoned this revision.Mar 25 2014, 1:34 PM

Updated D25.

Revision Contents

Diff 76

View Options

libtaskotron/directives/createrepo_directive.py

1import subprocess as sub 1​import subprocess as sub
2from libtaskotron.directives import BaseDirective 2​from libtaskotron.directives import BaseDirective
3from libtaskotron.logger import log 3​from libtaskotron.logger import log
4 4​
5directive_class = 'CreaterepoDirective' 5​directive_class = 'CreaterepoDirective'
6 6​
7# TODO handle exceptions properly (T85) 7​# TODO handle exceptions properly (T85)
8 8​
9class CreaterepoDirective(BaseDirective): 9​class CreaterepoDirective(BaseDirective):
10 def process(self, input_data, env_data): 10​ def process(self, input_data, env_data):
11 """For the createrepo directive, we're expecting a yaml declaration of the form 11​ """For the createrepo directive, we're expecting a yaml declaration of
12 "createrepo: repodir=/path/to/some/dir" 12​ the form "createrepo: repodir=/path/to/some/dir"
13 """ 13​ """
14 14​
15 repodir = input_data['repodir'] 15​ repodir = input_data['repodir']
16 16​
17 log.info('running createrepo on %s', repodir) 17​ log.info('running createrepo on %s', repodir)
18 18​
19 p = sub.Popen(['createrepo', repodir], stdout=sub.PIPE, stderr=sub.PIPE) 19​ p = sub.Popen(['createrepo', repodir], stdout=sub.PIPE, stderr=sub.PIPE)
20 20​
21 output, errors = p.communicate() 21​ output, errors = p.communicate()
22 22​
23 if p.returncode: 23​ if p.returncode:
24 raise Exception(errors) 24​ raise Exception(errors)
25 25​
26 log.debug(output) 26​ log.debug(output)
27 27​
28 return output 28​ return output
View Options

libtaskotron/directives/koji_directive.py

1from libtaskotron.koji_utils import KojiClient 1​from libtaskotron.koji_utils import KojiClient
2from libtaskotron.logger import log 2​from libtaskotron.logger import log
3from libtaskotron.directives import BaseDirective 3​from libtaskotron.directives import BaseDirective
4 4​
5directive_class = 'KojiDirective' 5​directive_class = 'KojiDirective'
6 6​
7class KojiDirective(BaseDirective): 7​class KojiDirective(BaseDirective):
8 """The koji directive interfaces with Koji to facilitate various koji actions. 8​ """The koji directive interfaces with Koji to facilitate various koji
9​ actions.
9 10​
10 format: "koji: command=[download] <command args> 11​ format: "koji: command=[download] <command args>
11 12​
12 Valid Commands: 13​ Valid Commands:
13 14​
14 download 15​ download
15 Downloads the envr specified as "envr" 16​ Downloads the envr specified as "envr"
16 format: "koji: command=download envr=<envr to download> arch=<desired arch> 17​ format: "koji: command=download envr=<envr to download> arch=<desired arch>
17 download_tag 18​ download_tag
18 Downloads rpms tagged by "tag" 19​ Downloads rpms tagged by "tag"
19 format: "koji: command=download_tag tag=<tag to download> arch=<desired arch> 20​ format: "koji: command=download_tag tag=<tag to download> arch=<desired arch>
20 """ 21​ """
21 22​
22 def __init__(self, koji_session=None): 23​ def __init__(self, koji_session=None):
23 if koji_session == None: 24​ super(KojiDirective, self).__init__()
25​ if koji_session is None:
24 self.koji = KojiClient(koji_session) 26​ self.koji = KojiClient(koji_session)
25 else: 27​ else:
26 self.koji = koji_session 28​ self.koji = koji_session
27 29​
28 def process(self, input_data, env_data): 30​ def process(self, input_data, env_data):
29 valid_actions = ['download', 'download_tag'] 31​ valid_actions = ['download', 'download_tag']
30 32​
31 output_data = {} 33​ output_data = {}
32 34​
33 action = input_data['action'] 35​ action = input_data['action']
34 36​
35 if not action in valid_actions: 37​ if not action in valid_actions:
36 raise Exception('%s is not a valid command for koji helper' % action) 38​ raise Exception('%s is not a valid command for koji helper' %
39​ action)
37 40​
38 if action == 'download': 41​ if action == 'download':
39 if 'envr' not in input_data or 'arch' not in input_data: 42​ if 'envr' not in input_data or 'arch' not in input_data:
40 # TODO T85 43​ # TODO T85
41 detected_args = ', '.join(input_data.keys()) 44​ detected_args = ', '.join(input_data.keys())
42 raise Exception("The koji directive requires both 'envr' and \ 45​ raise Exception("The koji directive requires both 'envr' and "
43 'arch' arguments. Detected arguments: %s" % detected_args) 46​ "'arch' arguments. Detected arguments: %s" % detected_args)
44 47​
45 tmpdir = env_data['workdir'] 48​ tmpdir = env_data['workdir']
46 envr = input_data['envr'] 49​ envr = input_data['envr']
47 arches = [input_data['arch']] 50​ arches = [input_data['arch']]
48 51​
49 log.info("getting koji builds for %s (%s) and downloading to %s", envr, arches, tmpdir) 52​ log.info("getting koji builds for %s (%s) and downloading to %s",
50 output_data['downloaded_rpms'] = self.koji.get_nvr_rpms(envr, 53​ envr, arches, tmpdir)
51 tmpdir, 54​ output_data['downloaded_rpms'] = self.koji.get_nvr_rpms(envr, tmpdir,
52 arches) 55​ arches)
53 elif action == 'download_tag': 56​ elif action == 'download_tag':
54 if 'tag' not in input_data or 'arch' not in input_data: 57​ if 'tag' not in input_data or 'arch' not in input_data:
55 # TODO T85 58​ # TODO T85
56 detected_args = ', '.join(input_data.keys()) 59​ detected_args = ', '.join(input_data.keys())
57 raise Exception("The koji directive requires both 'tag' and \ 60​ raise Exception("The koji directive requires both 'tag' and "
58 'arch' arguments. Detected arguments: %s" % detected_args) 61​ "'arch' arguments. Detected arguments: %s" % detected_args)
59 62​
60 tmpdir = env_data['workdir'] 63​ tmpdir = env_data['workdir']
61 tag = input_data['tag'] 64​ tag = input_data['tag']
62 arches = [input_data['arch']] 65​ arches = [input_data['arch']]
63 66​
64 log.info("getting koji builds for tag %s and downloading to %s", tag, tmpdir) 67​ log.info("getting koji builds for tag %s and downloading to %s",
65 output_data['downloaded_rpms'] = self.koji.get_tagged_rpms(tag, tmpdir, arches) 68​ tag, tmpdir)
69​ output_data['downloaded_rpms'] = self.koji.get_tagged_rpms(tag, tmpdir,
70​ arches)
66 71​
67 return output_data 72​ return output_data
View Options

libtaskotron/directives/python_directive.py

1import imp 1​import imp
2import os 2​import os
3 3​
4from libtaskotron.directives import BaseDirective 4​from libtaskotron.directives import BaseDirective
5 5​
6directive_class = 'PythonDirective' 6​directive_class = 'PythonDirective'
7 7​
8 8​
9class PythonDirective(BaseDirective): 9​class PythonDirective(BaseDirective):
10 """The python directive is designed to execute a piece of python code as part 10​ """The python directive is designed to execute a piece of python code as
11 of task execution. Given a callable name 'dothis_thing', possible implementations 11​ part of task execution. Given a callable name 'dothis_thing', possible
12 in the loaded code could include: 12​ implementations in the loaded code could include:
13 13​
14 class TaskClass(object): 14​ class TaskClass(object):
15 def my_task(self): 15​ def my_task(self):
16 return "I'm a task class method!" 16​ return "I'm a task class method!"
17 17​
18 _instantiated_class = TaskClass() 18​ _instantiated_class = TaskClass()
19 19​
20 task_class_target = _instantiated_class.my_task 20​ task_class_target = _instantiated_class.my_task
Show All 11 Lines30 In this case, you could pass in 'embedded_task_target' as the callable arg
32 def task_method(): 32​ def task_method():
33 return "I'm a task method!" 33​ return "I'm a task method!"
34 34​
35 And in this third case, you would pass in 'task_method' as the callable arg 35​ And in this third case, you would pass in 'task_method' as the callable arg
36 36​
37 format: 37​ format:
38 python: file=<python filename> callable=<method name> [<kwargname>=<kwargvalue>' 38​ python: file=<python filename> callable=<method name> [<kwargname>=<kwargvalue>'
39 required params: <python filename>, callable 39​ required params: <python filename>, callable
40 optional params: farther key/value pairs, to be passed into the callable as kwargs 40​ optional params: farther key/value pairs, to be passed into the callable
41​ as kwargs
41 """ 42​ """
42 43​
43 def _do_getattr(self, module, name): 44​ def _do_getattr(self, module, name):
44 """Isolation of getattr to make the code more easily testable 45​ """Isolation of getattr to make the code more easily testable
45 46​
46 @param module module from which to getattr from 47​ @param module module from which to getattr from
47 @param name name of object to getattr 48​ @param name name of object to getattr
48 @returns object retrieved from module 49​ @returns object retrieved from module
Show All 36 Lines80 def checkfile(self, filename):
85 86​
86 if not os.path.exists(filename): 87​ if not os.path.exists(filename):
87 raise Exception("The file %s does not exist!" % filename) 88​ raise Exception("The file %s does not exist!" % filename)
88 89​
89 def process(self, input_data, env_data): 90​ def process(self, input_data, env_data):
90 """For the python directive, we're expecting a yaml declaration of the form 91​ """For the python directive, we're expecting a yaml declaration of the form
91 "python: file=<python file> callable=<method name> [kwarg1=something ...]" 92​ "python: file=<python file> callable=<method name> [kwarg1=something ...]"
92 93​
93 @param input_data dictionary of all kwargs from yaml declaration, including callable and file 94​ @param input_data dictionary of all kwargs from yaml declaration,
95​ including callable and file
94 @param env_data information on the environment in which we're running 96​ @param env_data information on the environment in which we're running
95 @return output from the method specified, this is expected to be TAP output 97​ @return output from the method specified, this is expected to be TAP
98​ output
96 """ 99​ """
97 100​
98 if 'callable' not in input_data or 'file' not in input_data: 101​ if 'callable' not in input_data or 'file' not in input_data:
99 detected_args = ', '.join(input_data.keys()) 102​ detected_args = ', '.join(input_data.keys())
100 raise Exception("The python directive requires both 'callable' and \ 103​ raise Exception("The python directive requires both 'callable' and"
101'file' arguments. Detected arguments: %s" % detected_args) 104​ " 'file' arguments. Detected arguments: %s" %
105​ detected_args)
102 106​
103 method_name = input_data.pop('callable') 107​ method_name = input_data.pop('callable')
104 basedir = os.path.dirname(env_data['taskfile']) 108​ basedir = os.path.dirname(env_data['taskfile'])
105 109​
106 pyfile = os.path.join(basedir, input_data.pop('file')) 110​ pyfile = os.path.join(basedir, input_data.pop('file'))
107 111​
108 self.checkfile(pyfile) 112​ self.checkfile(pyfile)
109 113​
110 task_name, task = self.load_pyfile(pyfile) 114​ task_name, task = self.load_pyfile(pyfile)
111 115​
112 output = self.execute(task, method_name, input_data) 116​ output = self.execute(task, method_name, input_data)
113 117​
114 return output 118​ return output
View Options

libtaskotron/file_utils.py

1import os 1​import os
2import sys 2​import sys
3import urlgrabber.grabber, urlgrabber.progress 3​import urlgrabber.grabber
4​import urlgrabber.progress
4from libtaskotron.logger import log 5​from libtaskotron.logger import log
5 6​
6 7​
7def makedirs(fullpath): 8​def makedirs(fullpath):
8 '''This is the same as @method os.makedirs(), but does not throw an 9​ '''This is the same as @method os.makedirs(), but does not throw an
9 exception when the destination directory already exists.''' 10​ exception when the destination directory already exists.'''
10 try: 11​ try:
11 os.makedirs(fullpath) 12​ os.makedirs(fullpath)
▲ Show 20 Lines • Show All 83 Lines • Show Last 20 Lines
View Options

libtaskotron/koji_utils.py

1import koji 1​import koji
2 2​
3from libtaskotron import file_utils 3​from libtaskotron import file_utils
4from libtaskotron.logger import log 4​from libtaskotron.logger import log
5 5​
6KOJI_URL = 'http://koji.fedoraproject.org/kojihub' 6​KOJI_URL = 'http://koji.fedoraproject.org/kojihub'
7PKG_URL = 'http://kojipkgs.fedoraproject.org/packages' 7​PKG_URL = 'http://kojipkgs.fedoraproject.org/packages'
8 8​
9class KojiClient(object): 9​class KojiClient(object):
10 def __init__(self, koji_session=None): 10​ def __init__(self, koji_session=None):
11 self.session = koji_session 11​ self.session = koji_session
12 12​
13 def get_koji_session(self): 13​ def get_koji_session(self):
14 if self.session == None: 14​ if self.session is None:
15 self.session = koji.ClientSession(KOJI_URL) 15​ self.session = koji.ClientSession(KOJI_URL)
16 16​
17 return self.session 17​ return self.session
18 18​
19 def nvr_to_urls(self, nvr, arches=None, debuginfo=False, src=True): 19​ def nvr_to_urls(self, nvr, arches=None, debuginfo=False, src=True):
20 '''Get list of URLs for RPMs corresponding to a build. 20​ '''Get list of URLs for RPMs corresponding to a build.
21 @param nvr build NVR 21​ @param nvr build NVR
22 @param arches restrict the arches of builds to provide URLs for. If 22​ @param arches restrict the arches of builds to provide URLs for. If
23 basearch i386 is in the list, i686 arch is automatically 23​ basearch i386 is in the list, i686 arch is automatically
24 added (since that's the arch Koji API uses). 24​ added (since that's the arch Koji API uses).
25 ''' 25​ '''
26 26​
27 log.info('Querying Koji for a list of RPMS for: %s', nvr) 27​ log.info('Querying Koji for a list of RPMS for: %s', nvr)
28 28​
29 ksession = self.get_koji_session() 29​ ksession = self.get_koji_session()
30 30​
31 # add i686 arch if i386 is present in arches 31​ # add i686 arch if i386 is present in arches
32 if arches and 'i386' in arches and 'i686' not in arches: 32​ if arches and 'i386' in arches and 'i686' not in arches:
33 arches.append('i686') 33​ arches.append('i686')
34 34​
35 info = ksession.getBuild(nvr) 35​ info = ksession.getBuild(nvr)
36 if info == None: 36​ if info is None:
37 raise Exception("No build information found for %s" % nvr) 37​ raise Exception("No build information found for %s" % nvr)
38 38​
39 baseurl = '/'.join((PKG_URL, info['package_name'], 39​ baseurl = '/'.join((PKG_URL, info['package_name'],
40 info['version'], info['release'])) 40​ info['version'], info['release']))
41 41​
42 rpms = ksession.listRPMs(buildID=info['id'], arches=arches) 42​ rpms = ksession.listRPMs(buildID=info['id'], arches=arches)
43 if not debuginfo: 43​ if not debuginfo:
44 rpms = [r for r in rpms if not r['name'].endswith('-debuginfo')] 44​ rpms = [r for r in rpms if not r['name'].endswith('-debuginfo')]
▲ Show 20 Lines • Show All 49 Lines • Show Last 20 Lines
View Options

libtaskotron/logger.py

Show All 28 Lines
29log.addHandler(logging.NullHandler()) 29​log.addHandler(logging.NullHandler())
30 30​
31 31​
32def _log_excepthook(*exc_info): 32​def _log_excepthook(*exc_info):
33 '''Called when an exception is not caught''' 33​ '''Called when an exception is not caught'''
34 log.critical(''.join(traceback.format_exception(*exc_info))) 34​ log.critical(''.join(traceback.format_exception(*exc_info)))
35 35​
36 36​
37def init(name='libtaskotron', level=logging.INFO, stream=True, syslog=False, filelog=None): 37​def init(name='libtaskotron', level=logging.INFO, stream=True, syslog=False,
38​ filelog=None):
38 """Setup a logger 39​ """Setup a logger
39 @param name name of the logger, 'libtaskotron' is default 40​ @param name name of the logger, 'libtaskotron' is default
40 @param level level of logging 41​ @param level level of logging
41 @param stream enables logging to stderr 42​ @param stream enables logging to stderr
42 @param syslog enables logging to syslog 43​ @param syslog enables logging to syslog
43 @param filelog enables logging to a file, the value is the file name 44​ @param filelog enables logging to a file, the value is the file name
44 """ 45​ """
45 46​
46 # overwrite default except hook 47​ # overwrite default except hook
47 sys.excepthook = _log_excepthook 48​ sys.excepthook = _log_excepthook
48 49​
49 logger = logging.getLogger(name) 50​ logger = logging.getLogger(name)
50 logger.setLevel(level) 51​ logger.setLevel(level)
51 logger.addHandler(logging.NullHandler()) 52​ logger.addHandler(logging.NullHandler())
52 53​
53 fmt = '[%(name)s:%(filename)s:%(lineno)d] %(asctime)s %(levelname)-7s %(message)s' 54​ fmt = '[%(name)s:%(filename)s:%(lineno)d] '\
55​ '%(asctime)s %(levelname)-7s %(message)s'
54 datefmt = '%Y-%m-%d %H:%M:%S' 56​ datefmt = '%Y-%m-%d %H:%M:%S'
55 formatter = logging.Formatter(fmt=fmt, datefmt=datefmt) 57​ formatter = logging.Formatter(fmt=fmt, datefmt=datefmt)
56 58​
57 if stream: 59​ if stream:
58 stream_handler = logging.StreamHandler() 60​ stream_handler = logging.StreamHandler()
59 stream_handler.setFormatter(formatter) 61​ stream_handler.setFormatter(formatter)
60 logger.addHandler(stream_handler) 62​ logger.addHandler(stream_handler)
61 logger.debug("doing stream logging") 63​ logger.debug("doing stream logging")
62 64​
63 if syslog: 65​ if syslog:
64 syslog_handler = logging.handlers.SysLogHandler( 66​ syslog_handler = logging.handlers.SysLogHandler(address='/dev/log',
65 address='/dev/log', facility=logging.handlers.SysLogHandler.LOG_LOCAL4) 67​ facility=logging.handlers.SysLogHandler.LOG_LOCAL4)
66 syslog_handler.setFormatter(formatter) 68​ syslog_handler.setFormatter(formatter)
67 logger.addHandler(syslog_handler) 69​ logger.addHandler(syslog_handler)
68 logger.debug("doing syslog logging") 70​ logger.debug("doing syslog logging")
69 71​
70 if filelog: 72​ if filelog:
71 file_handler = logging.handlers.RotatingFileHandler(filelog, 73​ file_handler = logging.handlers.RotatingFileHandler(filelog,
72 maxBytes=500000, backupCount=5) 74​ maxBytes=500000, backupCount=5)
73 file_handler.setFormatter(formatter) 75​ file_handler.setFormatter(formatter)
74 logger.addHandler(file_handler) 76​ logger.addHandler(file_handler)
75 logger.debug('doing file logging to %s' % filelog) 77​ logger.debug('doing file logging to %s' % filelog)
76 78​
77 return logger 79​ return logger
View Options

libtaskotron/runner.py

1import logging 1​import logging
2import tempfile 2​import tempfile
3import os.path 3​import os.path
4import argparse 4​import argparse
5import imp 5​import imp
6from jinja2 import Template, Environment 6​from jinja2 import Environment
7import copy 7​import copy
8 8​
9from libtaskotron import taskyaml 9​from libtaskotron import taskyaml
10from libtaskotron import logger 10​from libtaskotron import logger
11from libtaskotron.logger import log 11​from libtaskotron.logger import log
12from libtaskotron.exceptions import TaskotronYamlError 12​from libtaskotron.exceptions import TaskotronYamlError
13 13​
14class Runner: 14​class Runner:
▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Line(s)80 def do_single_action(self, action):
92 output = directive_callable.process(rendered_action[directive_name], 92​ output = directive_callable.process(rendered_action[directive_name],
93 self.envdata) 93​ self.envdata)
94 94​
95 if 'export' in action: 95​ if 'export' in action:
96 self.working_data[action['export']] = output 96​ self.working_data[action['export']] = output
97 97​
98 def do_actions(self): 98​ def do_actions(self):
99 if 'task' not in self.taskdata or not self.taskdata['task']: 99​ if 'task' not in self.taskdata or not self.taskdata['task']:
100 raise TaskotronYamlError("At least one task should be specified in input yaml file") 100​ raise TaskotronYamlError("At least one task should be specified"
101​ " in input yaml file")
101 102​
102 for action in self.taskdata['task']: 103​ for action in self.taskdata['task']:
103 self.do_single_action(action) 104​ self.do_single_action(action)
104 105​
105 def _validate_input(self): 106​ def _validate_input(self):
106 if 'input' not in self.taskdata: 107​ if 'input' not in self.taskdata:
107 return 108​ return
108 109​
109 if not isinstance(self.taskdata['input'], dict): 110​ if not isinstance(self.taskdata['input'], dict):
110 raise TaskotronYamlError("Input yaml should contain correct 'input' section") 111​ raise TaskotronYamlError("Input yaml should contain correct"
112​ " 'input' section")
111 113​
112 if ('args' not in self.taskdata['input']) or not isinstance(self.taskdata['input']['args'], str): 114​ if ('args' not in self.taskdata['input']) or not isinstance(
113 raise TaskotronYamlError("Input yaml should contain correct 'args' section") 115​ self.taskdata['input']['args'], str):
116​ raise TaskotronYamlError("Input yaml should contain correct"
117​ " 'args' section")
114 118​
115 required_input = self.taskdata['input']['args'].split(',') 119​ required_input = self.taskdata['input']['args'].split(',')
116 120​
117 for arg in required_input: 121​ for arg in required_input:
118 if not arg in self.envdata: 122​ if not arg in self.envdata:
119 raise TaskotronYamlError('Required input arg %s was not defined' % arg) 123​ raise TaskotronYamlError('Required input arg %s '
124​ 'was not defined' % arg)
120 125​
121 def _validate_env(self): 126​ def _validate_env(self):
122 raise NotImplementedError("Environment validation is not yet implemented") 127​ raise NotImplementedError("Environment validation is not"
128​ " yet implemented")
123 129​
124 130​
125def get_argparser(): 131​def get_argparser():
126 parser = argparse.ArgumentParser() 132​ parser = argparse.ArgumentParser()
127 parser.add_argument("task", nargs=1, help="task to run") 133​ parser.add_argument("task", nargs=1, help="task to run")
128 parser.add_argument("-a", "--arch", choices=["i386", "x86_64", "armhfp", "noarch"], help=""" 134​ parser.add_argument("-a", "--arch",
129 Architecture to be used; support arches are x86, x86_64, and armhfp. 135​ choices=["i386", "x86_64", "armhfp", "noarch"], help=""
130 If omitted, defaults to noarch. 136​ "Architecture to be used; support arches are x86,\n"
131 """) 137​ "x86_64, and armhfp.\nIf omitted, defaults to noarch.")
132 parser.add_argument("-e", "--envr", help="envr to be used as input") 138​ parser.add_argument("-e", "--envr", help="envr to be used as input")
133 parser.add_argument("-t", "--tag", help="koji tag to be used as input") 139​ parser.add_argument("-t", "--tag", help="koji tag to be used as input")
134 return parser 140​ return parser
135 141​
136def main(): 142​def main():
137 parser = get_argparser() 143​ parser = get_argparser()
138 args = parser.parse_args() 144​ args = parser.parse_args()
139 145​
Show All 15 Lines
View Options

testing/conftest.py

Show All 13 Lines
14# GNU General Public License for more details. 14​# GNU General Public License for more details.
15# 15​#
16# You should have received a copy of the GNU General Public License along 16​# You should have received a copy of the GNU General Public License along
17# with this program; if not, write to the Free Software Foundation, Inc., 17​# with this program; if not, write to the Free Software Foundation, Inc.,
18# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. 18​# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19# 19​#
20# Author: Tim Flink <tflink@redhat.com> 20​# Author: Tim Flink <tflink@redhat.com>
21 21​
22import pytest ​
23import re ​
24 ​
25 22​
26def pytest_addoption(parser): 23​def pytest_addoption(parser):
27 """ 24​ """
28 Add an option to the py.test parser to detect when the functional tests 25​ Add an option to the py.test parser to detect when the functional tests
29 should be detected and run 26​ should be detected and run
30 """ 27​ """
31 28​
32 parser.addoption('--functional', action='store_true', default=False, 29​ parser.addoption('--functional', action='store_true', default=False,
Show All 21 Lines
View Options

testing/functest_python_directive.py

1import os 1​import os
2import copy 2​import copy
3from libtaskotron.directives import python_directive 3​from libtaskotron.directives import python_directive
4 4​
5 5​
6class TestTaskMethod(object): 6​class TestTaskMethod(object):
7 def setup_method(self, method): 7​ def setup_method(self, method):
8 self.ref_taskyaml = 'footask.yml' 8​ self.ref_taskyaml = 'footask.yml'
9 self.ref_pyfile = 'some_loadable_test.py' 9​ self.ref_pyfile = 'some_loadable_test.py'
10 self.ref_env = {'taskfile': os.path.join(os.path.dirname(os.path.abspath(__file__)), self.ref_taskyaml)} 10​ self.ref_env = {'taskfile': os.path.join(os.path.dirname(
11​ os.path.abspath(__file__)), self.ref_taskyaml)}
11 self.test_directive = python_directive.PythonDirective() 12​ self.test_directive = python_directive.PythonDirective()
12 13​
13 def testfunc_task_method(self): 14​ def testfunc_task_method(self):
14 ref_input = {'file': self.ref_pyfile, 'callable': 'task_method'} 15​ ref_input = {'file': self.ref_pyfile, 'callable': 'task_method'}
15 16​
16 test_output = self.test_directive.process(copy.copy(ref_input), self.ref_env) 17​ test_output = self.test_directive.process(copy.copy(ref_input),
18​ self.ref_env)
17 19​
18 assert test_output == "I'm a task method!" 20​ assert test_output == "I'm a task method!"
19 21​
20 def testfunc_class_embedded_call(self): 22​ def testfunc_class_embedded_call(self):
21 ref_input = {'file': self.ref_pyfile, 'callable': 'embedded_task_target'} 23​ ref_input = {'file': self.ref_pyfile,
24​ 'callable': 'embedded_task_target'}
22 25​
23 test_output = self.test_directive.process(copy.copy(ref_input), self.ref_env) 26​ test_output = self.test_directive.process(copy.copy(ref_input),
27​ self.ref_env)
24 28​
25 assert test_output == "I'm a __call__ method in a class!" 29​ assert test_output == "I'm a __call__ method in a class!"
26 30​
27 def testfunc_class_target(self): 31​ def testfunc_class_target(self):
28 ref_input = {'file': self.ref_pyfile, 'callable': 'task_class_target'} 32​ ref_input = {'file': self.ref_pyfile, 'callable': 'task_class_target'}
29 33​
30 test_output = self.test_directive.process(copy.copy(ref_input), self.ref_env) 34​ test_output = self.test_directive.process(copy.copy(ref_input),
35​ self.ref_env)
31 36​
32 assert test_output == "I'm a task class method!" 37​ assert test_output == "I'm a task class method!"
View Options

testing/test_koji_directive.py

1import pytest ​
2from dingus import Dingus 1​from dingus import Dingus
3 2​
4from libtaskotron.directives import koji_directive 3​from libtaskotron.directives import koji_directive
5 4​
6class TestKojiDirective(): 5​class TestKojiDirective():
7 def setup_method(self, method): 6​ def setup_method(self, method):
8 self.ref_nvr = 'foo-1.2-3.fc99' 7​ self.ref_nvr = 'foo-1.2-3.fc99'
9 self.ref_arch = 'noarch' 8​ self.ref_arch = 'noarch'
10 self.ref_tag = 'tagfoo' 9​ self.ref_tag = 'tagfoo'
11 self.ref_name = 'foo' 10​ self.ref_name = 'foo'
12 self.ref_version = '1.2' 11​ self.ref_version = '1.2'
13 self.ref_release = '3.fc99' 12​ self.ref_release = '3.fc99'
14 self.ref_buildid = 123456 13​ self.ref_buildid = 123456
15 14​
16 self.ref_build = {'package_name':self.ref_name, 'version':self.ref_version, 15​ self.ref_build = {'package_name': self.ref_name,
17 'release':self.ref_release, 'id': self.ref_buildid} 16​ 'version': self.ref_version,
17​ 'release': self.ref_release,
18​ 'id': self.ref_buildid}
18 19​
19 self.ref_rpms = [{'name':self.ref_name, 'version':self.ref_version, 'release':self.ref_release, 'nvr':self.ref_nvr, 'arch':self.ref_arch}, 20​ self.ref_rpms = [{'name': self.ref_name, 'version': self.ref_version,
20 {'name':self.ref_name, 'version':self.ref_version, 'release':self.ref_release, 'nvr':self.ref_nvr, 'arch':'src'}] 21​ 'release': self.ref_release, 'nvr': self.ref_nvr,
22​ 'arch': self.ref_arch},
23​ {'name': self.ref_name, 'version': self.ref_version,
24​ 'release': self.ref_release, 'nvr': self.ref_nvr,
25​ 'arch': 'src'}]
21 26​
22 self.ref_tagged = [{'nvr': self.ref_nvr}] 27​ self.ref_tagged = [{'nvr': self.ref_nvr}]
23 28​
24 self.ref_url = 'http://downloads.example.com/builds/%s.noarch.rpm' % self.ref_nvr 29​ self.ref_url = 'http://downloads.example.com/builds/%s.noarch.rpm'\
30​ % self.ref_nvr
25 self.helper = koji_directive.KojiDirective() 31​ self.helper = koji_directive.KojiDirective()
26 32​
27 def test_parse_download_command(self): 33​ def test_parse_download_command(self):
28 self.ref_rpms = ['/var/tmp/fake/%s.%s.rpm' % (self.ref_nvr, self.ref_arch)] 34​ self.ref_rpms = ['/var/tmp/fake/%s.%s.rpm'
35​ % (self.ref_nvr, self.ref_arch)]
29 36​
30 ref_input = {'action': 'download', 'arch': self.ref_arch, 'envr': self.ref_nvr} 37​ ref_input = {'action': 'download', 'arch': self.ref_arch,
31 ref_envdata = {'workdir':'/var/tmp/foo'} 38​ 'envr': self.ref_nvr}
39​ ref_envdata = {'workdir': '/var/tmp/foo'}
32 40​
33 stub_koji = Dingus(get_nvr_rpms__returns = self.ref_rpms) 41​ stub_koji = Dingus(get_nvr_rpms__returns = self.ref_rpms)
34 42​
35 test_helper = koji_directive.KojiDirective(stub_koji) 43​ test_helper = koji_directive.KojiDirective(stub_koji)
36 44​
37 test_helper.process(ref_input, ref_envdata) 45​ test_helper.process(ref_input, ref_envdata)
38 46​
39 getrpm_calls = stub_koji.calls() 47​ getrpm_calls = stub_koji.calls()
40 requested_nvr = getrpm_calls[0][1][0] 48​ requested_nvr = getrpm_calls[0][1][0]
41 49​
42 assert len(getrpm_calls) == 1 50​ assert len(getrpm_calls) == 1
43 assert requested_nvr == self.ref_nvr 51​ assert requested_nvr == self.ref_nvr
44 52​
45 def test_parse_download_tag_command(self): 53​ def test_parse_download_tag_command(self):
46 self.ref_rpms = ['/var/tmp/fake/%s.%s.rpm' % (self.ref_nvr, self.ref_arch)] 54​ self.ref_rpms = ['/var/tmp/fake/%s.%s.rpm'
55​ % (self.ref_nvr, self.ref_arch)]
47 56​
48 ref_input = {'action': 'download_tag', 'arch': self.ref_arch, 'tag': self.ref_tag} 57​ ref_input = {'action': 'download_tag', 'arch': self.ref_arch,
49 ref_envdata = {'workdir':'/var/tmp/foo'} 58​ 'tag': self.ref_tag}
59​ ref_envdata = {'workdir': '/var/tmp/foo'}
50 60​
51 stub_koji = Dingus(get_tagged_rpms__returns=self.ref_rpms) 61​ stub_koji = Dingus(get_tagged_rpms__returns=self.ref_rpms)
52 62​
53 test_helper = koji_directive.KojiDirective(stub_koji) 63​ test_helper = koji_directive.KojiDirective(stub_koji)
54 64​
55 test_helper.process(ref_input, ref_envdata) 65​ test_helper.process(ref_input, ref_envdata)
56 66​
57 getrpm_calls = stub_koji.calls() 67​ getrpm_calls = stub_koji.calls()
58 requested_tag = getrpm_calls[0][1][0] 68​ requested_tag = getrpm_calls[0][1][0]
59 69​
60 assert len(getrpm_calls) == 1 70​ assert len(getrpm_calls) == 1
61 assert requested_tag == self.ref_tag 71​ assert requested_tag == self.ref_tag
View Options

testing/test_koji_utils.py

1import pytest 1​import pytest
2from dingus import Dingus 2​from dingus import Dingus
3 3​
4from libtaskotron import koji_utils 4​from libtaskotron import koji_utils
5 5​
6class TestKojiClient(): 6​class TestKojiClient():
7 def setup_method(self, method): 7​ def setup_method(self, method):
8 self.ref_nvr = 'foo-1.2-3.fc99' 8​ self.ref_nvr = 'foo-1.2-3.fc99'
9 self.ref_arch = 'noarch' 9​ self.ref_arch = 'noarch'
10 self.ref_name = 'foo' 10​ self.ref_name = 'foo'
11 self.ref_version = '1.2' 11​ self.ref_version = '1.2'
12 self.ref_release = '3.fc99' 12​ self.ref_release = '3.fc99'
13 self.ref_buildid = 123456 13​ self.ref_buildid = 123456
14 14​
15 self.ref_build = {'package_name':self.ref_name, 'version':self.ref_version, 15​ self.ref_build = {'package_name': self.ref_name,
16 'release':self.ref_release, 'id': self.ref_buildid} 16​ 'version': self.ref_version,
17 17​ 'release': self.ref_release,
18 self.ref_rpms = [{'name':self.ref_name, 'version':self.ref_version, 'release':self.ref_release, 'nvr':self.ref_nvr, 'arch':self.ref_arch}, 18​ 'id': self.ref_buildid}
19 {'name':self.ref_name, 'version':self.ref_version, 'release':self.ref_release, 'nvr':self.ref_nvr, 'arch':'src'}] 19​
20 20​ self.ref_rpms = [{'name': self.ref_name, 'version': self.ref_version,
21 self.ref_url = 'http://downloads.example.com/builds/%s.noarch.rpm' % self.ref_nvr 21​ 'release': self.ref_release, 'nvr': self.ref_nvr,
22​ 'arch': self.ref_arch},
23​ {'name': self.ref_name, 'version': self.ref_version,
24​ 'release': self.ref_release, 'nvr': self.ref_nvr,
25​ 'arch': 'src'}]
26​
27​ self.ref_url = 'http://downloads.example.com/builds/%s.noarch.rpm' \
28​ % self.ref_nvr
22 self.helper = koji_utils.KojiClient() 29​ self.helper = koji_utils.KojiClient()
23 30​
24 def test_get_noarch_rpmurls_from_nvr(self): 31​ def test_get_noarch_rpmurls_from_nvr(self):
25 stub_koji = Dingus(getBuild__returns = self.ref_build, listRPMs__returns = self.ref_rpms) 32​ stub_koji = Dingus(getBuild__returns = self.ref_build,
33​ listRPMs__returns = self.ref_rpms)
26 34​
27 test_koji = koji_utils.KojiClient(stub_koji) 35​ test_koji = koji_utils.KojiClient(stub_koji)
28 36​
29 test_urls = test_koji.nvr_to_urls(self.ref_nvr) 37​ test_urls = test_koji.nvr_to_urls(self.ref_nvr)
30 38​
31 koji_baseurl = koji_utils.PKG_URL 39​ koji_baseurl = koji_utils.PKG_URL
32 ref_urls = ['%s/%s/%s/%s/%s/%s.%s.rpm' % (koji_baseurl, self.ref_name, self.ref_version, self.ref_release, self.ref_arch, self.ref_nvr, self.ref_arch), 40​ ref_urls = ['%s/%s/%s/%s/%s/%s.%s.rpm' % (koji_baseurl, self.ref_name,
33 '%s/%s/%s/%s/src/%s.src.rpm' % (koji_baseurl, self.ref_name, self.ref_version, self.ref_release, self.ref_nvr), 41​ self.ref_version, self.ref_release, self.ref_arch,
42​ self.ref_nvr, self.ref_arch),
43​ '%s/%s/%s/%s/src/%s.src.rpm' % (koji_baseurl, self.ref_name,
44​ self.ref_version, self.ref_release, self.ref_nvr),
34 ] 45​ ]
35 46​
36 assert test_urls == ref_urls 47​ assert test_urls == ref_urls
37 48​
38 def should_i386_rpmurls_query_i686(self): 49​ def should_i386_rpmurls_query_i686(self):
39 self.ref_arch = 'i386' 50​ self.ref_arch = 'i386'
40 51​
41 stub_koji = Dingus(getBuild__returns = self.ref_build) 52​ stub_koji = Dingus(getBuild__returns = self.ref_build)
42 53​
43 test_koji = koji_utils.KojiClient(stub_koji) 54​ test_koji = koji_utils.KojiClient(stub_koji)
44 55​
45 # this isn't setup to actually return urls since we don't care about that for this test 56​ # this isn't setup to actually return urls since we don't care about
57​ # that for this test
46 # it should throw an exception in this case. 58​ # it should throw an exception in this case.
47 try: 59​ try:
48 test_koji.nvr_to_urls(self.ref_nvr, arches = [self.ref_arch]) 60​ test_koji.nvr_to_urls(self.ref_nvr, arches = [self.ref_arch])
49 except Exception: 61​ except Exception:
50 pass 62​ pass
51 63​
52 listrpm_calls = stub_koji.calls('listRPMs') 64​ listrpm_calls = stub_koji.calls('listRPMs')
53 requested_arches = listrpm_calls[0][2]['arches'] 65​ requested_arches = listrpm_calls[0][2]['arches']
Show All 12 Lines
View Options

testing/test_python_directive.py

Show First 20 Lines • Show All 51 Lines • ▼ Show 20 Line(s)48 def test_task_method_is_found(self, monkeypatch):
52 52​
53 stub_getattr = Dingus() 53​ stub_getattr = Dingus()
54 54​
55 test_directive = python_directive.PythonDirective() 55​ test_directive = python_directive.PythonDirective()
56 monkeypatch.setattr(test_directive, '_do_getattr', stub_getattr) 56​ monkeypatch.setattr(test_directive, '_do_getattr', stub_getattr)
57 57​
58 test_directive.execute(ref_modulename, ref_methodname, ref_kwargs) 58​ test_directive.execute(ref_modulename, ref_methodname, ref_kwargs)
59 59​
60 assert (stub_getattr.calls()[0].name, stub_getattr.calls()[0].args) == ('()', (ref_modulename, ref_methodname)) 60​ assert (stub_getattr.calls()[0].name, stub_getattr.calls()[0].args) == (
61​ '()', (ref_modulename, ref_methodname))
61 62​
62 def test_task_method_is_executed(self, monkeypatch): 63​ def test_task_method_is_executed(self, monkeypatch):
63 ref_modulename = 'foo' 64​ ref_modulename = 'foo'
64 ref_methodname = 'bar' 65​ ref_methodname = 'bar'
65 ref_kwargs = {'baz': 'why', 'blah': 2} 66​ ref_kwargs = {'baz': 'why', 'blah': 2}
66 67​
67 stub_getattr = Dingus() 68​ stub_getattr = Dingus()
68 69​
69 test_directive = python_directive.PythonDirective() 70​ test_directive = python_directive.PythonDirective()
70 monkeypatch.setattr(test_directive, '_do_getattr', stub_getattr) 71​ monkeypatch.setattr(test_directive, '_do_getattr', stub_getattr)
71 72​
72 test_directive.execute(ref_modulename, ref_methodname, ref_kwargs) 73​ test_directive.execute(ref_modulename, ref_methodname, ref_kwargs)
73 74​
74 assert (stub_getattr.calls()[1].name, stub_getattr.calls()[1].args, stub_getattr.calls()[1].kwargs) == ('()', (), (ref_kwargs)) 75​ assert (stub_getattr.calls()[1].name, stub_getattr.calls()[1].args,
76​ stub_getattr.calls()[1].kwargs) == ('()', (), ref_kwargs)
75 77​
76 78​
77class TestProcess(object): 79​class TestProcess(object):
78 80​
79 def setup_method(self, method): 81​ def setup_method(self, method):
80 82​
81 self.ref_taskyaml = '/path/to/foo/footask.yml' 83​ self.ref_taskyaml = '/path/to/foo/footask.yml'
82 self.ref_input = {'file': 'foo.py', 'callable': 'bar', 'kwarg1': 'baz'} 84​ self.ref_input = {'file': 'foo.py', 'callable': 'bar', 'kwarg1': 'baz'}
83 self.ref_env = {'taskfile': self.ref_taskyaml} 85​ self.ref_env = {'taskfile': self.ref_taskyaml}
84 self.ref_pyfile = os.path.join(os.path.dirname(self.ref_taskyaml), self.ref_input['file']) 86​ self.ref_pyfile = os.path.join(os.path.dirname(self.ref_taskyaml),
87​ self.ref_input['file'])
85 88​
86 self.stub_checkfile = Dingus('checkfile') 89​ self.stub_checkfile = Dingus('checkfile')
87 self.stub_imported = Dingus('imported') 90​ self.stub_imported = Dingus('imported')
88 self.stub_pyload = Dingus('pyload', dostuff__returns=('name', self.stub_imported)) 91​ self.stub_pyload = Dingus('pyload', dostuff__returns=('name',
92​ self.stub_imported))
89 self.stub_execute = Dingus('execute') 93​ self.stub_execute = Dingus('execute')
90 94​
91 def test_process_checks_file(self, monkeypatch): 95​ def test_process_checks_file(self, monkeypatch):
92 96​
93 97​
94 test_directive = python_directive.PythonDirective() 98​ test_directive = python_directive.PythonDirective()
95 monkeypatch.setattr(test_directive, 'checkfile', self.stub_checkfile) 99​ monkeypatch.setattr(test_directive, 'checkfile', self.stub_checkfile)
96 monkeypatch.setattr(test_directive, 'load_pyfile', self.stub_pyload.dostuff) 100​ monkeypatch.setattr(test_directive, 'load_pyfile',
101​ self.stub_pyload.dostuff)
97 monkeypatch.setattr(test_directive, 'execute', self.stub_execute) 102​ monkeypatch.setattr(test_directive, 'execute', self.stub_execute)
98 103​
99 test_output = test_directive.process(copy.copy(self.ref_input), self.ref_env) 104​ test_output = test_directive.process(copy.copy(self.ref_input),
105​ self.ref_env)
100 106​
101 checkfile_call = self.stub_checkfile.calls[0] 107​ checkfile_call = self.stub_checkfile.calls[0]
102 108​
103 109​ assert (checkfile_call.name, checkfile_call.args,
104 assert (checkfile_call.name, checkfile_call.args, checkfile_call.kwargs) == ('()', (self.ref_pyfile,), {}) 110​ checkfile_call.kwargs) == ('()', (self.ref_pyfile,), {})
105 111​
106 def test_process_loads_file(self, monkeypatch): 112​ def test_process_loads_file(self, monkeypatch):
107 113​
108 test_directive = python_directive.PythonDirective() 114​ test_directive = python_directive.PythonDirective()
109 monkeypatch.setattr(test_directive, 'checkfile', self.stub_checkfile) 115​ monkeypatch.setattr(test_directive, 'checkfile', self.stub_checkfile)
110 monkeypatch.setattr(test_directive, 'load_pyfile', self.stub_pyload.dostuff) 116​ monkeypatch.setattr(test_directive, 'load_pyfile',
117​ self.stub_pyload.dostuff)
111 monkeypatch.setattr(test_directive, 'execute', self.stub_execute) 118​ monkeypatch.setattr(test_directive, 'execute', self.stub_execute)
112 119​
113 test_output = test_directive.process(copy.copy(self.ref_input), self.ref_env) 120​ test_output = test_directive.process(copy.copy(self.ref_input),
121​ self.ref_env)
114 122​
115 pyload_call = self.stub_pyload.calls[0] 123​ pyload_call = self.stub_pyload.calls[0]
116 124​
117 assert (pyload_call.name, pyload_call.args, pyload_call.kwargs) == ('dostuff', (self.ref_pyfile,), {}) 125​ assert (pyload_call.name, pyload_call.args, pyload_call.kwargs) == (
126​ 'dostuff', (self.ref_pyfile,), {})
118 127​
119 def test_process_executes(self, monkeypatch): 128​ def test_process_executes(self, monkeypatch):
120 129​
121 ref_kwargs = {'kwarg1': 'baz'} 130​ ref_kwargs = {'kwarg1': 'baz'}
122 131​
123 test_directive = python_directive.PythonDirective() 132​ test_directive = python_directive.PythonDirective()
124 monkeypatch.setattr(test_directive, 'checkfile', self.stub_checkfile) 133​ monkeypatch.setattr(test_directive, 'checkfile', self.stub_checkfile)
125 monkeypatch.setattr(test_directive, 'load_pyfile', self.stub_pyload.dostuff) 134​ monkeypatch.setattr(test_directive, 'load_pyfile', self.stub_pyload.dostuff)
126 monkeypatch.setattr(test_directive, 'execute', self.stub_execute) 135​ monkeypatch.setattr(test_directive, 'execute', self.stub_execute)
127 136​
128 test_output = test_directive.process(copy.copy(self.ref_input), self.ref_env) 137​ test_output = test_directive.process(copy.copy(self.ref_input), self.ref_env)
129 138​
130 execute_call = self.stub_execute.calls[0] 139​ execute_call = self.stub_execute.calls[0]
131 140​
132 assert (execute_call.name, execute_call.args, execute_call.kwargs) == ('()', (self.stub_imported, self.ref_input['callable'], ref_kwargs), {}) 141​ assert (execute_call.name, execute_call.args, execute_call.kwargs) == (
133No newline at end of file 142​ '()', (self.stub_imported, self.ref_input['callable'], ref_kwargs), {})
143​No newline at end of file
View Options

testing/test_python_utils.py

1# -*- coding: utf-8 -*- 1​# -*- coding: utf-8 -*-
2# Copyright 2014, Red Hat, Inc. 2​# Copyright 2014, Red Hat, Inc.
3# License: GNU General Public License version 2 or later 3​# License: GNU General Public License version 2 or later
4 4​
5'''Unit tests for libtaskotron/python_utils.py''' 5​'''Unit tests for libtaskotron/python_utils.py'''
6 6​
7from libtaskotron import python_utils 7​from libtaskotron import python_utils
8 8​
9class TestPythonUtils: 9​class TestPythonUtils:
10 10​
11 def test_listlike_true(self): 11​ def test_listlike_true(self):
12 '''Test 'listlike' method when it returns True''' 12​ '''Test 'listlike' method when it returns True'''
13 assert python_utils.listlike([1, 2]) 13​ assert python_utils.listlike([1, 2])
14 assert python_utils.listlike(['a', 'b']) 14​ assert python_utils.listlike(['a', 'b'])
15 assert python_utils.listlike(('foo',)) 15​ assert python_utils.listlike(('foo',))
16 assert python_utils.listlike(set(('foo','bar','baz'))) 16​ assert python_utils.listlike({'foo', 'bar', 'baz'})
17 assert python_utils.listlike(set()) 17​ assert python_utils.listlike(set())
18 assert python_utils.listlike(()) 18​ assert python_utils.listlike(())
19 assert python_utils.listlike([u'foo', u'bar']) 19​ assert python_utils.listlike([u'foo', u'bar'])
20 assert python_utils.listlike({'a': 1, 'b': 2}) 20​ assert python_utils.listlike({'a': 1, 'b': 2})
21 21​
22 def test_listlike_false(self): 22​ def test_listlike_false(self):
23 '''Test 'listlike' method when it returns False''' 23​ '''Test 'listlike' method when it returns False'''
24 assert not python_utils.listlike('a') 24​ assert not python_utils.listlike('a')
25 assert not python_utils.listlike(u'a') 25​ assert not python_utils.listlike(u'a')
26 assert not python_utils.listlike(unicode('foo')) 26​ assert not python_utils.listlike(unicode('foo'))
27 assert not python_utils.listlike(3) 27​ assert not python_utils.listlike(3)
28 assert not python_utils.listlike(3.14) 28​ assert not python_utils.listlike(3.14)
29 assert not python_utils.listlike(None) 29​ assert not python_utils.listlike(None)
30 assert not python_utils.listlike(object()) 30​ assert not python_utils.listlike(object())
View Options

testing/test_runner.py

1import pytest 1​import pytest
2import os 2​import os
3import sys 3​import sys
4 4​
5from libtaskotron import runner 5​from libtaskotron import runner
6from libtaskotron.exceptions import TaskotronYamlError 6​from libtaskotron.exceptions import TaskotronYamlError
7 7​
8 8​
9class TestRunnerInputVerify(): 9​class TestRunnerInputVerify():
10 def test_yamlrunner_valid_input(self): 10​ def test_yamlrunner_valid_input(self):
11 ref_cmd = '-e foo-1.2-3.fc99 -a x86_64 footask.yml' 11​ ref_cmd = '-e foo-1.2-3.fc99 -a x86_64 footask.yml'
12 ref_argdata = {'input': {'args': 'envr,arch'}} 12​ ref_argdata = {'input': {'args': 'envr,arch'}}
13 ref_data = {'envr': 'foo-1.2-3.fc99', 'arch': 'x86_64', 'taskfile': 'footask.yml'} 13​ ref_data = {'envr': 'foo-1.2-3.fc99', 'arch': 'x86_64',
14​ 'taskfile': 'footask.yml'}
14 15​
15 test_parser = runner.get_argparser() 16​ test_parser = runner.get_argparser()
16 test_args = vars(test_parser.parse_args(ref_cmd.split())) 17​ test_args = vars(test_parser.parse_args(ref_cmd.split()))
17 18​
18 test_runner = runner.Runner(ref_argdata, test_args) 19​ test_runner = runner.Runner(ref_argdata, test_args)
19 20​
20 test_runner._validate_input() 21​ test_runner._validate_input()
21 22​
▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Line(s)95 def test_trivial_creation(self, tmpdir):
99 ref_inputdata = {} 100​ ref_inputdata = {}
100 101​
101 test_runner = runner.Runner(ref_taskdata, ref_inputdata, tmpdir) 102​ test_runner = runner.Runner(ref_taskdata, ref_inputdata, tmpdir)
102 103​
103 assert test_runner.taskdata == ref_taskdata 104​ assert test_runner.taskdata == ref_taskdata
104 105​
105class TestRunnerPrep(): 106​class TestRunnerPrep():
106 def test_prep_invalid_commands(self, tmpdir): 107​ def test_prep_invalid_commands(self, tmpdir):
107 ref_taskdata = {'preparation': {'turboencabulator':'reduce sinusoidal repleneration'}} 108​ ref_taskdata = {'preparation': {'turboencabulator':
109​ 'reduce sinusoidal repleneration'}}
108 ref_inputdata = {} 110​ ref_inputdata = {}
109 111​
110 test_runner = runner.Runner(ref_taskdata, ref_inputdata, tmpdir) 112​ test_runner = runner.Runner(ref_taskdata, ref_inputdata, tmpdir)
111 113​
112 with pytest.raises(Exception): 114​ with pytest.raises(Exception):
113 test_runner.do_prep() 115​ test_runner.do_prep()
114 116​
115class TestRunnerDirectiveLoader(): 117​class TestRunnerDirectiveLoader():
Show All 33 Lines150class TestRunnerSingleAction():
149 151​
150 def setup_method(self, method): 152​ def setup_method(self, method):
151 self.ref_taskdata = {} 153​ self.ref_taskdata = {}
152 self.ref_inputdata = {} 154​ self.ref_inputdata = {}
153 155​
154 def test_runner_extract_directive_name(self): 156​ def test_runner_extract_directive_name(self):
155 ref_directivename = 'dummy' 157​ ref_directivename = 'dummy'
156 ref_action = {'name': 'test action', 158​ ref_action = {'name': 'test action',
157 ref_directivename: {'result': 'pass', 'msg': 'dummy message'}} 159​ ref_directivename: {'result': 'pass',
160​ 'msg': 'dummy message'}}
158 161​
159 test_runner = runner.Runner(self.ref_taskdata, self.ref_inputdata) 162​ test_runner = runner.Runner(self.ref_taskdata, self.ref_inputdata)
160 163​
161 test_directivename = test_runner._extract_directive_from_action(ref_action) 164​ test_directivename = test_runner._extract_directive_from_action(
165​ ref_action)
162 166​
163 assert test_directivename == ref_directivename 167​ assert test_directivename == ref_directivename
164 168​
165 def test_runner_single_dummy_action_pass(self): 169​ def test_runner_single_dummy_action_pass(self):
166 ref_action = {'dummy': {'result': 'pass'}} 170​ ref_action = {'dummy': {'result': 'pass'}}
167 ref_action.update({'name': 'Dummy Action' }) 171​ ref_action.update({'name': 'Dummy Action' })
168 172​
169 test_runner = runner.Runner(self.ref_taskdata, self.ref_inputdata) 173​ test_runner = runner.Runner(self.ref_taskdata, self.ref_inputdata)
▲ Show 20 Lines • Show All 115 Lines • ▼ Show 20 Line(s)288class TestRunnerRenderAction():
285 289​
286 def setup_method(self, method): 290​ def setup_method(self, method):
287 self.ref_taskdata = {} 291​ self.ref_taskdata = {}
288 self.ref_inputdata = {} 292​ self.ref_inputdata = {}
289 293​
290 def test_runner_render_single_variable(self): 294​ def test_runner_render_single_variable(self):
291 ref_message = 'This is a variable message' 295​ ref_message = 'This is a variable message'
292 ref_messagename = 'variable_messsage' 296​ ref_messagename = 'variable_messsage'
293 ref_action = {'name': 'test dummy action', 'dummy': {'result': 'pass', 'msg': '{{ %s }}' % ref_messagename}} 297​ ref_action = {'name': 'test dummy action', 'dummy':
298​ {'result': 'pass', 'msg': '{{ %s }}' % ref_messagename}}
294 299​
295 test_runner = runner.Runner(self.ref_taskdata, self.ref_inputdata) 300​ test_runner = runner.Runner(self.ref_taskdata, self.ref_inputdata)
296 test_runner.working_data[ref_messagename] = ref_message 301​ test_runner.working_data[ref_messagename] = ref_message
297 302​
298 test_rendered_action = test_runner._render_action(ref_action) 303​ test_rendered_action = test_runner._render_action(ref_action)
299 304​
300 assert test_rendered_action['dummy']['msg'] == ref_message 305​ assert test_rendered_action['dummy']['msg'] == ref_message
301 306​
302 307​