Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_findplugins(self):
self.assertEqual(citellus.findplugins('@#~@#æßð'), [])
def callcitellus(path=False, plugins=False):
"""
Do actual execution of citellus against data
:param path: sosreport path
:param plugins: plugins enabled as provided to citellus
:return: dict with results
"""
# Call citellus and format data returned
results = citellus.docitellus(path=path, plugins=plugins)
# Process plugin output from multiple plugins
new_dict = {}
for item in results:
name = item['plugin']
new_dict[name] = item
return new_dict
if options.live is True:
for playbook in playbooksnap:
if citellus.regexpfile(file=playbook, regexp="CITELLUS_HYBRID"):
# Add to the list of playbooks to run
playbooks.append(playbook)
# Remove from the skipped playbooks
playbookskipped.remove(playbook)
commands = []
ansible = which("ansible-playbook")
for playbook in playbooks:
commands.append("%s -i localhost --connection=local, %s" % (ansible, playbook))
# Actually run the tests
results = citellus.docitellus(live=options.live, path=None, plugins=commands, lang='en_US')
# Do formatting of results to remove ansible-playbook -i localhost, and adjust return codes to citellus standards
for result in results:
# Convert RC codes to what citellus expects
if result['result']['rc'] == 2:
result['result']['rc'] = citellus.RC_FAILED
elif result['result']['rc'] == 0:
result['result']['rc'] = citellus.RC_OKAY
# Convert stdout to stderr for citellus handling
result['result']['err'] = result['result']['out']
result['result']['out'] = ''
# Remove ansible-playbook command and just leave yml file
result['plugin'] = result['plugin'].replace(which('ansible-playbook'), '').replace(' -i localhost --connection=local, ', '')
def list(options):
playbooksdir = os.path.join(citellus.citellusdir, PlaybooksFolder)
playbooks = citellus.findplugins(folders=[playbooksdir], include=options.include, exclude=options.exclude,
executables=False, extension=".yml")
return playbooks
def main():
"""
Main code stub
"""
options = parse_args()
# Configure logging
logging.basicConfig(level=options.loglevel)
if not options.quiet:
show_logo()
# Each argument in sosreport is a sosreport
# Prefill enabled citellus plugins from args
citellusplugins = citellus.findplugins(folders=options.pluginpath, include=options.include, exclude=options.exclude)
# Grab the data
grouped = domagui(sosreports=options.sosreports, citellusplugins=citellusplugins)
# For now, let's only print plugins that have rc ! 0 in quiet
if options.quiet:
toprint = maguiformat(grouped)
else:
toprint = grouped
pprint.pprint(toprint, width=1)
# We need to run our plugins against that data
def run(options): # do not edit this line
"""
Executes plugin
:param options: options passed to main binary
:return:
"""
if not which("ansible-playbook"):
print("# skipping ansible per missing ansible-playbook binary.")
return
playbooksdir = os.path.join(citellus.citellusdir, PlaybooksFolder)
playbooks = citellus.findplugins(folders=[playbooksdir], include=options.include, exclude=options.exclude,
executables=False, extension=".yml")
playbooklive = []
playbooksnap = []
for playbook in playbooks:
if citellus.regexpfile(file=playbook, regexp="CITELLUS_ROOT"):
playbooksnap.append(playbook)
else:
playbooklive.append(playbook)
# Restrict the playbooks to process to the running mode
if options.live is True:
playbooks = playbooklive
playbookskipped = playbooksnap
else:
playbooks = playbooksnap
"""
# Base path to find files
CITELLUS_ROOT = os.environ["CITELLUS_ROOT"]
if runninglive():
# Running on LIVE environment
# For example, next condition might be an existing file like:
# os.path.exists(os.join.path(CITELLUS_ROOT,'/etc/nova/nova.conf'))
if True:
# Example: File does exist, check file contents or other checks
if True:
# Plugin tests passed
exitcitellus(code=citellus.RC_OKAY)
else:
# Error with plugin tests
# Provide messages on STDERR
exitcitellus(
code=citellus.RC_FAILED, msg="There was an error because of 'xxx'"
)
else:
# Plugin script skipped per conditions
# Provide reason for skipping:
exitcitellus(code=citellus.RC_SKIPPED, msg="Required file 'xxx' not found")
elif not runninglive():
# Running on snapshot/sosreport environment
# Provide messages on STDERR
exitcitellus(
code=citellus.RC_FAILED, msg="There was an error because of 'xxx'"
)
else:
# Plugin script skipped per conditions
# Provide reason for skipping:
exitcitellus(code=citellus.RC_SKIPPED, msg="Required file 'xxx' not found")
elif not runninglive():
# Running on snapshot/sosreport environment
if True:
if True:
# Plugin tests passed
exitcitellus(code=citellus.RC_OKAY)
else:
# Error with plugin tests
# Provide messages on STDERR
exitcitellus(
code=citellus.RC_FAILED, msg="There was an error because of 'xxx'"
)
else:
# Plugin script skipped per conditions
# Provide reason for skipping:
exitcitellus(code=citellus.RC_SKIPPED, msg="Required file 'xxx' not found")
commands = []
ansible = which("ansible-playbook")
for playbook in playbooks:
commands.append("%s -i localhost --connection=local, %s" % (ansible, playbook))
# Actually run the tests
results = citellus.docitellus(live=options.live, path=None, plugins=commands, lang='en_US')
# Do formatting of results to remove ansible-playbook -i localhost, and adjust return codes to citellus standards
for result in results:
# Convert RC codes to what citellus expects
if result['result']['rc'] == 2:
result['result']['rc'] = citellus.RC_FAILED
elif result['result']['rc'] == 0:
result['result']['rc'] = citellus.RC_OKAY
# Convert stdout to stderr for citellus handling
result['result']['err'] = result['result']['out']
result['result']['out'] = ''
# Remove ansible-playbook command and just leave yml file
result['plugin'] = result['plugin'].replace(which('ansible-playbook'), '').replace(' -i localhost --connection=local, ', '')
# Now, fake 'skipped' for all the plugins which were tied to the mode we're not running in:
for playbook in playbookskipped:
dictionary = {'plugin': playbook,
'result': {'rc': citellus.RC_SKIPPED, 'err': 'Skipped for incompatible operating mode', 'out': ''}}
results.append(dictionary)
return results
# For example, next condition might be an existing file like:
# os.path.exists(os.join.path(CITELLUS_ROOT,'/etc/nova/nova.conf'))
if True:
# Example: File does exist, check file contents or other checks
if True:
# Plugin tests passed
exitcitellus(code=citellus.RC_OKAY)
else:
# Error with plugin tests
# Provide messages on STDERR
exitcitellus(
code=citellus.RC_FAILED, msg="There was an error because of 'xxx'"
)
else:
# Plugin script skipped per conditions
# Provide reason for skipping:
exitcitellus(code=citellus.RC_SKIPPED, msg="Required file 'xxx' not found")
elif not runninglive():
# Running on snapshot/sosreport environment
if True:
if True:
# Plugin tests passed
exitcitellus(code=citellus.RC_OKAY)
else:
# Error with plugin tests