# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt

from unittest.mock import patch

import frappe
from frappe.utils import add_days, getdate, nowdate

from erpnext.projects.doctype.task.task import (
	CircularReferenceError,
	ParentIsGroupError,
	set_tasks_as_overdue,
)
from erpnext.tests.utils import ERPNextTestSuite


class TestTask(ERPNextTestSuite):
	def test_task_total_costing_and_billing_amount(self):
		from erpnext.projects.doctype.project.test_project import make_project
		from erpnext.projects.doctype.timesheet.test_timesheet import make_timesheet
		from erpnext.setup.doctype.employee.test_employee import make_employee

		project_name = "Test Project Costing"
		employee = make_employee("employee@frappe.io", company="_Test Company")
		project = make_project({"project_name": project_name})
		task = create_task("_Test Task 1")
		task.project = project.name
		task.save()
		timesheet = make_timesheet(
			employee=employee,
			is_billable=1,
			currency="USD",
			project=project.name,
			simulate=True,
			exchange_rate=80,
			task=task.name,
		)
		timesheet.reload()
		project.reload()
		task.reload()
		self.assertEqual(task.total_costing_amount, 3200)
		self.assertEqual(task.total_billing_amount, 8000)

	def test_move_task_to_another_project(self):
		from erpnext.projects.doctype.project.test_project import make_project

		old_project = make_project({"project_name": "_Test Project Task Moved Out"})
		new_project = make_project({"project_name": "_Test Project Task Moved In"})
		completed_task = create_task("_Test Task Completed In Old Project", save=False)
		completed_task.update({"project": old_project.name, "status": "Completed"})
		completed_task.save()
		moved_task = create_task("_Test Task Moved", save=False)
		moved_task.project = old_project.name
		moved_task.save()

		moved_task.project = new_project.name
		moved_task.save()

		self.assertEqual(frappe.db.get_value("Project", old_project.name, "percent_complete"), 100)

	def test_timesheet_on_cancelled_task(self):
		from erpnext.projects.doctype.timesheet.test_timesheet import make_timesheet
		from erpnext.setup.doctype.employee.test_employee import make_employee

		task = create_task("_Test Cancelled Task With Timesheet")
		task.status = "Cancelled"
		task.save()
		employee = make_employee("test_task_cancelled_timesheet@example.com", company="_Test Company")

		self.assertRaises(
			frappe.ValidationError,
			make_timesheet,
			employee,
			simulate=True,
			project=task.project,
			task=task.name,
		)

	def test_circular_reference(self):
		task1 = create_task("_Test Task 1", add_days(nowdate(), -15), add_days(nowdate(), -10))
		task2 = create_task("_Test Task 2", add_days(nowdate(), 11), add_days(nowdate(), 15), task1.name)
		task3 = create_task("_Test Task 3", add_days(nowdate(), 11), add_days(nowdate(), 15), task2.name)

		task1.reload()
		task1.append("depends_on", {"task": task3.name})

		self.assertRaises(CircularReferenceError, task1.save)

		task1.set("depends_on", [])
		task1.save()

		task4 = create_task("_Test Task 4", nowdate(), add_days(nowdate(), 15), task1.name)

		task3.append("depends_on", {"task": task4.name})

	def test_reschedule_dependent_task(self):
		task1 = create_task("_Test Task 1", nowdate(), add_days(nowdate(), 10))
		task2 = create_task("_Test Task 2", add_days(nowdate(), 11), add_days(nowdate(), 15), task1.name)
		task3 = create_task("_Test Task 3", add_days(nowdate(), 11), add_days(nowdate(), 15), task2.name)

		task1.update({"exp_end_date": add_days(nowdate(), 20)})
		task1.save()

		self.assertEqual(
			getdate(frappe.db.get_value("Task", task2.name, "exp_start_date")),
			getdate(add_days(nowdate(), 21)),
		)

		self.assertEqual(
			getdate(frappe.db.get_value("Task", task2.name, "exp_end_date")), getdate(add_days(nowdate(), 25))
		)

		self.assertEqual(
			getdate(frappe.db.get_value("Task", task3.name, "exp_start_date")),
			getdate(add_days(nowdate(), 26)),
		)

		self.assertEqual(
			getdate(frappe.db.get_value("Task", task3.name, "exp_end_date")), getdate(add_days(nowdate(), 30))
		)

	def test_child_task_does_not_reschedule_its_parent(self):
		parent = create_task("_Test Parent Not Rescheduled", nowdate(), add_days(nowdate(), 10), is_group=1)
		child = create_task(
			"_Test Child Of Parent Not Rescheduled",
			nowdate(),
			add_days(nowdate(), 3),
			parent_task=parent.name,
		)

		child.save()

		self.assertEqual(
			getdate(frappe.db.get_value("Task", parent.name, "exp_start_date")), getdate(nowdate())
		)

	def test_reschedule_started_dependent_tasks(self):
		prerequisite = create_task("_Test Task Before Started Tasks", nowdate(), add_days(nowdate(), 2))
		working = create_task(
			"_Test Working Dependent", add_days(nowdate(), 1), add_days(nowdate(), 3), prerequisite.name
		)
		overdue = create_task(
			"_Test Overdue Dependent", add_days(nowdate(), -3), add_days(nowdate(), -1), prerequisite.name
		)
		working.db_set("status", "Working")
		overdue.db_set("status", "Overdue")

		prerequisite.save()

		for task in (working, overdue):
			self.assertEqual(
				getdate(frappe.db.get_value("Task", task.name, "exp_start_date")),
				getdate(add_days(nowdate(), 3)),
			)

	def test_reschedule_past_parent_end_date_warns(self):
		parent = create_task("_Test Group Ending Soon", nowdate(), add_days(nowdate(), 10), is_group=1)
		prerequisite = create_task("_Test Task Before Grouped Task", nowdate(), add_days(nowdate(), 2))
		dependent = create_task(
			"_Test Grouped Dependent",
			add_days(nowdate(), 3),
			add_days(nowdate(), 5),
			prerequisite.name,
			parent_task=parent.name,
		)

		prerequisite.exp_end_date = add_days(nowdate(), 8)
		prerequisite.save()

		self.assertEqual(
			getdate(frappe.db.get_value("Task", dependent.name, "exp_end_date")),
			getdate(add_days(nowdate(), 11)),
		)
		self.assertIn(dependent.name, frappe.get_message_log()[-1]["message"])

	def test_rescheduled_task_saves_after_reload(self):
		parent = create_task("_Test Group Of Rescheduled", nowdate(), add_days(nowdate(), 10), is_group=1)
		prerequisite = create_task("_Test Task Before Rescheduled", nowdate(), add_days(nowdate(), 2))
		dependent = create_task(
			"_Test Rescheduled Past Group",
			add_days(nowdate(), 3),
			add_days(nowdate(), 5),
			prerequisite.name,
			parent_task=parent.name,
		)
		prerequisite.exp_end_date = add_days(nowdate(), 8)
		prerequisite.save()

		dependent.reload()
		dependent.status = "Working"
		dependent.save()

		dependent.exp_end_date = add_days(nowdate(), 12)
		self.assertRaises(frappe.exceptions.InvalidDates, dependent.save)

	def test_reschedule_dependent_task_from_actual_end_date(self):
		prerequisite = create_task("_Test Task Actual End", save=False)
		prerequisite.exp_start_date = prerequisite.exp_end_date = None
		prerequisite.save()
		dependent = create_task(
			"_Test Task After Actual End", add_days(nowdate(), -5), add_days(nowdate(), -3), prerequisite.name
		)

		prerequisite.act_end_date = nowdate()
		prerequisite.save()

		self.assertEqual(
			getdate(frappe.db.get_value("Task", dependent.name, "exp_start_date")),
			getdate(add_days(nowdate(), 1)),
		)

	def test_actual_dates_outside_project_dates(self):
		project = frappe.get_value("Project", {"project_name": "_Test Project"})
		frappe.db.set_value(
			"Project",
			project,
			{"expected_start_date": add_days(nowdate(), -10), "expected_end_date": add_days(nowdate(), 5)},
		)
		task = create_task("_Test Task Late Work", nowdate(), add_days(nowdate(), 3))
		task.act_start_date = task.act_end_date = add_days(nowdate(), 8)

		with patch.object(frappe, "in_test", False):
			task.validate_parent_project_dates()

			task.exp_end_date = add_days(nowdate(), 8)
			self.assertRaises(frappe.exceptions.InvalidDates, task.validate_parent_project_dates)

	def test_close_assignment(self):
		if not frappe.db.exists("Task", "Test Close Assignment"):
			task = frappe.new_doc("Task")
			task.subject = "Test Close Assignment"
			task.insert()

		def assign():
			from frappe.desk.form import assign_to

			assign_to.add(
				{
					"assign_to": ["test@example.com"],
					"doctype": task.doctype,
					"name": task.name,
					"description": "Close this task",
				}
			)

		def get_owner_and_status():
			return frappe.db.get_value(
				"ToDo",
				filters={
					"reference_type": task.doctype,
					"reference_name": task.name,
					"description": "Close this task",
				},
				fieldname=("allocated_to", "status"),
				as_dict=True,
			)

		assign()
		todo = get_owner_and_status()
		self.assertEqual(todo.allocated_to, "test@example.com")
		self.assertEqual(todo.status, "Open")

		# assignment should be
		task.load_from_db()
		task.status = "Completed"
		task.save()
		todo = get_owner_and_status()
		self.assertEqual(todo.allocated_to, "test@example.com")
		self.assertEqual(todo.status, "Closed")

	def test_complete_assigned_task_without_task_read_permission(self):
		from frappe.core.doctype.user_permission.test_user_permission import create_user
		from frappe.desk.form import assign_to

		task = create_task("_Test Assigned Task Completed By Timesheet")
		assign_to.add({"doctype": task.doctype, "name": task.name, "assign_to": ["test@example.com"]})
		user = create_user("test_task_timesheet_user@example.com", "HR User")

		with self.set_user(user.name):
			task.status = "Completed"
			task.save(ignore_permissions=True)

		self.assertEqual(
			frappe.db.get_value(
				"ToDo", {"reference_type": task.doctype, "reference_name": task.name}, "status"
			),
			"Closed",
		)

	def test_overdue(self):
		task = create_task("Testing Overdue", add_days(nowdate(), -10), add_days(nowdate(), -5))

		set_tasks_as_overdue()

		self.assertEqual(frappe.db.get_value("Task", task.name, "status"), "Overdue")

	def test_overdue_cleared_when_end_date_moves_out(self):
		task = create_task("_Test Task Overdue Moved Out", add_days(nowdate(), -10), add_days(nowdate(), -5))
		set_tasks_as_overdue()

		task.reload()
		task.exp_end_date = add_days(nowdate(), 10)
		task.save()

		self.assertEqual(task.status, "Open")

	def test_task_due_today_is_not_overdue(self):
		task = create_task("_Test Task Due Today", add_days(nowdate(), -2), nowdate())

		set_tasks_as_overdue()

		self.assertEqual(frappe.db.get_value("Task", task.name, "status"), "Open")

	def test_template_task_is_not_overdue(self):
		task = create_task(
			"_Test Template Task Overdue", add_days(nowdate(), -10), add_days(nowdate(), -5), is_template=1
		)

		set_tasks_as_overdue()

		self.assertEqual(frappe.db.get_value("Task", task.name, "status"), "Template")

	def test_parent_task_must_be_group(self):
		parent_task = create_task(
			subject="_Test Parent Task Non Group",
			is_group=0,
		)

		child_task = create_task(
			subject="_Test Child Task",
			parent_task=parent_task.name,
			save=False,
		)

		self.assertRaises(ParentIsGroupError, child_task.save)

	def test_open_task_under_completed_parent(self):
		parent = create_task("_Test Completed Parent", is_group=1)
		parent.status = "Completed"
		parent.save()
		child = create_task("_Test Open Child Of Completed Parent", parent_task=parent.name, save=False)

		self.assertRaises(frappe.ValidationError, child.save)

	def test_expected_end_date(self):
		task = create_task("Testing End Date", add_days(nowdate(), 1), add_days(nowdate(), 5))
		task.expected_time = 72
		task.save()
		self.assertEqual(getdate(task.exp_end_date), getdate(add_days(nowdate(), 5)))

	def test_completing_sets_completed_on(self):
		from frappe.core.doctype.user_permission.test_user_permission import create_user

		task = create_task("_Test Complete Without Date")
		user = create_user("test_task_completer@example.com", "Projects User")
		with self.set_user(user.name):
			frappe.set_value("Task", task.name, "status", "Completed")

		self.assertEqual(frappe.db.get_value("Task", task.name, "completed_on"), getdate())

	def test_delete_child_task(self):
		parent = create_task("_Test Parent Of Deleted Child", is_group=1)
		child = create_task("_Test Deleted Child", parent_task=parent.name)

		child.delete()

		parent.reload()
		self.assertFalse(parent.depends_on)

	def test_delete_child_task_without_parent_write_permission(self):
		from frappe.core.doctype.user_permission.test_user_permission import create_user

		from erpnext.projects.doctype.project.test_project import make_project

		parent = create_task("_Test Parent In Restricted Project", is_group=1)
		child = create_task("_Test Child In Permitted Project", parent_task=parent.name)
		child_project = make_project({"project_name": "_Test Project Child Only"}).name
		frappe.db.set_value("Task", child.name, "project", child_project)
		user = create_user("test_task_child_deleter@example.com", "Projects User")
		frappe.permissions.add_user_permission("Project", child_project, user.name)

		with self.set_user(user.name):
			frappe.delete_doc("Task", child.name)

		self.assertFalse(frappe.db.exists("Task", child.name))

	def test_move_child_task_to_another_parent(self):
		old_parent = create_task("_Test Old Parent", is_group=1)
		new_parent = create_task("_Test New Parent", is_group=1)
		child = create_task("_Test Moved Child", parent_task=old_parent.name)

		child.parent_task = new_parent.name
		child.save()

		self.assertFalse(frappe.get_all("Task Depends On", filters={"parent": old_parent.name}))
		self.assertEqual(
			frappe.get_all("Task Depends On", filters={"parent": new_parent.name}, pluck="task"), [child.name]
		)

	def test_reopen_completed_task(self):
		task = create_task("_Test Task Reopened")
		task.status = "Completed"
		task.save()

		task.status = "Open"
		task.save()
		self.assertEqual((task.progress, task.completed_on), (0, None))

		task.status = "Completed"
		task.save()
		task.update({"status": "Working", "progress": 60})
		task.save()
		self.assertEqual(task.progress, 60)

	def test_negative_progress(self):
		task = create_task("_Test Task Negative Progress", save=False)
		task.progress = -50

		self.assertRaises(frappe.ValidationError, task.save)

	def test_add_multiple_tasks_under_project_root(self):
		from erpnext.projects.doctype.task.task import add_multiple_tasks

		project = frappe.get_value("Project", {"project_name": "_Test Project"})

		add_multiple_tasks(frappe.as_json([{"subject": "_Test Bulk Root Task"}]), project, project)

		self.assertEqual(
			frappe.db.get_value("Task", {"subject": "_Test Bulk Root Task"}, ["project", "parent_task"]),
			(project, None),
		)

	def test_moving_child_out_does_not_reschedule_old_parent(self):
		group = create_task("_Test Group Left By Child", nowdate(), add_days(nowdate(), 19), is_group=1)
		child = create_task(
			"_Test Child Leaving Group",
			add_days(nowdate(), 4),
			add_days(nowdate(), 9),
			parent_task=group.name,
		)
		dependent = create_task(
			"_Test Task After Left Group", add_days(nowdate(), 20), add_days(nowdate(), 21), group.name
		)

		child.parent_task = None
		child.save()

		self.assertEqual(
			getdate(frappe.db.get_value("Task", group.name, "exp_end_date")), getdate(add_days(nowdate(), 19))
		)
		self.assertEqual(
			getdate(frappe.db.get_value("Task", dependent.name, "exp_start_date")),
			getdate(add_days(nowdate(), 20)),
		)

	def test_timesheet_by_user_without_task_access_reschedules_dependent(self):
		from frappe.core.doctype.user_permission.test_user_permission import create_user

		from erpnext.projects.doctype.timesheet.test_timesheet import make_timesheet
		from erpnext.setup.doctype.employee.test_employee import make_employee

		prerequisite = create_task("_Test Task Logged By HR User", nowdate(), add_days(nowdate(), 5))
		dependent = create_task(
			"_Test Task After HR User Log", nowdate(), add_days(nowdate(), 1), prerequisite.name
		)
		employee = make_employee("test_employee_6@salary.com", company="_Test Company")
		user = create_user("test_task_hr_logger@example.com", "HR User")

		with self.set_user(user.name):
			make_timesheet(employee, simulate=True, task=prerequisite.name)

		self.assertEqual(
			getdate(frappe.db.get_value("Task", dependent.name, "exp_start_date")),
			getdate(add_days(nowdate(), 6)),
		)

	def test_parent_task_must_be_in_the_same_project(self):
		other_project = frappe.get_doc(
			doctype="Project", project_name="_Test Parent Task Project", company="_Test Company"
		).insert()
		group = frappe.get_doc(
			doctype="Task", subject="_Test Other Project Group", project=other_project.name, is_group=1
		).insert()

		child_task = create_task("_Test Child In Another Project", parent_task=group.name, save=False)
		self.assertRaises(frappe.ValidationError, child_task.save)

		child_without_project = frappe.get_doc(
			doctype="Task", subject="_Test Child Without Project", parent_task=group.name
		).insert()
		self.assertEqual(child_without_project.project, other_project.name)

	def test_child_takes_the_company_of_the_inherited_project(self):
		project = frappe.get_doc(
			doctype="Project", project_name="_Test Inherited Company Project", company="_Test Company 1"
		).insert()
		group = frappe.get_doc(
			doctype="Task", subject="_Test Inherited Company Group", project=project.name, is_group=1
		).insert()

		child = frappe.get_doc(
			doctype="Task", subject="_Test Inherited Company Child", parent_task=group.name
		).insert()
		self.assertEqual(child.company, "_Test Company 1")

	def test_child_with_blank_project_saves_under_a_group_without_project(self):
		group = frappe.get_doc(doctype="Task", subject="_Test Group Without Project", is_group=1).insert()

		child = frappe.get_doc(
			doctype="Task", subject="_Test Child With Blank Project", project="", parent_task=group.name
		).insert()
		self.assertFalse(child.project)

	def test_group_with_children_cannot_move_to_another_project(self):
		from erpnext.projects.doctype.project.test_project import make_project

		group = create_task("_Test Group Moving Project", is_group=1)
		create_task("_Test Child Of Group Moving Project", parent_task=group.name)
		group.reload()

		group.project = make_project({"project_name": "_Test Project Group Target"}).name
		self.assertRaises(frappe.ValidationError, group.save)

		group_without_project = frappe.get_doc(
			doctype="Task", subject="_Test Group Without Project", is_group=1
		).insert()
		frappe.get_doc(
			doctype="Task", subject="_Test Child Without Project", parent_task=group_without_project.name
		).insert()
		group_without_project.reload()
		group_without_project.project = group.project
		self.assertRaises(frappe.ValidationError, group_without_project.save)

	def test_child_tasks_are_listed_only_for_a_readable_task(self):
		from frappe.core.doctype.user_permission.test_user_permission import create_user

		from erpnext.projects.doctype.project.test_project import make_project
		from erpnext.projects.doctype.task.task import check_if_child_exists

		group = create_task("_Test Group Hidden From User", is_group=1)
		create_task("_Test Child Hidden From User", parent_task=group.name)
		user = create_user("test_task_child_lister@example.com", "Projects User")
		other_project = make_project({"project_name": "_Test Project Child Lister"}).name
		frappe.permissions.add_user_permission("Project", other_project, user.name)

		with self.set_user(user.name):
			self.assertRaises(frappe.PermissionError, check_if_child_exists, group.name)

	def test_root_tasks_are_not_listed_as_children_of_an_empty_name(self):
		from frappe.core.doctype.user_permission.test_user_permission import create_user

		from erpnext.projects.doctype.project.test_project import make_project
		from erpnext.projects.doctype.task.task import check_if_child_exists

		create_task("_Test Root Hidden From User")
		user = create_user("test_task_root_lister@example.com", "Projects User")
		other_project = make_project({"project_name": "_Test Project Root Lister"}).name
		frappe.permissions.add_user_permission("Project", other_project, user.name)

		with self.set_user(user.name):
			self.assertRaises(frappe.DoesNotExistError, check_if_child_exists, "")


def create_task(
	subject,
	start=None,
	end=None,
	depends_on=None,
	project=None,
	parent_task=None,
	is_group=0,
	is_template=0,
	begin=0,
	duration=0,
	save=True,
	priority=None,
):
	if not frappe.db.exists("Task", subject):
		task = frappe.new_doc("Task")
		task.status = "Open"
		task.subject = subject
		task.exp_start_date = start or nowdate()
		task.exp_end_date = end or nowdate()
		task.project = (
			project or None if is_template else frappe.get_value("Project", {"project_name": "_Test Project"})
		)
		task.is_template = is_template
		task.start = begin
		task.duration = duration
		task.is_group = is_group
		task.parent_task = parent_task
		task.priority = priority
		if save:
			task.save()
	else:
		task = frappe.get_doc("Task", subject)

	if depends_on:
		task.append("depends_on", {"task": depends_on})
		if save:
			task.save()
	return task