import json
import logging

from unittest import mock

from django.conf import settings
from django.contrib.auth.models import Group, Permission
from django.core import mail
from django.core.mail import EmailMultiAlternatives
from django.test import TestCase, override_settings
from django.urls import reverse
from freezegun import freeze_time

from wagtail.core.models import (
    GroupApprovalTask, Page, Task, TaskState, Workflow, WorkflowPage, WorkflowState, WorkflowTask)
from wagtail.core.signals import page_published
from wagtail.tests.testapp.models import SimplePage, SimpleTask
from wagtail.tests.utils import WagtailTestUtils
from wagtail.users.models import UserProfile


def delete_existing_workflows():
    WorkflowPage.objects.all().delete()
    Workflow.objects.all().delete()
    Task.objects.all().delete()
    WorkflowTask.objects.all().delete()


class TestWorkflowsIndexView(TestCase, WagtailTestUtils):

    def setUp(self):
        delete_existing_workflows()
        self.login()

        self.editor = self.create_user(
            username='editor',
            email='editor@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.editor)

        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.permissions.add(Permission.objects.get(codename="add_workflow"))

    def get(self, params={}):
        return self.client.get(reverse('wagtailadmin_workflows:index'), params)

    def test_simple(self):
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/index.html')

        # Initially there should be no workflows listed
        self.assertContains(response, "There are no enabled workflows.")

        Workflow.objects.create(name="test_workflow", active=True)

        # Now the listing should contain our workflow
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/index.html')
        self.assertNotContains(response, "There are no enabled workflows.")
        self.assertContains(response, "test_workflow")

    def test_deactivated(self):
        Workflow.objects.create(name="test_workflow", active=False)

        # The listing should contain our workflow, as well as marking it as disabled
        response = self.get(params={'show_disabled': 'true'})
        self.assertEqual(response.status_code, 200)
        self.assertNotContains(response, "No workflows have been created.")
        self.assertContains(response, "test_workflow")
        self.assertContains(response, '<span class="status-tag">Disabled</span>', html=True)

        # If we set 'show_disabled' to 'False', the workflow should not be displayed
        response = self.get(params={})
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "There are no enabled workflows.")

    def test_permissions(self):
        self.login(user=self.editor)
        response = self.get()
        self.assertEqual(response.status_code, 302)
        full_context = {key: value for context in response.context for key, value in context.items()}
        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')

        self.login(user=self.moderator)
        response = self.get()
        self.assertEqual(response.status_code, 200)


class TestWorkflowsCreateView(TestCase, WagtailTestUtils):

    def setUp(self):
        delete_existing_workflows()
        self.login()
        self.task_1 = SimpleTask.objects.create(name="first_task")
        self.task_2 = SimpleTask.objects.create(name="second_task")

        self.editor = self.create_user(
            username='editor',
            email='editor@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.editor)

        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.permissions.add(Permission.objects.get(codename="add_workflow"))

        self.root_page = Page.objects.get(depth=1)

    def get(self, params={}):
        return self.client.get(reverse('wagtailadmin_workflows:add'), params)

    def post(self, post_data={}):
        return self.client.post(reverse('wagtailadmin_workflows:add'), post_data)

    def test_get(self):
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/create.html')

    def test_post(self):
        response = self.post({
            'name': ['test_workflow'],
            'active': ['on'],
            'workflow_tasks-TOTAL_FORMS': ['2'],
            'workflow_tasks-INITIAL_FORMS': ['0'],
            'workflow_tasks-MIN_NUM_FORMS': ['0'],
            'workflow_tasks-MAX_NUM_FORMS': ['1000'],
            'workflow_tasks-0-task': [str(self.task_1.id)],
            'workflow_tasks-0-id': [''],
            'workflow_tasks-0-ORDER': ['1'],
            'workflow_tasks-0-DELETE': [''],
            'workflow_tasks-1-task': [str(self.task_2.id)],
            'workflow_tasks-1-id': [''],
            'workflow_tasks-1-ORDER': ['2'],
            'workflow_tasks-1-DELETE': [''],
            'pages-TOTAL_FORMS': ['2'],
            'pages-INITIAL_FORMS': ['1'],
            'pages-MIN_NUM_FORMS': ['0'],
            'pages-MAX_NUM_FORMS': ['1000'],
            'pages-0-page': [str(self.root_page.id)],
            'pages-0-DELETE': [''],
            'pages-1-page': [''],
            'pages-1-DELETE': [''],
        })

        # Should redirect back to index
        self.assertRedirects(response, reverse('wagtailadmin_workflows:index'))

        # Check that the workflow was created
        workflows = Workflow.objects.filter(name="test_workflow", active=True)
        self.assertEqual(workflows.count(), 1)

        workflow = workflows.first()

        # Check that the tasks are associated with the workflow
        self.assertEqual([self.task_1.task_ptr, self.task_2.task_ptr], list(workflow.tasks))

        # Check that the tasks have sort_order set on WorkflowTask correctly
        self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_1.task_ptr).sort_order, 0)
        self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_2.task_ptr).sort_order, 1)

    def test_permissions(self):
        self.login(user=self.editor)
        response = self.get()
        self.assertEqual(response.status_code, 302)
        full_context = {key: value for context in response.context for key, value in context.items()}
        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')

        self.login(user=self.moderator)
        response = self.get()
        self.assertEqual(response.status_code, 200)

    def test_page_already_has_workflow_check(self):
        workflow = Workflow.objects.create(name="existing_workflow")
        WorkflowPage.objects.create(workflow=workflow, page=self.root_page)

        response = self.post({
            'name': ['test_workflow'],
            'active': ['on'],
            'workflow_tasks-TOTAL_FORMS': ['2'],
            'workflow_tasks-INITIAL_FORMS': ['0'],
            'workflow_tasks-MIN_NUM_FORMS': ['0'],
            'workflow_tasks-MAX_NUM_FORMS': ['1000'],
            'workflow_tasks-0-task': [str(self.task_1.id)],
            'workflow_tasks-0-id': [''],
            'workflow_tasks-0-ORDER': ['1'],
            'workflow_tasks-0-DELETE': [''],
            'workflow_tasks-1-task': [str(self.task_2.id)],
            'workflow_tasks-1-id': [''],
            'workflow_tasks-1-ORDER': ['2'],
            'workflow_tasks-1-DELETE': [''],
            'pages-TOTAL_FORMS': ['2'],
            'pages-INITIAL_FORMS': ['1'],
            'pages-MIN_NUM_FORMS': ['0'],
            'pages-MAX_NUM_FORMS': ['1000'],
            'pages-0-page': [str(self.root_page.id)],
            'pages-0-DELETE': [''],
            'pages-1-page': [''],
            'pages-1-DELETE': [''],
        })

        self.assertEqual(response.status_code, 200)
        self.assertFormsetError(response, 'pages_formset', 0, 'page', ["This page already has workflow 'existing_workflow' assigned."])


