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_get_task_queue_default():
"""
Given a steps dictionary that sets the task queue to `test_queue` return
`test_queue` as the queue name.
"""
steps = {"run": {"task_queue": "test_queue"}}
queue = Step.get_task_queue_from_dict(steps)
assert queue == "test_queue"
def test_get_task_queue_run_missing():
"""
Given an empty steps dictionary return `merlin` as the queue name.
"""
steps = {}
queue = Step.get_task_queue_from_dict(steps)
assert queue == "merlin"
def step(self, task_name):
"""Return a Step object for the given task name
:param `task_name`: The task name.
:return: A Merlin Step object.
"""
return Step(self.dag.values[task_name])
step_dict["run"]["cmd"] = re.sub(re.escape(str1), str2, cmd, flags=re.I)
restart_cmd = step_dict["run"]["restart"]
if restart_cmd:
step_dict["run"]["restart"] = re.sub(
re.escape(str1), str2, restart_cmd, flags=re.I
)
if new_workspace is None:
new_workspace = self.get_workspace()
LOG.debug(f"cloned step with workspace {new_workspace}")
study_step = StudyStep()
study_step.name = step_dict["name"]
study_step.description = step_dict["description"]
study_step.run = step_dict["run"]
return Step(_StepRecord(new_workspace, study_step))