(self, set_new_name=True)
| 471 | return {} |
| 472 | |
| 473 | def duplicate(self, set_new_name=True): |
| 474 | try: |
| 475 | with transaction.atomic(): |
| 476 | task = Task.objects.get(pk=self.pk) |
| 477 | task.pk = None |
| 478 | if set_new_name: |
| 479 | task.name = gettext('Copy of %(task)s') % {'task': self.name} |
| 480 | task.created_at = timezone.now() |
| 481 | task.save() |
| 482 | task.refresh_from_db() |
| 483 | |
| 484 | logger.info("Duplicating {} to {}".format(self, task)) |
| 485 | |
| 486 | if os.path.isdir(self.task_path()): |
| 487 | try: |
| 488 | # Try to use hard links first |
| 489 | shutil.copytree(self.task_path(), task.task_path(), copy_function=os.link) |
| 490 | except Exception as e: |
| 491 | logger.warning("Cannot duplicate task using hard links, will use normal copy instead: {}".format(str(e))) |
| 492 | shutil.copytree(self.task_path(), task.task_path()) |
| 493 | else: |
| 494 | logger.warning("Task {} doesn't have folder, will skip copying".format(self)) |
| 495 | |
| 496 | self.project.owner.profile.clear_used_quota_cache() |
| 497 | |
| 498 | from app.plugins import signals as plugin_signals |
| 499 | plugin_signals.task_duplicated.send_robust(sender=self.__class__, task_id=task.id) |
| 500 | |
| 501 | return task |
| 502 | except Exception as e: |
| 503 | logger.warning("Cannot duplicate task: {}".format(str(e))) |
| 504 | |
| 505 | return False |
| 506 | |
| 507 | def write_backup_file(self): |
| 508 | """Dump this tasks's fields to a backup file""" |
nothing calls this directly
no test coverage detected