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_set_unsaved_task_as_dependency(self):
# Adds only one dependency to task with no dependencies
t = Task(self.tw, description='test task')
dependency = Task(self.tw, description='needs to be done first')
# We only save the parent task, dependency task is unsaved
t.save()
t['depends'] = set([dependency])
self.assertRaises(Task.NotSaved, t.save)
def start(self):
if not self.saved:
raise Task.NotSaved(
'Task needs to be saved before it can be started',
)
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if self.completed:
raise Task.CompletedTask('Cannot start a completed task')
elif self.deleted:
raise Task.DeletedTask('Deleted task cannot be started')
elif self.active:
raise Task.ActiveTask('Task is already active')
self.backend.start_task(self)
# Refresh the status again, so that we have updated info stored
def delete(self):
if not self.saved:
raise Task.NotSaved(
'Task needs to be saved before it can be deleted',
)
# Refresh the status, and raise exception if the task is deleted
self.refresh(only_fields=['status'])
if self.deleted:
raise Task.DeletedTask('Task was already deleted')
self.backend.delete_task(self)
# Refresh the status again, so that we have updated info stored
self.refresh(only_fields=['status', 'start', 'end'])
def stop(self):
if not self.saved:
raise Task.NotSaved(
'Task needs to be saved before it can be stopped',
)
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if not self.active:
raise Task.InactiveTask('Cannot stop an inactive task')
self.backend.stop_task(self)
# Refresh the status again, so that we have updated info stored
self.refresh(only_fields=['status', 'start'])
def serialize_depends(self, cur_dependencies):
# Check that all the tasks are saved
for task in (cur_dependencies or set()):
if not task.saved:
raise Task.NotSaved(
'Task \'%s\' needs to be saved before '
'it can be set as dependency.' % task,
)
return super(Task, self).serialize_depends(cur_dependencies)
def remove_annotation(self, annotation):
if not self.saved:
raise Task.NotSaved('Task needs to be saved to remove annotation')
if isinstance(annotation, TaskAnnotation):
annotation = annotation['description']
self.backend.denotate_task(self, annotation)
self.refresh(only_fields=['annotations'])
def add_annotation(self, annotation):
if not self.saved:
raise Task.NotSaved('Task needs to be saved to add annotation')
self.backend.annotate_task(self, annotation)
self.refresh(only_fields=['annotations'])
def refresh(self, only_fields=None, after_save=False):
# Raise error when trying to refresh a task that has not been saved
if not self.saved:
raise Task.NotSaved('Task needs to be saved to be refreshed')
new_data = self.backend.refresh_task(self, after_save=after_save)
if only_fields:
to_update = dict(
[(k, new_data.get(k)) for k in only_fields],
)
self._update_data(to_update, update_original=True)
else:
self._load_data(new_data)
def done(self):
if not self.saved:
raise Task.NotSaved(
'Task needs to be saved before it can be completed',
)
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if self.completed:
raise Task.CompletedTask('Cannot complete a completed task')
elif self.deleted:
raise Task.DeletedTask('Deleted task cannot be completed')
self.backend.complete_task(self)
# Refresh the status again, so that we have updated info stored
self.refresh(only_fields=['status', 'start', 'end'])