class TestWorkflowsEditView(TestCase, WagtailTestUtils):

    def setUp(self):
        delete_existing_workflows()
        self.login()
        self.workflow = Workflow.objects.create(name="workflow_to_edit")
        self.task_1 = SimpleTask.objects.create(name="first_task")
        self.task_2 = SimpleTask.objects.create(name="second_task")
        self.inactive_task = SimpleTask.objects.create(name="inactive_task", active=False)
        self.workflow_task = WorkflowTask.objects.create(workflow=self.workflow, task=self.task_1.task_ptr, sort_order=0)
        self.page = Page.objects.first()
        WorkflowPage.objects.create(workflow=self.workflow, page=self.page)

        self.editor = self.create_user(
            username='editor',
            email='editor@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.editor)

        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.permissions.add(Permission.objects.get(codename="change_workflow"))

    def get(self, params={}):
        return self.client.get(reverse('wagtailadmin_workflows:edit', args=[self.workflow.id]), params)

    def post(self, post_data={}):
        return self.client.post(reverse('wagtailadmin_workflows:edit', args=[self.workflow.id]), post_data)

    def test_get(self):
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/edit.html')

        # Check that the list of pages has the page to which this workflow is assigned
        self.assertContains(response, self.page.title)

    def test_post(self):
        response = self.post({
            'name': [str(self.workflow.name)],
            'active': ['on'],
            'workflow_tasks-TOTAL_FORMS': ['2'],
            'workflow_tasks-INITIAL_FORMS': ['1'],
            'workflow_tasks-MIN_NUM_FORMS': ['0'],
            'workflow_tasks-MAX_NUM_FORMS': ['1000'],
            'workflow_tasks-0-task': [str(self.task_1.id)],
            'workflow_tasks-0-id': [str(self.workflow_task.id)],
            'workflow_tasks-0-ORDER': ['1'],
            'workflow_tasks-0-DELETE': [''],
            'workflow_tasks-1-task': [str(self.task_2.id)],
            'workflow_tasks-1-id': [''],
            'workflow_tasks-1-ORDER': ['2'],
            'workflow_tasks-1-DELETE': [''],
            'pages-TOTAL_FORMS': ['2'],
            'pages-INITIAL_FORMS': ['1'],
            'pages-MIN_NUM_FORMS': ['0'],
            'pages-MAX_NUM_FORMS': ['1000'],
            'pages-0-page': [str(self.page.id)],
            'pages-0-DELETE': [''],
            'pages-1-page': [''],
            'pages-1-DELETE': [''],
        })

        # Should redirect back to index
        self.assertRedirects(response, reverse('wagtailadmin_workflows:index'))

        # Check that the workflow was created
        workflows = Workflow.objects.filter(name="workflow_to_edit", active=True)
        self.assertEqual(workflows.count(), 1)

        workflow = workflows.first()

        # Check that the tasks are associated with the workflow
        self.assertEqual([self.task_1.task_ptr, self.task_2.task_ptr], list(workflow.tasks))

        # Check that the tasks have sort_order set on WorkflowTask correctly
        self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_1.task_ptr).sort_order, 0)
        self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_2.task_ptr).sort_order, 1)

    def test_permissions(self):
        self.login(user=self.editor)
        response = self.get()
        self.assertEqual(response.status_code, 302)
        full_context = {key: value for context in response.context for key, value in context.items()}
        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')

        self.login(user=self.moderator)
        response = self.get()
        self.assertEqual(response.status_code, 200)

    def test_duplicate_page_check(self):
        response = self.post({
            'name': [str(self.workflow.name)],
            'active': ['on'],
            'workflow_tasks-TOTAL_FORMS': ['2'],
            'workflow_tasks-INITIAL_FORMS': ['1'],
            'workflow_tasks-MIN_NUM_FORMS': ['0'],
            'workflow_tasks-MAX_NUM_FORMS': ['1000'],
            'workflow_tasks-0-task': [str(self.task_1.id)],
            'workflow_tasks-0-id': [str(self.workflow_task.id)],
            'workflow_tasks-0-ORDER': ['1'],
            'workflow_tasks-0-DELETE': [''],
            'workflow_tasks-1-task': [str(self.task_2.id)],
            'workflow_tasks-1-id': [''],
            'workflow_tasks-1-ORDER': ['2'],
            'workflow_tasks-1-DELETE': [''],
            'pages-TOTAL_FORMS': ['2'],
            'pages-INITIAL_FORMS': ['1'],
            'pages-MIN_NUM_FORMS': ['0'],
            'pages-MAX_NUM_FORMS': ['1000'],
            'pages-0-page': [str(self.page.id)],
            'pages-0-DELETE': [''],
            'pages-1-page': [str(self.page.id)],
            'pages-1-DELETE': [''],
        })

        self.assertEqual(response.status_code, 200)
        self.assertFormsetError(response, 'pages_formset', None, None, ['You cannot assign this workflow to the same page multiple times.'])

    def test_pages_ignored_if_workflow_disabled(self):
        self.workflow.active = False
        self.workflow.save()
        self.workflow.workflow_pages.all().delete()

        response = self.post({
            'name': [str(self.workflow.name)],
            'active': ['on'],
            'workflow_tasks-TOTAL_FORMS': ['2'],
            'workflow_tasks-INITIAL_FORMS': ['1'],
            'workflow_tasks-MIN_NUM_FORMS': ['0'],
            'workflow_tasks-MAX_NUM_FORMS': ['1000'],
            'workflow_tasks-0-task': [str(self.task_1.id)],
            'workflow_tasks-0-id': [str(self.workflow_task.id)],
            'workflow_tasks-0-ORDER': ['1'],
            'workflow_tasks-0-DELETE': [''],
            'workflow_tasks-1-task': [str(self.task_2.id)],
            'workflow_tasks-1-id': [''],
            'workflow_tasks-1-ORDER': ['2'],
            'workflow_tasks-1-DELETE': [''],
            'pages-TOTAL_FORMS': ['2'],
            'pages-INITIAL_FORMS': ['1'],
            'pages-MIN_NUM_FORMS': ['0'],
            'pages-MAX_NUM_FORMS': ['1000'],
            'pages-0-page': [str(self.page.id)],
            'pages-0-DELETE': [''],
            'pages-1-page': [''],
            'pages-1-DELETE': [''],
        })

        # Should redirect back to index
        self.assertRedirects(response, reverse('wagtailadmin_workflows:index'))

        # Check that the pages weren't added to the workflow
        self.workflow.refresh_from_db()
        self.assertFalse(self.workflow.workflow_pages.exists())


