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_api(self):
httpretty.register_uri(
httpretty.POST,
'http://api.barium.cc/jobs',
body='{"message":"Resource creation started",'
'"url":"http://barium.cc/jobs/5722"}'
)
response = api.post(
url=Arguments.url,
repo_token=Arguments.repo_token,
service_job_id=Arguments.service_job_id,
service_name=Arguments.service_name,
git={},
cc_data={}
)
self.assertEqual(response.json(),
{'url': 'http://barium.cc/jobs/5722',
'message': 'Resource creation started'})
def testCheck(self):
self.assertEqual(core.check(self.a, self.b), self.r)
def test_run(self):
code = core.find_infractions(self.args, self.logger, self.r)
self.assertEqual(code != 0, self.exit_code)
def testAv(self):
self.assertEqual(core.av(self.m, self.n), self.r)
def _exit_code(self):
try:
main(Arguments)
except SystemExit as e:
return e.code
from xenon.api import post
from xenon.core import analyze
from xenon.repository import gitrepo
args = args or parse_args()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('xenon')
if args.url and len(args.path) > 1:
logger.error(
'-u, --url cannot be used when multiple paths are specified',
)
sys.exit(1)
errors, cc_data = analyze(args, logger)
exit_code = 0
if args.url:
response = post(
url=args.url,
repo_token=args.repo_token,
service_job_id=args.service_job_id,
service_name=args.service_name,
git=gitrepo(args.path),
cc_data=cc_data
)
logger.info('HTTP: %s', response.status_code)
logger.info('HTTP: %s', response.text)
if 'error' in response.json():
exit_code = 3
if errors:
exit_code = 1
sys.exit(exit_code)
def start_job(self, job_id):
"""Get a job from the job store and start it on the compute resource.
Args:
job_id (str): The id of the job to start.
"""
self._logger.debug('Starting job ' + job_id)
with self._job_store:
job = self._job_store.get_job(job_id)
# submit job
xenon_jobdesc = xenon.jobs.JobDescription()
xenon_jobdesc.setWorkingDirectory(job.remote_workdir_path)
xenon_jobdesc.setExecutable(self._remote_cwlrunner)
args = [
job.remote_workflow_path,
job.remote_input_path
]
xenon_jobdesc.setArguments(args)
xenon_jobdesc.setStdout(job.remote_stdout_path)
xenon_jobdesc.setStderr(job.remote_stderr_path)
xenon_jobdesc.setMaxTime(60)
xenon_jobdesc.setProcessesPerNode(4)
xenon_jobdesc.setStartSingleProcess(True)
print("Starting job: " + str(xenon_jobdesc))
xenon_job = self._x.jobs().submitJob(self._sched, xenon_jobdesc)
job.remote_job_id = xenon_job.getIdentifier()
self._logger.debug('Job submitted')
def __init__(self, bot, session, data):
self.session = session
self.data = data
self.bot = bot
self.id_translator = {}
self.options = types.BooleanArgs([])
self.semaphore = asyncio.Semaphore(2)
import os
from setuptools import setup
import xenon
with open(os.path.join(os.path.dirname(__file__), 'README.rst')) as fobj:
readme = fobj.read()
with open(os.path.join(os.path.dirname(__file__), 'requirements.txt')) as fobj:
reqs = fobj.read().splitlines()
setup(name='xenon',
version=xenon.__version__,
author='Michele Lacchia',
author_email='michelelacchia@gmail.com',
url='https://xenon.readthedocs.org/',
download_url='https://pypi.python.org/xenon/',
license='MIT',
description='Monitor code metrics for Python on your CI server',
platforms='any',
long_description=readme,
packages=['xenon'],
tests_require=['tox', 'httpretty'],
install_requires=reqs,
entry_points={'console_scripts': ['xenon = xenon:main']},
classifiers=[
'Development Status :: 4 - Beta',
'Environment :: Console',
'Intended Audience :: Developers',
def setup(bot):
bot.add_cog(Admin(bot))