feat: clone entire application (hl-1464) (#3369)

* feat: wip frotned clone application

* chore: loosen validation for null/false case on association fields

* feat: clone the entire application

* test: clone application

* fix: migration for company_bank_account_number

* fix: needs .png extension to file's write path

* fix: sometimes testiaineisto does not provide with company address

backend/benefit/applications/api/v1/application_views.py

@@ -623,16 +623,9 @@ def get_queryset(self):
  )
  @action(methods=["GET"], detail=True, url_path="clone_as_draft")
  @transaction.atomic
- def clone_as_draft(self, request, pk=None) -> HttpResponse:
+ def clone_as_draft(self, request, pk) -> HttpResponse:
  application_base = self.get_object()
-
- clone_employee = request.query_params.get("employee") or None
- clone_work = request.query_params.get("work") or None
- clone_subsidies = request.query_params.get("pay_subsidy") or None
-
- cloned_application = clone_application_based_on_other(
- application_base, clone_employee, clone_work, clone_subsidies
- )
+ cloned_application = clone_application_based_on_other(application_base)
 
  return Response(
  {"id": cloned_application.id},
@@ -667,7 +660,7 @@ def clone_as_draft(self, request, pk=None) -> HttpResponse:
  )
  @action(methods=["GET"], detail=False, url_path="clone_latest")
  @transaction.atomic
- def clone_latest(self, request, pk=None) -> HttpResponse:
+ def clone_latest(self, request) -> HttpResponse:
  company = get_company_from_request(request)
 
  try:
@@ -688,13 +681,7 @@ def clone_latest(self, request, pk=None) -> HttpResponse:
  status=status.HTTP_404_NOT_FOUND,
  )
 