class TestRemoveWorkflow(TestCase, WagtailTestUtils):
    fixtures = ['test.json']

    def setUp(self):
        delete_existing_workflows()
        self.login()
        self.workflow = Workflow.objects.create(name="workflow")
        self.page = Page.objects.first()
        WorkflowPage.objects.create(workflow=self.workflow, page=self.page)

        self.editor = self.create_user(
            username='editor',
            email='editor@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.editor)

        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.permissions.add(Permission.objects.get(codename="change_workflow"))

    def post(self, post_data={}):
        return self.client.post(reverse('wagtailadmin_workflows:remove', args=[self.page.id, self.workflow.id]), post_data)

    def test_post(self):
        # Check that a WorkflowPage instance is removed correctly
        self.post()
        self.assertEqual(WorkflowPage.objects.filter(workflow=self.workflow, page=self.page).count(), 0)

    def test_no_permissions(self):
        self.login(user=self.editor)
        response = self.post()
        self.assertEqual(response.status_code, 302)

    def test_post_with_permission(self):
        self.login(user=self.moderator)
        response = self.post()
        self.assertEqual(response.status_code, 302)


class TestTaskIndexView(TestCase, WagtailTestUtils):

    def setUp(self):
        delete_existing_workflows()
        self.login()

        self.editor = self.create_user(
            username='editor',
            email='editor@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.editor)

        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.permissions.add(Permission.objects.get(codename="change_task"))

    def get(self, params={}):
        return self.client.get(reverse('wagtailadmin_workflows:task_index'), params)

    def test_simple(self):
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/task_index.html')

        # Initially there should be no tasks listed
        self.assertContains(response, "There are no enabled tasks")

        SimpleTask.objects.create(name="test_task", active=True)

        # Now the listing should contain our task
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/task_index.html')
        self.assertNotContains(response, "There are no enabled tasks")
        self.assertContains(response, "test_task")

    def test_deactivated(self):
        Task.objects.create(name="test_task", active=False)

        # The listing should contain our task, as well as marking it as disabled
        response = self.get(params={'show_disabled': 'true'})
        self.assertEqual(response.status_code, 200)
        self.assertNotContains(response, "No tasks have been created.")
        self.assertContains(response, "test_task")
        self.assertContains(response, '<span class="status-tag">Disabled</span>', html=True)

        # The listing should not contain task if show_disabled query parameter is 'False'
        response = self.get(params={})
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "There are no enabled tasks")
        self.assertNotContains(response, "test_task")

    def test_permissions(self):
        self.login(user=self.editor)
        response = self.get()
        self.assertEqual(response.status_code, 302)
        full_context = {key: value for context in response.context for key, value in context.items()}
        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')

        self.login(user=self.moderator)
        response = self.get()
        self.assertEqual(response.status_code, 200)


class TestCreateTaskView(TestCase, WagtailTestUtils):

    def setUp(self):
        delete_existing_workflows()
        self.login()

        self.editor = self.create_user(
            username='editor',
            email='editor@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.editor)

        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.permissions.add(Permission.objects.get(codename="add_task"))

    def get(self, url_kwargs=None, params={}):
        url_kwargs = url_kwargs or {}
        url_kwargs.setdefault('app_label', SimpleTask._meta.app_label)
        url_kwargs.setdefault('model_name', SimpleTask._meta.model_name)
        return self.client.get(reverse('wagtailadmin_workflows:add_task', kwargs=url_kwargs), params)

    def post(self, post_data={}):
        return self.client.post(reverse('wagtailadmin_workflows:add_task', kwargs={'app_label': SimpleTask._meta.app_label, 'model_name': SimpleTask._meta.model_name}), post_data)

    def test_get(self):
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/create_task.html')

    def test_get_with_non_task_model(self):
        response = self.get(url_kwargs={'app_label': 'wagtailcore', 'model_name': 'Site'})
        self.assertEqual(response.status_code, 404)

    def test_get_with_base_task_model(self):
        response = self.get(url_kwargs={'app_label': 'wagtailcore', 'model_name': 'Task'})
        self.assertEqual(response.status_code, 404)

    def test_post(self):
        response = self.post({'name': 'test_task', 'active': 'on'})

        # Should redirect back to index
        self.assertRedirects(response, reverse('wagtailadmin_workflows:task_index'))

        # Check that the task was created
        tasks = Task.objects.filter(name="test_task", active=True)
        self.assertEqual(tasks.count(), 1)

    def test_permissions(self):
        self.login(user=self.editor)
        response = self.get()
        self.assertEqual(response.status_code, 302)
        full_context = {key: value for context in response.context for key, value in context.items()}
        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')

        self.login(user=self.moderator)
        response = self.get()
        self.assertEqual(response.status_code, 200)


class TestSelectTaskTypeView(TestCase, WagtailTestUtils):

    def setUp(self):
        delete_existing_workflows()
        self.login()

    def get(self):
        return self.client.get(reverse('wagtailadmin_workflows:select_task_type'))

    def test_get(self):
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/select_task_type.html')

        # Check that the list of available task types includes SimpleTask and GroupApprovalTask
        self.assertContains(response, SimpleTask.get_verbose_name())
        self.assertContains(response, GroupApprovalTask.get_verbose_name())
        self.assertContains(response, GroupApprovalTask.get_description())


class TestEditTaskView(TestCase, WagtailTestUtils):

    def setUp(self):
        delete_existing_workflows()
        self.login()
        self.task = GroupApprovalTask.objects.create(name="test_task")

        self.editor = self.create_user(
            username='editor',
            email='editor@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.editor)

        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.permissions.add(Permission.objects.get(codename="change_task"))

    def get(self, params={}):
        return self.client.get(reverse('wagtailadmin_workflows:edit_task', args=[self.task.id]), params)

    def post(self, post_data={}):
        return self.client.post(reverse('wagtailadmin_workflows:edit_task', args=[self.task.id]), post_data)

    def test_get(self):
        response = self.get()
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/workflows/edit_task.html')

    def test_post(self):
        self.assertEqual(self.task.groups.count(), 0)
        editors = Group.objects.get(name='Editors')

        response = self.post({'name': 'test_task_modified', 'active': 'on', 'groups': [str(editors.id)]})

        # Should redirect back to index
        self.assertRedirects(response, reverse('wagtailadmin_workflows:task_index'))

        # Check that the task was updated
        task = GroupApprovalTask.objects.get(id=self.task.id)

        # The task name cannot be changed
        self.assertEqual(task.name, "test_task")

        # This request should've added a group to the task
        self.assertEqual(task.groups.count(), 1)
        self.assertTrue(task.groups.filter(id=editors.id).exists())

    def test_permissions(self):
        self.login(user=self.editor)
        response = self.get()
        self.assertEqual(response.status_code, 302)
        full_context = {key: value for context in response.context for key, value in context.items()}
        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')

        self.login(user=self.moderator)
        response = self.get()
        self.assertEqual(response.status_code, 200)


class TestSubmitToWorkflow(TestCase, WagtailTestUtils):
    def setUp(self):
        delete_existing_workflows()
        self.submitter = self.create_user(
            username='submitter',
            email='submitter@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.submitter)
        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)

        self.superuser = self.create_superuser(
            username='superuser',
            email='superuser@email.com',
            password='password',
        )

        self.login(user=self.submitter)

        # Create a page
        root_page = Page.objects.get(id=2)
        self.page = SimplePage(
            title="Hello world!",
            slug='hello-world',
            content="hello",
            live=False,
            has_unpublished_changes=True,
        )
        root_page.add_child(instance=self.page)
        self.page.save_revision()

        self.workflow, self.task_1, self.task_2 = self.create_workflow_and_tasks()

        WorkflowPage.objects.create(workflow=self.workflow, page=self.page)

    def create_workflow_and_tasks(self):
        workflow = Workflow.objects.create(name='test_workflow')
        task_1 = GroupApprovalTask.objects.create(name='test_task_1')
        task_2 = GroupApprovalTask.objects.create(name='test_task_2')
        task_1.groups.set(Group.objects.filter(name='Moderators'))
        task_2.groups.set(Group.objects.filter(name='Moderators'))
        WorkflowTask.objects.create(workflow=workflow, task=task_1, sort_order=1)
        WorkflowTask.objects.create(workflow=workflow, task=task_2, sort_order=2)
        return workflow, task_1, task_2

    def submit(self):
        post_data = {
            'title': str(self.page.title),
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            'action-submit': "True",
        }
        return self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id,)), post_data)

    def test_submit_for_approval_creates_states(self):
        """Test that WorkflowState and TaskState objects are correctly created when a Page is submitted for approval"""

        self.submit()

        workflow_state = self.page.current_workflow_state

        self.assertEqual(type(workflow_state), WorkflowState)
        self.assertEqual(workflow_state.workflow, self.workflow)
        self.assertEqual(workflow_state.status, workflow_state.STATUS_IN_PROGRESS)
        self.assertEqual(workflow_state.requested_by, self.submitter)

        task_state = workflow_state.current_task_state

        self.assertEqual(type(task_state), TaskState)
        self.assertEqual(task_state.task.specific, self.task_1)
        self.assertEqual(task_state.status, task_state.STATUS_IN_PROGRESS)

    def test_submit_for_approval_changes_status_in_header_meta(self):
        edit_url = reverse('wagtailadmin_pages:edit', args=(self.page.id, ))

        response = self.client.get(edit_url)
        self.assertContains(response, 'Draft', count=1)

        # submit for approval
        self.submit()

        response = self.client.get(edit_url)
        workflow_status_url = reverse('wagtailadmin_pages:workflow_status', args=(self.page.id, ))
        self.assertContains(response, workflow_status_url)
        self.assertRegex(response.content.decode('utf-8'), r'Awaiting[\s|\n]+{}'.format(self.page.current_workflow_task.name))
        self.assertNotContains(response, 'Draft')

    @mock.patch.object(EmailMultiAlternatives, 'send', side_effect=IOError('Server down'))
    def test_email_send_error(self, mock_fn):
        logging.disable(logging.CRITICAL)

        response = self.submit()
        logging.disable(logging.NOTSET)

        # An email that fails to send should return a message rather than crash the page
        self.assertEqual(response.status_code, 302)
        response = self.client.get(reverse('wagtailadmin_home'))

    def test_resume_rejected_workflow(self):
        # test that an existing workflow can be resumed by submitting when rejected
        self.workflow.start(self.page, user=self.submitter)
        workflow_state = self.page.current_workflow_state
        workflow_state.current_task_state.approve(user=self.superuser)
        workflow_state.refresh_from_db()
        workflow_state.current_task_state.reject(user=self.superuser)
        workflow_state.refresh_from_db()
        self.assertEqual(workflow_state.current_task_state.task.specific, self.task_2)
        self.assertEqual(workflow_state.status, WorkflowState.STATUS_NEEDS_CHANGES)

        self.submit()
        workflow_state.refresh_from_db()

        # check that the same workflow state's status is now in progress
        self.assertEqual(workflow_state.status, WorkflowState.STATUS_IN_PROGRESS)

        # check that the workflow remains on the rejecting task, rather than resetting
        self.assertEqual(workflow_state.current_task_state.task.specific, self.task_2)

    def test_restart_rejected_workflow(self):
        # test that an existing workflow can be restarted when rejected
        self.workflow.start(self.page, user=self.submitter)
        workflow_state = self.page.current_workflow_state
        workflow_state.current_task_state.approve(user=self.superuser)
        workflow_state.refresh_from_db()
        workflow_state.current_task_state.reject(user=self.superuser)
        workflow_state.refresh_from_db()
        self.assertEqual(workflow_state.current_task_state.task.specific, self.task_2)
        self.assertEqual(workflow_state.status, WorkflowState.STATUS_NEEDS_CHANGES)

        post_data = {
            'title': str(self.page.title),
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            'action-restart-workflow': "True",
        }
        self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id,)), post_data)
        workflow_state.refresh_from_db()

        # check that the same workflow state's status is now cancelled
        self.assertEqual(workflow_state.status, WorkflowState.STATUS_CANCELLED)

        # check that the new workflow has started on the first task
        new_workflow_state = self.page.current_workflow_state
        self.assertEqual(new_workflow_state.status, WorkflowState.STATUS_IN_PROGRESS)
        self.assertEqual(new_workflow_state.current_task_state.task.specific, self.task_1)

    def test_cancel_workflow(self):
        # test that an existing workflow can be cancelled after submission by the submitter
        self.workflow.start(self.page, user=self.submitter)
        workflow_state = self.page.current_workflow_state
        self.assertEqual(workflow_state.current_task_state.task.specific, self.task_1)
        self.assertEqual(workflow_state.status, WorkflowState.STATUS_IN_PROGRESS)
        post_data = {
            'title': str(self.page.title),
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            'action-cancel-workflow': "True",
        }
        self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id,)), post_data)
        workflow_state.refresh_from_db()

        # check that the workflow state's status is now cancelled
        self.assertEqual(workflow_state.status, WorkflowState.STATUS_CANCELLED)
        self.assertEqual(workflow_state.current_task_state.status, TaskState.STATUS_CANCELLED)

    def test_email_headers(self):
        # Submit
        self.submit()

        msg_headers = set(mail.outbox[0].message().items())
        headers = {('Auto-Submitted', 'auto-generated')}
        self.assertTrue(headers.issubset(msg_headers), msg='Message is missing the Auto-Submitted header.',)


