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_quotes(self):
self.assertEqual(
{'result': 'complex kwarg'},
parse_search_terms('result:"complex kwarg"')
)
self.assertEqual(
{'kwargs': {'some_kwarg1': 'some value1', 'some_kwarg2': 'some value2'}},
parse_search_terms('kwargs:some_kwarg1="some value1" kwargs:some_kwarg2="some value2"')
)
def test_args(self):
self.assertEqual(
{'args': ['some_value']},
parse_search_terms('args:some_value')
)
self.assertEqual(
{'args': ['some_value1', 'some_value2']},
parse_search_terms('args:some_value1 args:some_value2')
)
def test_args(self):
self.assertEqual(
{'args': ['some_value']},
parse_search_terms('args:some_value')
)
self.assertEqual(
{'args': ['some_value1', 'some_value2']},
parse_search_terms('args:some_value1 args:some_value2')
)
def test_kwargs(self):
self.assertEqual(
{'kwargs': {'some_kwarg': 'some_value'}},
parse_search_terms('kwargs:some_kwarg=some_value')
)
self.assertEqual(
{'kwargs': {'some_kwarg1': 'some_value1', 'some_kwarg2': 'some_value2'}},
parse_search_terms('kwargs:some_kwarg1=some_value1 kwargs:some_kwarg2=some_value2')
)
def test_strip_spaces(self):
self.assertEqual(
{'any': 'someval'},
parse_search_terms(' someval ')
)
self.assertEqual(
{'kwargs': {'some_kwarg': 'some_value'}},
parse_search_terms(' kwargs:some_kwarg=some_value ')
)
def test_kwargs(self):
self.assertEqual(
{'kwargs': {'some_kwarg': 'some_value'}},
parse_search_terms('kwargs:some_kwarg=some_value')
)
self.assertEqual(
{'kwargs': {'some_kwarg1': 'some_value1', 'some_kwarg2': 'some_value2'}},
parse_search_terms('kwargs:some_kwarg1=some_value1 kwargs:some_kwarg2=some_value2')
)
def test_result_value(self):
self.assertEqual(
{'result': 'resval'},
parse_search_terms('result:resval')
)
def test_strip_spaces(self):
self.assertEqual(
{'any': 'someval'},
parse_search_terms(' someval ')
)
self.assertEqual(
{'kwargs': {'some_kwarg': 'some_value'}},
parse_search_terms(' kwargs:some_kwarg=some_value ')
)
def test_quotes(self):
self.assertEqual(
{'result': 'complex kwarg'},
parse_search_terms('result:"complex kwarg"')
)
self.assertEqual(
{'kwargs': {'some_kwarg1': 'some value1', 'some_kwarg2': 'some value2'}},
parse_search_terms('kwargs:some_kwarg1="some value1" kwargs:some_kwarg2="some value2"')
)
def iter_tasks(events, limit=None, type=None, worker=None, state=None,
sort_by=None, received_start=None, received_end=None,
started_start=None, started_end=None, search=None):
i = 0
tasks = events.state.tasks_by_timestamp()
if sort_by is not None:
tasks = sort_tasks(tasks, sort_by)
def convert(x):
return time.mktime(datetime.datetime.strptime(x, '%Y-%m-%d %H:%M').timetuple())
search_terms = parse_search_terms(search or {})
for uuid, task in tasks:
if type and task.name != type:
continue
if worker and task.worker and task.worker.hostname != worker:
continue
if state and task.state != state:
continue
if received_start and task.received and\
task.received < convert(received_start):
continue
if received_end and task.received and\
task.received > convert(received_end):
continue
if started_start and task.started and\
task.started < convert(started_start):