- clone_employee = request.query_params.get("employee") or None
- clone_work = request.query_params.get("work") or None
- clone_subsidies = request.query_params.get("pay_subsidy") or None
-
- cloned_application = clone_application_based_on_other(
- application_base, clone_employee, clone_work, clone_subsidies
- )
+ cloned_application = clone_application_based_on_other(application_base)
 
  return Response(
  {"id": cloned_application.id},
@@ -831,6 +818,28 @@ def change_handler(self, request, pk=None) -> HttpResponse:
  application.save()
  return Response(status=status.HTTP_200_OK)
 
+ @action(methods=["GET"], detail=True, url_path="clone_as_draft")
+ @transaction.atomic
+ def clone_as_draft(self, request, pk) -> HttpResponse:
+ application_base = self.get_object()
+ cloned_application = clone_application_based_on_other(application_base, True)
+
+ try:
+ cloned_application.full_clean()
+ cloned_application.employee.full_clean()
+ cloned_application.company.full_clean()
+ cloned_application.calculation.full_clean()
+
+ except exceptions.ValidationError as e:
+ return Response(
+ {"detail": e.message_dict}, status=status.HTTP_400_BAD_REQUEST
+ )
+
+ return Response(
+ {"id": cloned_application.id},
+ status=status.HTTP_201_CREATED,
+ )
+
  def _create_application_batch(self, status) -> QuerySet[Application]:
  """
  Create a new application batch out of the existing applications in the given status

backend/benefit/applications/api/v1/serializers/application.py

@@ -709,7 +709,7 @@ def _validate_association_immediate_manager_check(
  )
  }
  )
- elif association_immediate_manager_check is not None:
+ elif association_immediate_manager_check not in [None, False]:
  raise serializers.ValidationError(
  {
  "association_immediate_manager_check": _(
@@ -907,11 +907,12 @@ def _validate_non_draft_required_fields(self, data):
  def _validate_association_has_business_activities(
  self, company, association_has_business_activities
  ):
- if (
- OrganizationType.resolve_organization_type(company.company_form_code)
- == OrganizationType.COMPANY
- and association_has_business_activities is not None
- ):
+ if OrganizationType.resolve_organization_type(
+ company.company_form_code
+ ) == OrganizationType.COMPANY and association_has_business_activities not in [
+ None,
+ False,
+ ]:
  raise serializers.ValidationError(
  {
  "association_has_business_activities": _(

backend/benefit/applications/services/clone_application.py

@@ -1,25 +1,37 @@
-from applications.enums import ApplicationStep
-from applications.models import Application, DeMinimisAid, Employee
+from applications.enums import ApplicationStatus, ApplicationStep
+from applications.models import (
+ Application,
+ ApplicationLogEntry,
+ Attachment,
+ DeMinimisAid,
+ Employee,
+)
+from calculator.models import Calculation
 from companies.models import Company
+from helsinkibenefit.settings import MEDIA_ROOT
 
 
 def clone_application_based_on_other(
  application_base,
- clone_employee=False,
- clone_work=False,
- clone_subsidies=False,
+ clone_all_data=False,
 ):
  company = Company.objects.get(id=application_base.company.id)
+ company.street_address = (
+ company.street_address if len(company.street_address) > 0 else "Testikatu 123"
+ )
  cloned_application = Application(
  **{
  "alternative_company_city": application_base.alternative_company_city,
  "alternative_company_postcode": application_base.alternative_company_postcode,
  "alternative_company_street_address": application_base.alternative_company_street_address,
  "applicant_language": "fi",
+ "application_origin": application_base.application_origin,
  "application_step": ApplicationStep.STEP_1,
  "archived": False,
- "association_has_business_activities": application_base.association_has_business_activities,
- "association_immediate_manager_check": application_base.association_immediate_manager_check,
+ "association_has_business_activities": application_base.association_has_business_activities
+ or False,
+ "association_immediate_manager_check": application_base.association_immediate_manager_check
+ or False,
  "benefit_type": "salary_benefit",
  "co_operation_negotiations": application_base.co_operation_negotiations,
  "co_operation_negotiations_description": application_base.co_operation_negotiations_description,
@@ -29,58 +41,126 @@ def clone_application_based_on_other(
  "company_contact_person_last_name": application_base.company_contact_person_last_name,
  "company_contact_person_phone_number": application_base.company_contact_person_phone_number,
  "company_department": application_base.company_department,
+ "company_form": application_base.company_form,
  "company_form_code": company.company_form_code,
+ "company_name": application_base.company_name,
  "de_minimis_aid": application_base.de_minimis_aid,
- "status": "draft",
+ "status": ApplicationStatus.DRAFT,
+ "official_company_street_address": company.street_address,
+ "official_company_city": application_base.official_company_city,
+ "official_company_postcode": application_base.official_company_postcode,
  "use_alternative_address": application_base.use_alternative_address,
  }
  )
 
+ company.save()
  cloned_application.company = company
 
- if clone_employee or clone_work:
- employee = Employee.objects.get(id=application_base.employee.id)
- employee.pk = None
- employee.application = cloned_application
+ de_minimis_aids = application_base.de_minimis_aid_set.all()
+ if de_minimis_aids.exists():
+ cloned_application.de_minimis_aid = True
+
+ last_order = DeMinimisAid.objects.last().ordering + 1
+ for index, aid in enumerate(de_minimis_aids):
+ aid.pk = None
+ aid.ordering = last_order + index
+ aid.application = cloned_application
+ aid.save()
+
+ if clone_all_data:
+ cloned_application = _clone_handler_data(application_base, cloned_application)
  else:
  employee = Employee.objects.create(application=cloned_application)
-
- if not clone_employee:
  employee.first_name = ""
  employee.last_name = ""
  employee.social_security_number = ""
  employee.is_living_in_helsinki = False
-
- if not clone_work:
  employee.job_title = ""
  employee.monthly_pay = None
  employee.vacation_money = None
  employee.other_expenses = None
  employee.working_hours = None
  employee.collective_bargaining_agreement = ""
+ employee.save()
 
+ cloned_application.save()
+ return cloned_application
+
+
+def _clone_handler_data(application_base, cloned_application):
+ employee = Employee.objects.get(id=application_base.employee.id)
+ employee.pk = None
+ employee.application = cloned_application
  employee.save()
 
- if clone_subsidies:
- cloned_application.pay_subsidy_granted = application_base.pay_subsidy_granted
- cloned_application.apprenticeship_program = (
- application_base.apprenticeship_program
- )
- cloned_application.pay_subsidy_percent = application_base.pay_subsidy_percent
+ cloned_application.handled_by_ahjo_automation = (
+ application_base.handled_by_ahjo_automation
+ )
+ cloned_application.paper_application_date = application_base.paper_application_date
+ cloned_application.applicant_language = application_base.applicant_language
+ cloned_application.pay_subsidy_granted = application_base.pay_subsidy_granted
+ cloned_application.apprenticeship_program = (
+ application_base.apprenticeship_program or False
+ )
+ cloned_application.pay_subsidy_percent = application_base.pay_subsidy_percent
+ cloned_application.additional_pay_subsidy_percent = (
+ application_base.additional_pay_subsidy_percent
+ )
 
- de_minimis_aids = DeMinimisAid.objects.filter(
- application__pk=application_base.id
- ).all()
+ # Create fake image to be used as attachment's body
+ from PIL import Image
 
- if de_minimis_aids.exists():
- cloned_application.de_minimis_aid = True
+ attachment_name = f"test-application-{cloned_application.id}"
+ temp_image = Image.new("RGB", (1, 1))
+ temp_image.save(
+ format="PNG",
+ fp=f"{MEDIA_ROOT}/{attachment_name}.png",
+ )
 
- last_order = DeMinimisAid.objects.last().ordering + 1
- for index, aid in enumerate(de_minimis_aids):
- aid.pk = None
- aid.ordering = last_order + index
- aid.application = cloned_application
- aid.save()
+ # Mimick the attachments by retaining attachment type
+ for base_attachment in application_base.attachments.all():
+ Attachment.objects.create(
+ attachment_type=base_attachment.attachment_type,
+ application=cloned_application,
+ attachment_file=f"{attachment_name}.png",
+ content_type="image/png",
+ )
 
+ # Clone calculation with compensations and pay subsidies
+ cloned_application.start_date = application_base.start_date
+ cloned_application.end_date = application_base.end_date
  cloned_application.save()
+
+ calculation_base = application_base.calculation
+ Calculation.objects.create_for_application(
+ cloned_application,
+ start_date=calculation_base.start_date,
+ end_date=calculation_base.end_date,
+ state_aid_max_percentage=calculation_base.state_aid_max_percentage,
+ override_monthly_benefit_amount=calculation_base.override_monthly_benefit_amount,
+ override_monthly_benefit_amount_comment=calculation_base.override_monthly_benefit_amount_comment,
+ )
+
+ cloned_application.calculation.calculate(override_status=True)
+ cloned_application.calculation.save()
+
+ for compensation in application_base.training_compensations.all():
+ compensation.pk = None
+ compensation.application = cloned_application
+ compensation.save()
+
+ # Remove pay subsidies made by create_for_application, then clone the old ones
+ cloned_application.pay_subsidies.filter(start_date__isnull=True).delete()
+ for pay_subsidy in application_base.pay_subsidies.all():
+ pay_subsidy.pk = None
+ pay_subsidy.application = cloned_application
+ pay_subsidy.save()
+
+ cloned_application.status = ApplicationStatus.RECEIVED
+ ApplicationLogEntry.objects.create(
+ application=cloned_application,
+ from_status=ApplicationStatus.DRAFT,
+ to_status=cloned_application.status,
+ comment="",
+ )
  return cloned_application