@freeze_time("2020-03-31 12:00:00")
class TestApproveRejectWorkflow(TestCase, WagtailTestUtils):
    def setUp(self):
        delete_existing_workflows()
        self.submitter = self.create_user(
            username='submitter',
            first_name='Sebastian',
            last_name='Mitter',
            email='submitter@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.submitter)
        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)

        self.superuser = self.create_superuser(
            username='superuser',
            email='superuser@email.com',
            password='password',
        )

        self.login(user=self.submitter)

        # Create a page
        root_page = Page.objects.get(id=2)
        self.page = SimplePage(
            title="Hello world!",
            slug='hello-world',
            content="hello",
            live=False,
            has_unpublished_changes=True,
        )
        root_page.add_child(instance=self.page)

        self.workflow, self.task_1 = self.create_workflow_and_tasks()

        WorkflowPage.objects.create(workflow=self.workflow, page=self.page)

        self.submit()

        self.login(user=self.moderator)

    def create_workflow_and_tasks(self):
        workflow = Workflow.objects.create(name='test_workflow')
        task_1 = GroupApprovalTask.objects.create(name='test_task_1')
        task_1.groups.set(Group.objects.filter(name='Moderators'))
        WorkflowTask.objects.create(workflow=workflow, task=task_1, sort_order=1)
        return workflow, task_1

    def submit(self):
        post_data = {
            'title': str(self.page.title),
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            'action-submit': "True",
        }
        return self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id,)), post_data)

    @override_settings(WAGTAIL_FINISH_WORKFLOW_ACTION='')
    def test_approve_task_and_workflow(self):
        """
        This posts to the approve task view and checks that the page was approved and published
        """
        # Unset WAGTAIL_FINISH_WORKFLOW_ACTION - default action should be to publish
        del settings.WAGTAIL_FINISH_WORKFLOW_ACTION
        # Connect a mock signal handler to page_published signal
        mock_handler = mock.MagicMock()
        page_published.connect(mock_handler)

        # Post
        self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)), {'comment': 'my comment'})

        # Check that the workflow was approved

        workflow_state = WorkflowState.objects.get(page=self.page, requested_by=self.submitter)

        self.assertEqual(workflow_state.status, workflow_state.STATUS_APPROVED)

        # Check that the task was approved

        task_state = workflow_state.current_task_state

        self.assertEqual(task_state.status, task_state.STATUS_APPROVED)

        # Check that the comment was added to the task state correctly

        self.assertEqual(task_state.comment, 'my comment')

        page = Page.objects.get(id=self.page.id)
        # Page must be live
        self.assertTrue(page.live, "Approving moderation failed to set live=True")
        # Page should now have no unpublished changes
        self.assertFalse(
            page.has_unpublished_changes,
            "Approving moderation failed to set has_unpublished_changes=False"
        )

        # Check that the page_published signal was fired
        self.assertEqual(mock_handler.call_count, 1)
        mock_call = mock_handler.mock_calls[0][2]

        self.assertEqual(mock_call['sender'], self.page.specific_class)
        self.assertEqual(mock_call['instance'], self.page)
        self.assertIsInstance(mock_call['instance'], self.page.specific_class)

    def test_workflow_action_get(self):
        """
        This tests that a GET request to the workflow action view (for the approve action) returns a modal with a form for extra data entry:
        adding a comment
        """
        response = self.client.get(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)))
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/pages/workflow_action_modal.html')
        html = json.loads(response.content)['html']
        self.assertTagInHTML('<form action="' + reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)) + '" method="POST" novalidate>', html)
        self.assertIn('Comment', html)

    def test_workflow_action_view_bad_page_id(self):
        """
        This tests that the workflow action view handles invalid page ids correctly
        """
        # Post
        response = self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(127777777777, 'approve', self.page.current_workflow_task_state.id)))

        # Check that the user received a 404 response
        self.assertEqual(response.status_code, 404)

    def test_workflow_action_view_not_in_group(self):
        """
        This tests that the workflow action view for a GroupApprovalTask won't allow approval from a user not in the
        specified group/a superuser
        """
        # Remove privileges from user
        self.login(user=self.submitter)

        # Post
        response = self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)))
        # Check that the user received a permission denied response
        self.assertRedirects(response, '/admin/')

    def test_edit_view_workflow_cancellation_not_in_group(self):
        """
        This tests that the page edit view for a GroupApprovalTask, locked to a user not in the
        specified group/a superuser, still allows the submitter to cancel workflows
        """
        self.login(user=self.submitter)

        # Post
        response = self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id, )), {'action-cancel-workflow': 'True'})

        # Check that the user received a 200 response
        self.assertEqual(response.status_code, 200)

        # Check that the workflow state was marked as cancelled
        workflow_state = WorkflowState.objects.get(page=self.page, requested_by=self.submitter)
        self.assertEqual(workflow_state.status, WorkflowState.STATUS_CANCELLED)

    def test_reject_task_and_workflow(self):
        """
        This posts to the reject task view and checks that the page was rejected and not published
        """
        # Post
        self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'reject', self.page.current_workflow_task_state.id)))

        # Check that the workflow was marked as needing changes

        workflow_state = WorkflowState.objects.get(page=self.page, requested_by=self.submitter)

        self.assertEqual(workflow_state.status, workflow_state.STATUS_NEEDS_CHANGES)

        # Check that the task was rejected

        task_state = workflow_state.current_task_state

        self.assertEqual(task_state.status, task_state.STATUS_REJECTED)

        page = Page.objects.get(id=self.page.id)
        # Page must not be live
        self.assertFalse(page.live)

    def test_workflow_action_view_rejection_not_in_group(self):
        """
        This tests that the workflow action view for a GroupApprovalTask won't allow rejection from a user not in the
        specified group/a superuser
        """
        # Remove privileges from user
        self.login(user=self.submitter)

        # Post
        response = self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'reject', self.page.current_workflow_task_state.id)))

        # Check that the user received a permission denied response
        self.assertRedirects(response, '/admin/')

    def test_collect_workflow_action_data_get(self):
        """
        This tests that a GET request to the collect_workflow_action_data view (for the approve action) returns a modal with a form for extra data entry:
        adding a comment
        """
        response = self.client.get(reverse('wagtailadmin_pages:collect_workflow_action_data', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)))
        self.assertEqual(response.status_code, 200)
        self.assertTemplateUsed(response, 'wagtailadmin/pages/workflow_action_modal.html')
        html = json.loads(response.content)['html']
        self.assertTagInHTML('<form action="' + reverse('wagtailadmin_pages:collect_workflow_action_data', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)) + '" method="POST" novalidate>', html)
        self.assertIn('Comment', html)

    def test_collect_workflow_action_data_post(self):
        """
        This tests that a POST request to the collect_workflow_action_data view (for the approve action) returns a modal response with the validated data
        """
        response = self.client.post(
            reverse('wagtailadmin_pages:collect_workflow_action_data', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)),
            {'comment': "This is my comment"}
        )
        self.assertEqual(response.status_code, 200)
        response_json = json.loads(response.content)
        self.assertEqual(response_json['step'], 'success')
        self.assertEqual(response_json['cleaned_data'], {'comment': "This is my comment"})

    def test_workflow_action_via_edit_view(self):
        """
        Posting to the 'edit' view with 'action-workflow-action' set should perform the given workflow action in addition to updating page content
        """
        # Post
        self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id,)), {
            'title': "This title was edited while approving",
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            'action-workflow-action': "True",
            'workflow-action-name': 'approve',
            'workflow-action-extra-data': '{"comment": "my comment"}'
        })

        # Check that the workflow was approved

        workflow_state = WorkflowState.objects.get(page=self.page, requested_by=self.submitter)

        self.assertEqual(workflow_state.status, workflow_state.STATUS_APPROVED)

        # Check that the task was approved

        task_state = workflow_state.current_task_state

        self.assertEqual(task_state.status, task_state.STATUS_APPROVED)

        # Check that the comment was added to the task state correctly

        self.assertEqual(task_state.comment, 'my comment')

        # Check that page edits made at the same time as the action have been saved
        page = Page.objects.get(id=self.page.id)
        self.assertEqual(page.get_latest_revision_as_page().title, "This title was edited while approving")

    def test_workflow_report(self):
        response = self.client.get(reverse('wagtailadmin_reports:workflow'))
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Hello world!")
        self.assertContains(response, "test_workflow")
        self.assertContains(response, "Sebastian Mitter")
        self.assertContains(response, "March 31, 2020")

        response = self.client.get(reverse('wagtailadmin_reports:workflow_tasks'))
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Hello world!")

    def test_workflow_report_filtered(self):
        # the moderator can review the task, so the workflow state should show up even when reports are filtered by reviewable
        response = self.client.get(reverse('wagtailadmin_reports:workflow'), {'reviewable': 'true'})
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Hello world!")
        self.assertContains(response, "test_workflow")
        self.assertContains(response, "Sebastian Mitter")
        self.assertContains(response, "March 31, 2020")

        response = self.client.get(reverse('wagtailadmin_reports:workflow_tasks'), {'reviewable': 'true'})
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Hello world!")

        # the submitter cannot review the task, so the workflow state shouldn't show up when reports are filtered by reviewable
        self.login(self.submitter)
        response = self.client.get(reverse('wagtailadmin_reports:workflow'), {'reviewable': 'true'})
        self.assertEqual(response.status_code, 200)
        self.assertNotContains(response, "Hello world!")
        self.assertNotContains(response, "Sebastian Mitter")
        self.assertNotContains(response, "March 31, 2020")

        response = self.client.get(reverse('wagtailadmin_reports:workflow_tasks'), {'reviewable': 'true'})
        self.assertEqual(response.status_code, 200)
        self.assertNotContains(response, "Hello world!")


class TestNotificationPreferences(TestCase, WagtailTestUtils):
    def setUp(self):
        delete_existing_workflows()
        self.submitter = self.create_user(
            username='submitter',
            email='submitter@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.submitter)
        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        self.moderator2 = self.create_user(
            username='moderator2',
            email='moderator2@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.user_set.add(self.moderator2)

        self.superuser = self.create_superuser(
            username='superuser',
            email='superuser@email.com',
            password='password',
        )

        self.superuser_profile = UserProfile.get_for_user(self.superuser)
        self.moderator2_profile = UserProfile.get_for_user(self.moderator2)
        self.submitter_profile = UserProfile.get_for_user(self.submitter)

        # Create a page
        root_page = Page.objects.get(id=2)
        self.page = SimplePage(
            title="Hello world!",
            slug='hello-world',
            content="hello",
            live=False,
            has_unpublished_changes=True,
        )
        root_page.add_child(instance=self.page)

        self.workflow, self.task_1 = self.create_workflow_and_tasks()

        WorkflowPage.objects.create(workflow=self.workflow, page=self.page)

    def create_workflow_and_tasks(self):
        workflow = Workflow.objects.create(name='test_workflow')
        task_1 = GroupApprovalTask.objects.create(name='test_task_1')
        task_1.groups.set(Group.objects.filter(name='Moderators'))
        WorkflowTask.objects.create(workflow=workflow, task=task_1, sort_order=1)
        return workflow, task_1

    def submit(self):
        post_data = {
            'title': str(self.page.title),
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            'action-submit': "True",
        }
        return self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id,)), post_data)

    def approve(self):
        return self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)))

    def reject(self):
        return self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'reject', self.page.current_workflow_task_state.id)))

    def test_vanilla_profile(self):
        # Check that the vanilla profile has rejected notifications on
        self.assertEqual(self.submitter_profile.rejected_notifications, True)

        # Check that the vanilla profile has approved notifications on
        self.assertEqual(self.submitter_profile.approved_notifications, True)

    @override_settings(WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS=True)
    def test_submitted_email_notifications_sent(self):
        """Test that 'submitted' notifications for WorkflowState and TaskState are both sent correctly"""
        self.login(self.submitter)
        self.submit()

        self.assertEqual(len(mail.outbox), 4)

        task_submission_emails = [email for email in mail.outbox if "task" in email.subject]
        task_submission_emailed_addresses = [address for email in task_submission_emails for address in email.to]
        workflow_submission_emails = [email for email in mail.outbox if "workflow" in email.subject]
        workflow_submission_emailed_addresses = [address for email in workflow_submission_emails for address in email.to]

        self.assertEqual(len(task_submission_emails), 3)
        # the moderator is in the Group assigned to the GroupApproval task, so should get an email
        self.assertIn(self.moderator.email, task_submission_emailed_addresses)
        self.assertIn(self.moderator2.email, task_submission_emailed_addresses)
        # with `WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS`, the superuser should get a task email
        self.assertIn(self.superuser.email, task_submission_emailed_addresses)
        # the submitter triggered this workflow update, so should not get an email
        self.assertNotIn(self.submitter.email, task_submission_emailed_addresses)

        self.assertEqual(len(workflow_submission_emails), 1)
        # the moderator should not get a workflow email
        self.assertNotIn(self.moderator.email, workflow_submission_emailed_addresses)
        self.assertNotIn(self.moderator2.email, workflow_submission_emailed_addresses)
        # with `WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS`, the superuser should get a workflow email
        self.assertIn(self.superuser.email, workflow_submission_emailed_addresses)
        # as the submitter was the triggering user, the submitter should not get an email notification
        self.assertNotIn(self.submitter.email, workflow_submission_emailed_addresses)

    @override_settings(WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS=False)
    def test_submitted_email_notifications_superuser_settings(self):
        """Test that 'submitted' notifications for WorkflowState and TaskState are not sent to superusers if
        `WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS=False`"""
        self.login(self.submitter)
        self.submit()

        task_submission_emails = [email for email in mail.outbox if "task" in email.subject]
        task_submission_emailed_addresses = [address for email in task_submission_emails for address in email.to]
        workflow_submission_emails = [email for email in mail.outbox if "workflow" in email.subject]
        workflow_submission_emailed_addresses = [address for email in workflow_submission_emails for address in email.to]

        # with `WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS` off, the superuser should not get a task email
        self.assertNotIn(self.superuser.email, task_submission_emailed_addresses)

        # with `WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS` off, the superuser should not get a workflow email
        self.assertNotIn(self.superuser.email, workflow_submission_emailed_addresses)

    @override_settings(WAGTAILADMIN_NOTIFICATION_INCLUDE_SUPERUSERS=True)
    def test_submit_notification_preferences_respected(self):
        # moderator2 doesn't want emails
        self.moderator2_profile.submitted_notifications = False
        self.moderator2_profile.save()

        # superuser doesn't want emails
        self.superuser_profile.submitted_notifications = False
        self.superuser_profile.save()

        # Submit
        self.login(self.submitter)
        self.submit()

        # Check that only one moderator got a task submitted email
        workflow_submission_emails = [email for email in mail.outbox if "workflow" in email.subject]
        workflow_submission_emailed_addresses = [address for email in workflow_submission_emails for address in email.to]
        task_submission_emails = [email for email in mail.outbox if "task" in email.subject]
        task_submission_emailed_addresses = [address for email in task_submission_emails for address in email.to]
        self.assertNotIn(self.moderator2.email, task_submission_emailed_addresses)

        # Check that the superuser didn't receive a workflow or task email
        self.assertNotIn(self.superuser.email, task_submission_emailed_addresses)
        self.assertNotIn(self.superuser.email, workflow_submission_emailed_addresses)

    def test_approved_notifications(self):
        self.login(self.submitter)
        self.submit()
        # Approve
        self.login(self.moderator)
        self.approve()

        # Submitter must receive a workflow approved email
        workflow_approved_emails = [email for email in mail.outbox if ("workflow" in email.subject and "approved" in email.subject)]
        self.assertEqual(len(workflow_approved_emails), 1)
        self.assertIn(self.submitter.email, workflow_approved_emails[0].to)

    def test_approved_notifications_preferences_respected(self):
        # Submitter doesn't want 'approved' emails
        self.submitter_profile.approved_notifications = False
        self.submitter_profile.save()

        self.login(self.submitter)
        self.submit()
        # Approve
        self.login(self.moderator)
        self.approve()

        # Submitter must not receive a workflow approved email, so there should be no emails in workflow_approved_emails
        workflow_approved_emails = [email for email in mail.outbox if ("workflow" in email.subject and "approved" in email.subject)]
        self.assertEqual(len(workflow_approved_emails), 0)

    def test_rejected_notifications(self):
        self.login(self.submitter)
        self.submit()
        # Reject
        self.login(self.moderator)
        self.reject()

        # Submitter must receive a workflow rejected email
        workflow_rejected_emails = [email for email in mail.outbox if ("workflow" in email.subject and "rejected" in email.subject)]
        self.assertEqual(len(workflow_rejected_emails), 1)
        self.assertIn(self.submitter.email, workflow_rejected_emails[0].to)

    def test_rejected_notification_preferences_respected(self):
        # Submitter doesn't want 'rejected' emails
        self.submitter_profile.rejected_notifications = False
        self.submitter_profile.save()

        self.login(self.submitter)
        self.submit()
        # Reject
        self.login(self.moderator)
        self.reject()

        # Submitter must not receive a workflow rejected email
        workflow_rejected_emails = [email for email in mail.outbox if ("workflow" in email.subject and "rejected" in email.subject)]
        self.assertEqual(len(workflow_rejected_emails), 0)


class TestDisableViews(TestCase, WagtailTestUtils):
    def setUp(self):
        delete_existing_workflows()
        self.submitter = self.create_user(
            username='submitter',
            email='submitter@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.submitter)
        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        self.moderator2 = self.create_user(
            username='moderator2',
            email='moderator2@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)
        moderators.user_set.add(self.moderator2)

        self.superuser = self.create_superuser(
            username='superuser',
            email='superuser@email.com',
            password='password',
        )

        # Create a page
        root_page = Page.objects.get(id=2)
        self.page = SimplePage(
            title="Hello world!",
            slug='hello-world',
            content="hello",
            live=False,
            has_unpublished_changes=True,
        )
        root_page.add_child(instance=self.page)

        self.workflow, self.task_1, self.task_2 = self.create_workflow_and_tasks()

        WorkflowPage.objects.create(workflow=self.workflow, page=self.page)

    def create_workflow_and_tasks(self):
        workflow = Workflow.objects.create(name='test_workflow')
        task_1 = GroupApprovalTask.objects.create(name='test_task_1')
        task_1.groups.set(Group.objects.filter(name='Moderators'))
        task_2 = GroupApprovalTask.objects.create(name='test_task_2')
        task_2.groups.set(Group.objects.filter(name='Moderators'))
        WorkflowTask.objects.create(workflow=workflow, task=task_1, sort_order=1)
        WorkflowTask.objects.create(workflow=workflow, task=task_2, sort_order=2)
        return workflow, task_1, task_2

    def submit(self):
        post_data = {
            'title': str(self.page.title),
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            'action-submit': "True",
        }
        return self.client.post(reverse('wagtailadmin_pages:edit', args=(self.page.id,)), post_data)

    def approve(self):
        return self.client.post(reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, 'approve', self.page.current_workflow_task_state.id)))

    def test_disable_workflow(self):
        """Test that deactivating a workflow sets it to inactive and cancels in progress states"""
        self.login(self.submitter)
        self.submit()
        self.login(self.superuser)
        self.approve()

        response = self.client.post(reverse('wagtailadmin_workflows:disable', args=(self.workflow.pk,)))
        self.assertEqual(response.status_code, 302)
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.active, False)
        states = WorkflowState.objects.filter(page=self.page, workflow=self.workflow)
        self.assertEqual(states.filter(status=WorkflowState.STATUS_IN_PROGRESS).count(), 0)
        self.assertEqual(states.filter(status=WorkflowState.STATUS_CANCELLED).count(), 1)

        self.assertEqual(TaskState.objects.filter(workflow_state__workflow=self.workflow, status=TaskState.STATUS_IN_PROGRESS).count(), 0)

    def test_disable_task(self):
        """Test that deactivating a task sets it to inactive and cancels in progress states"""
        self.login(self.submitter)
        self.submit()
        self.login(self.superuser)

        response = self.client.post(reverse('wagtailadmin_workflows:disable_task', args=(self.task_1.pk,)))
        self.assertEqual(response.status_code, 302)
        self.task_1.refresh_from_db()
        self.assertEqual(self.task_1.active, False)
        states = TaskState.objects.filter(workflow_state__page=self.page, task=self.task_1.task_ptr)
        self.assertEqual(states.filter(status=TaskState.STATUS_IN_PROGRESS).count(), 0)
        self.assertEqual(states.filter(status=TaskState.STATUS_CANCELLED).count(), 1)

        # Check that the page's WorkflowState has moved on to the next active task
        self.assertEqual(self.page.current_workflow_state.current_task_state.task.specific, self.task_2)

    def test_enable_workflow(self):
        self.login(self.superuser)
        self.workflow.active = False
        self.workflow.save()

        response = self.client.post(reverse('wagtailadmin_workflows:enable', args=(self.workflow.pk,)))
        self.assertEqual(response.status_code, 302)
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.active, True)

    def test_enable_task(self):
        self.login(self.superuser)
        self.task_1.active = False
        self.task_1.save()

        response = self.client.post(reverse('wagtailadmin_workflows:enable_task', args=(self.task_1.pk,)))
        self.assertEqual(response.status_code, 302)
        self.task_1.refresh_from_db()
        self.assertEqual(self.task_1.active, True)


class TestTaskChooserView(TestCase, WagtailTestUtils):
    def setUp(self):
        self.login()

    def test_get(self):
        response = self.client.get(reverse('wagtailadmin_workflows:task_chooser'))

        self.assertEqual(response.status_code, 200)

        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/chooser.html")

        # Check that the "select task type" view was shown in the "new" tab
        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/includes/select_task_type.html")
        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/includes/results.html")
        self.assertTemplateNotUsed(response, "wagtailadmin/workflows/task_chooser/includes/create_form.html")
        self.assertFalse(response.context['searchform'].is_searching())

    def test_search(self):
        response = self.client.get(reverse('wagtailadmin_workflows:task_chooser') + '?q=foo')

        self.assertEqual(response.status_code, 200)

        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/includes/results.html")
        self.assertTemplateNotUsed(response, "wagtailadmin/workflows/task_chooser/chooser.html")
        self.assertTrue(response.context['searchform'].is_searching())

    def test_pagination(self):
        response = self.client.get(reverse('wagtailadmin_workflows:task_chooser') + '?p=2')

        self.assertEqual(response.status_code, 200)

        # When pagination is used, only the results template should be rendered
        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/includes/results.html")
        self.assertTemplateNotUsed(response, "wagtailadmin/workflows/task_chooser/chooser.html")
        self.assertFalse(response.context['searchform'].is_searching())

    def test_get_with_create_model_selected(self):
        response = self.client.get(reverse('wagtailadmin_workflows:task_chooser') + '?create_model=wagtailcore.GroupApprovalTask')

        self.assertEqual(response.status_code, 200)

        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/chooser.html")

        # Check that the "create" view was shown in the "new" tab
        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/includes/create_form.html")
        self.assertTemplateNotUsed(response, "wagtailadmin/workflows/task_chooser/includes/select_task_type.html")

    def test_get_with_non_task_create_model_selected(self):
        response = self.client.get(reverse('wagtailadmin_workflows:task_chooser') + '?create_model=wagtailcore.Page')

        self.assertEqual(response.status_code, 404)

    def test_get_with_base_task_create_model_selected(self):
        # Task is technically a subclass of itself so we need an extra test for it
        response = self.client.get(reverse('wagtailadmin_workflows:task_chooser') + '?create_model=wagtailcore.Task')

        self.assertEqual(response.status_code, 404)

    @mock.patch('wagtail.admin.views.workflows.get_task_types')
    def test_get_with_single_task_model(self, get_task_types):
        # When a single task type exists there's no need to specify create_model
        get_task_types.return_value = [GroupApprovalTask]

        response = self.client.get(reverse('wagtailadmin_workflows:task_chooser'))

        self.assertEqual(response.status_code, 200)

        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/chooser.html")

        # Check that the "create" view was shown in the "new" tab
        self.assertTemplateUsed(response, "wagtailadmin/workflows/task_chooser/includes/create_form.html")
        self.assertTemplateNotUsed(response, "wagtailadmin/workflows/task_chooser/includes/select_task_type.html")

    # POST requests are for creating new tasks

    def get_post_data(self):
        return {
            'create-task-name': 'Editor approval task',
            'create-task-groups': [str(Group.objects.get(name="Editors").id)],
        }

    def test_post_with_create_model_selected(self):
        response = self.client.post(reverse('wagtailadmin_workflows:task_chooser') + '?create_model=wagtailcore.GroupApprovalTask', self.get_post_data())

        self.assertEqual(response.status_code, 201)

        # Check that the task was created
        task = Task.objects.get(name="Editor approval task", active=True)

        # Check the response JSON
        self.assertEqual(response.json(), {
            "step": "task_chosen",
            "result": {
                "id": task.id,
                "name": "Editor approval task",
                "edit_url": reverse('wagtailadmin_workflows:edit_task', args=[task.id])
            }
        })

    @mock.patch('wagtail.admin.views.workflows.get_task_types')
    def test_post_with_single_task_model(self, get_task_types):
        # When a single task type exists there's no need to specify create_model
        get_task_types.return_value = [GroupApprovalTask]

        response = self.client.post(reverse('wagtailadmin_workflows:task_chooser'), self.get_post_data())

        self.assertEqual(response.status_code, 201)

        # Check that the task was created
        task = Task.objects.get(name="Editor approval task", active=True)

        # Check the response JSON
        self.assertEqual(response.json(), {
            "step": "task_chosen",
            "result": {
                "id": task.id,
                "name": "Editor approval task",
                "edit_url": reverse('wagtailadmin_workflows:edit_task', args=[task.id])
            }
        })

    def test_post_without_create_model_selected(self):
        response = self.client.post(reverse('wagtailadmin_workflows:task_chooser'), self.get_post_data())

        self.assertEqual(response.status_code, 400)

        # Check that the task wasn't created
        self.assertFalse(Task.objects.filter(name="Editor approval task", active=True).exists())

    def test_post_with_non_task_create_model_selected(self):
        response = self.client.post(reverse('wagtailadmin_workflows:task_chooser') + '?create_model=wagtailcore.Page', self.get_post_data())

        self.assertEqual(response.status_code, 404)

        # Check that the task wasn't created
        self.assertFalse(Task.objects.filter(name="Editor approval task", active=True).exists())

    def test_post_with_base_task_create_model_selected(self):
        # Task is technically a subclass of itself so we need an extra test for it
        response = self.client.post(reverse('wagtailadmin_workflows:task_chooser') + '?create_model=wagtailcore.Task', self.get_post_data())

        self.assertEqual(response.status_code, 404)

        # Check that the task wasn't created
        self.assertFalse(Task.objects.filter(name="Editor approval task", active=True).exists())


class TestTaskChooserChosenView(TestCase, WagtailTestUtils):
    def setUp(self):
        delete_existing_workflows()
        self.login()
        self.task = SimpleTask.objects.create(name="test_task")

    def test_get(self):
        response = self.client.get(reverse('wagtailadmin_workflows:task_chosen', args=[self.task.id]))

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response['Content-Type'], 'application/json')
        self.assertEqual(response.json(), {
            'result': {
                'edit_url': reverse('wagtailadmin_workflows:edit_task', args=[self.task.id]),
                'id': self.task.id,
                'name': 'test_task'
            },
            'step': 'task_chosen'
        })


class TestWorkflowUsageView(TestCase, WagtailTestUtils):
    def setUp(self):
        self.login()
        self.workflow = Workflow.objects.get()

        self.root_page = Page.objects.get(depth=1)
        self.home_page = Page.objects.get(depth=2)

        self.child_page_with_another_workflow = self.home_page.add_child(instance=SimplePage(title="Another page", content="I'm another page"))
        self.another_workflow = Workflow.objects.create(name="Another workflow")
        self.another_workflow.workflow_pages.create(page=self.child_page_with_another_workflow)

    def test_get(self):
        response = self.client.get(reverse('wagtailadmin_workflows:usage', args=[self.workflow.id]))

        self.assertEqual(response.status_code, 200)

        object_set = set(page.id for page in response.context['used_by'].object_list)
        self.assertIn(self.root_page.id, object_set)
        self.assertIn(self.home_page.id, object_set)
        self.assertNotIn(self.child_page_with_another_workflow.id, object_set)


@freeze_time("2020-06-01 12:00:00")
class TestWorkflowStatus(TestCase, WagtailTestUtils):
    def setUp(self):
        delete_existing_workflows()
        self.submitter = self.create_user(
            username='submitter',
            email='submitter@email.com',
            password='password',
        )
        editors = Group.objects.get(name='Editors')
        editors.user_set.add(self.submitter)
        self.moderator = self.create_user(
            username='moderator',
            email='moderator@email.com',
            password='password',
        )
        moderators = Group.objects.get(name='Moderators')
        moderators.user_set.add(self.moderator)

        self.superuser = self.create_superuser(
            username='superuser',
            email='superuser@email.com',
            password='password',
        )

        self.login(self.superuser)

        # Create a page
        root_page = Page.objects.get(id=2)
        self.page = SimplePage(
            title="Hello world!",
            slug='hello-world',
            content="hello",
            live=False,
            has_unpublished_changes=True,
        )
        root_page.add_child(instance=self.page)

        self.workflow, self.task_1, self.task_2 = self.create_workflow_and_tasks()

        WorkflowPage.objects.create(workflow=self.workflow, page=self.page)

        self.edit_url = reverse('wagtailadmin_pages:edit', args=(self.page.id, ))

    def create_workflow_and_tasks(self):
        workflow = Workflow.objects.create(name='test_workflow')
        task_1 = GroupApprovalTask.objects.create(name='test_task_1')
        task_1.groups.set(Group.objects.filter(name='Moderators'))
        WorkflowTask.objects.create(workflow=workflow, task=task_1, sort_order=1)

        task_2 = GroupApprovalTask.objects.create(name='test_task_2')
        task_2.groups.set(Group.objects.filter(name='Editors'))
        WorkflowTask.objects.create(workflow=workflow, task=task_2, sort_order=2)
        return workflow, task_1, task_2

    def submit(self, action='action-submit'):
        post_data = {
            'title': str(self.page.title),
            'slug': str(self.page.slug),
            'content': str(self.page.content),
            action: "True",
        }
        return self.client.post(self.edit_url, post_data)

    def workflow_action(self, action):
        post_data = {
            'action': action,
            'comment': 'good work' if action == 'approve' else 'needs some changes',
            'next': self.edit_url
        }
        return self.client.post(
            reverse('wagtailadmin_pages:workflow_action', args=(self.page.id, action, self.page.current_workflow_task_state.id)),
            post_data,
            follow=True
        )

    def test_workflow_status_modal(self):
        workflow_status_url = reverse('wagtailadmin_pages:workflow_status', args=(self.page.id, ))

        # The page workflow status view should return permission denied when the page is but a draft
        response = self.client.get(workflow_status_url)
        self.assertRedirects(response, '/admin/')

        # Submit for moderation
        self.submit()

        response = self.client.get(workflow_status_url)
        self.assertEqual(response.status_code, 200)
        html = response.json().get('html')
        self.assertIn(self.task_1.name, html)
        self.assertIn('{}: In progress'.format(self.task_1.name), html)
        self.assertIn(self.task_2.name, html)
        self.assertIn('{}: Not started'.format(self.task_2.name), html)
        self.assertIn(reverse('wagtailadmin_pages:history', args=(self.page.id, )), html)

        self.assertTemplateUsed(response, 'wagtailadmin/workflows/workflow_status.html')

    def test_status_through_workflow_cycle(self):
        self.login(self.superuser)
        response = self.client.get(self.edit_url)
        self.assertContains(response, 'Draft', 1)

        self.page.save_revision()
        response = self.client.get(self.edit_url)
        self.assertContains(response, 'Draft saved', 1)

        self.submit()
        response = self.client.get(self.edit_url)
        self.assertRegex(response.content.decode('utf-8'), r'Awaiting[\s|\n]+{}'.format(self.task_1.name))

        response = self.workflow_action('approve')
        self.assertRegex(response.content.decode('utf-8'), r'Awaiting[\s|\n]+{}'.format(self.task_2.name))

        response = self.workflow_action('reject')
        self.assertContains(response, 'Changes requested')

        # resubmit
        self.submit()
        response = self.client.get(self.edit_url)
        self.assertRegex(response.content.decode('utf-8'), r'Awaiting[\s|\n]+{}'.format(self.task_2.name))

        response = self.workflow_action('approve')
        self.assertContains(response, 'Published')

    def test_status_after_cancel(self):
        # start workflow, then cancel
        self.submit()
        self.submit('action-cancel-workflow')
        response = self.client.get(self.edit_url)
        self.assertContains(response, 'Draft saved')

    def test_status_after_restart(self):
        self.submit()
        response = self.workflow_action('approve')
        self.assertRegex(response.content.decode('utf-8'), r'Awaiting[\s|\n]+{}'.format(self.task_2.name))
        self.workflow_action('reject')
        self.submit('action-restart-workflow')
        response = self.client.get(self.edit_url)
        self.assertRegex(response.content.decode('utf-8'), r'Awaiting[\s|\n]+{}'.format(self.task_1.name))

    def test_workflow_status_modal_task_comments(self):
        workflow_status_url = reverse('wagtailadmin_pages:workflow_status', args=(self.page.id,))

        self.submit()
        self.workflow_action('reject')

        response = self.client.get(workflow_status_url)
        self.assertIn('needs some changes', response.json().get('html'))

        self.submit()
        self.workflow_action('approve')
        response = self.client.get(workflow_status_url)
        self.assertIn('good work', response.json().get('html'))

    def test_workflow_edit_locked_message(self):
        self.submit()
        self.login(self.submitter)
        response = self.client.get(self.edit_url)

        needle = "This page is awaiting <b>\'test_task_1\'</b> in the <b>\'test_workflow\'</b> workflow. Only reviewers for this task can edit the page."
        self.assertTagInHTML(needle, str(response.content), count=1)

        self.login(self.moderator)
        response = self.client.get(self.edit_url)
        self.assertNotInHTML(needle, str(response.content))
