diff --git a/process/tests/test_ops.py b/process/tests/test_ops.py index 8f25d23150fe6b0b36ba864291a0ea69520537dc..764207aabd14a26cbaff11b5b8ccb8f8d485c3c8 100644 --- a/process/tests/test_ops.py +++ b/process/tests/test_ops.py @@ -177,23 +177,6 @@ class TestOps(ProcessFixtureMixin, TestCase): o = pops.ProcessStatementRemove(audit_author=self.persons.fd, statement=st) self.check_op(o, check_contents) - def test_process_approve_rt(self): - def check_contents(o): - self.assertEqual(o.audit_author, self.persons.fd) - self.assertEqual(o.audit_notes, "Process approved") - self.assertIsInstance(o.audit_time, datetime.datetime) - self.assertEqual(o.process, self.processes.app) - self.assertEqual(o.rt_id, "ticket/new") - self.assertEqual(o.rt_queue, "Keyring") - self.assertEqual(o.rt_requestor, "da-manager@debian.org") - self.assertEqual(o.rt_subject, "Dc to become Debian Developer, uploading") - self.assertEqual(o.rt_cc, "dc@example.org, archive-{}@nm.debian.org, da-manager@debian.org".format(self.processes.app.pk)) - self.assertIsNone(o.rt_text) - - o = pops.ProcessApproveRT(audit_author=self.persons.fd, process=self.processes.app) - self.check_op(o, check_contents) - - class TestProcessClose(ProcessFixtureMixin, TestCase): @classmethod diff --git a/process/tests/test_process_approve.py b/process/tests/test_process_approve.py new file mode 100644 index 0000000000000000000000000000000000000000..e542a23b59472df15a9b00529b6ded3887c5a92d --- /dev/null +++ b/process/tests/test_process_approve.py @@ -0,0 +1,217 @@ +from django.test import TestCase +from django.utils.timezone import now +from django.urls import reverse +from django.core import mail +from unittest.mock import patch +from backend.unittest import PersonFixtureMixin +from backend import const +import process.models as pmodels +import process.views as pviews +from process.unittest import ProcessFixtureMixin +from process import ops as pops + +class TestApproveOp(ProcessFixtureMixin, TestCase): + @classmethod + def __add_extra_tests__(cls): + cls._add_method(cls._test_process_approve_dsa, "dc", const.STATUS_DC_GA) + cls._add_method(cls._test_process_approve_dsa, "dm", const.STATUS_DM_GA) + + cls._add_method(cls._test_process_approve_fd_keyring, "dc", const.STATUS_DM) + cls._add_method(cls._test_process_approve_dam_keyring, "dc", const.STATUS_DD_NU) + cls._add_method(cls._test_process_approve_dam_keyring, "dc", const.STATUS_DD_U) + cls._add_method(cls._test_process_approve_dam_keyring, "dc_ga", const.STATUS_DD_NU) + cls._add_method(cls._test_process_approve_dam_keyring, "dc_ga", const.STATUS_DD_U) + cls._add_method(cls._test_process_approve_dam_keyring, "dm", const.STATUS_DD_NU) + cls._add_method(cls._test_process_approve_dam_keyring, "dm", const.STATUS_DD_U) + cls._add_method(cls._test_process_approve_dam_keyring, "dm_ga", const.STATUS_DD_NU) + cls._add_method(cls._test_process_approve_dam_keyring, "dm_ga", const.STATUS_DD_U) + cls._add_method(cls._test_process_approve_dam_keyring, "dd_nu", const.STATUS_DD_U) + + def assertApproveCommon(self, o): + self.assertEqual(o.audit_author, self.persons.fd) + self.assertEqual(o.audit_notes, "Process approved") + self.assertEqual(o.rt_id, "ticket/new") + self.assertIsNone(o.rt_text) + + def _test_process_approve_dsa(self, person, applying_for): + person = self.persons[person] + # dc to have a guest account + process = pmodels.Process.objects.create(person=person, applying_for=applying_for) + o = pops.ProcessApproveRT(audit_author=self.persons.fd, process=process) + @self.assertOperationSerializes(o) + def _(o): + self.assertApproveCommon(o) + self.assertEqual(o.process, process) + self.assertEqual(o.rt_queue, "DSA - Incoming") + self.assertEqual(o.rt_requestor, "da-manager@debian.org") + self.assertEqual(o.rt_subject, "Guest account on porter machines for " + person.fullname) + self.assertEqual(o.rt_cc, "{}, archive-{}@nm.debian.org, da-manager@debian.org".format(person.email, process.pk)) + + def _test_process_approve_fd_keyring(self, person, applying_for): + person = self.persons[person] + # dc to have a guest account + process = pmodels.Process.objects.create(person=person, applying_for=applying_for) + o = pops.ProcessApproveRT(audit_author=self.persons.fd, process=process) + @self.assertOperationSerializes(o) + def _(o): + self.assertApproveCommon(o) + self.assertEqual(o.process, process) + self.assertEqual(o.rt_queue, "Keyring") + self.assertEqual(o.rt_requestor, "nm@debian.org") + self.assertEqual(o.rt_subject, person.fullname + " to become " + const.ALL_STATUS_DESCS[applying_for]) + self.assertEqual(o.rt_cc, "{}, archive-{}@nm.debian.org, nm@debian.org".format(person.email, process.pk)) + + def _test_process_approve_dam_keyring(self, person, applying_for): + person = self.persons[person] + # dc to have a guest account + process = pmodels.Process.objects.create(person=person, applying_for=applying_for) + o = pops.ProcessApproveRT(audit_author=self.persons.fd, process=process) + @self.assertOperationSerializes(o) + def _(o): + self.assertApproveCommon(o) + self.assertEqual(o.process, process) + self.assertEqual(o.rt_queue, "Keyring") + self.assertEqual(o.rt_requestor, "da-manager@debian.org") + self.assertEqual(o.rt_subject, person.fullname + " to become " + const.ALL_STATUS_DESCS[applying_for]) + self.assertEqual(o.rt_cc, "{}, archive-{}@nm.debian.org, da-manager@debian.org".format(person.email, process.pk)) + + +class TestApprove(ProcessFixtureMixin, TestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.processes.create("proc", person=cls.persons.dc, applying_for=const.STATUS_DD_U) + + def test_op(self): + mail.outbox = [] + o = pops.ProcessApproveRT( + audit_author=self.persons.fd, + process=self.processes.proc, + rt_id="test id", + rt_queue="test queue", + rt_requestor="test requestor", + rt_subject="test subject", + rt_cc="test cc", + rt_text="test text") + + class MockResponse: + text = "\n".join([ + "RT/3.4.5 200 Ok", + "", + "# Ticket 1 created.", + ]) + def raise_for_status(self): + pass + + + with self.settings(RT_USER="rtuser", RT_PASS="rtpass"): + with patch('requests.post') as mock_post: + mock_post.return_value = MockResponse() + o.execute() + + post_args, post_kw = mock_post.call_args + self.assertEqual(len(post_args), 1) + self.assertEqual(post_args[0], "https://rt.debian.org/REST/1.0/ticket/new") + self.assertEqual(post_kw["params"], { "user": "rtuser", "pass": "rtpass" }) + content = post_kw["data"]["content"] + + self.assertEqual(content.splitlines(), [ + "id: test id", + "Queue: test queue", + "Requestor: test requestor", + "Subject: test subject", + "Cc: test cc", + "Text:", + " test text" + ]) + + proc = self.processes.proc + proc.refresh_from_db() + self.assertEqual(proc.rt_ticket, 1) + self.assertEqual(proc.rt_request, "test text") + self.assertEqual(proc.approved_by, self.persons.fd) + self.assertEqual(proc.approved_time, o.audit_time) + + log = proc.log.get() + self.assertEqual(log.changed_by, self.persons.fd) + self.assertIsNone(log.requirement) + self.assertTrue(log.is_public) + self.assertEqual(log.logdate, o.audit_time) + self.assertEqual(log.action, "proc_approve") + self.assertEqual(log.logtext, "Process approved") + self.assertEqual(len(mail.outbox), 0) + + +class TestApprovePerms(ProcessFixtureMixin, TestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.processes.create("proc", person=cls.persons.dc, applying_for=const.STATUS_DD_NU) + cls.processes.proc.frozen_by = cls.persons.fd + cls.processes.proc.frozen_time = now() + cls.processes.proc.save() + cls.url = reverse("process_approve", args=[cls.processes.proc.pk]) + + @classmethod + def __add_extra_tests__(cls): + for visitor in "fd", "dam": + cls._add_method(cls._test_success, visitor) + + for visitor in "pending", "dc", "dc_ga", "dm", "dm_ga", "dd_nu", "dd_u", "dd_e", "dd_r": + cls._add_method(cls._test_forbidden, visitor) + + def _test_success(self, visitor): + client = self.make_test_client(visitor) + + with self.collect_operations() as ops: + response = client.get(self.url) + self.assertEqual(response.status_code, 200) + self.assertEqual(len(ops), 0) + + response = client.post(self.url, data={"signed": "test statement"}) + self.assertRedirectMatches(response, self.processes.proc.get_absolute_url()) + self.assertEqual(len(ops), 1) + + op = ops[0] + self.assertEqual(op.audit_author, client.visitor) + self.assertEqual(op.audit_notes, "Process approved") + self.assertEqual(op.process, self.processes.proc) + self.assertEqual(op.rt_text, "test statement") + + with self.collect_operations() as ops: + self.processes.proc.frozen_by = None + self.processes.proc.frozen_time = None + self.processes.proc.save() + + # Denied because the process is not frozen + response = client.get(self.url) + self.assertPermissionDenied(response) + response = client.post(self.url, data={"signed": "test statement"}) + self.assertPermissionDenied(response) + self.assertEqual(len(ops), 0) + + self.processes.proc.frozen_by = self.persons.fd + self.processes.proc.frozen_time = now() + self.processes.proc.approved_by = self.persons.fd + self.processes.proc.approved_time = now() + self.processes.proc.save() + + # Denied because the process is already approved + response = client.get(self.url) + self.assertPermissionDenied(response) + response = client.post(self.url, data={"signed": "test statement"}) + self.assertPermissionDenied(response) + self.assertEqual(len(ops), 0) + + + def _test_forbidden(self, visitor): + client = self.make_test_client(visitor) + + with self.collect_operations() as ops: + response = client.get(self.url) + self.assertPermissionDenied(response) + self.assertEqual(len(ops), 0) + + response = client.post(self.url, data={"signed": "test statement"}) + self.assertPermissionDenied(response) + self.assertEqual(len(ops), 0) diff --git a/process/views.py b/process/views.py index dac0fefdd1186b614ef305393e3ee8a61962cb34..9adf97a20d4674b5a982b277e7848d0757659480 100644 --- a/process/views.py +++ b/process/views.py @@ -508,12 +508,13 @@ class Approve(VisitProcessMixin, FormView): def get_context_data(self, **kw): ctx = super().get_context_data(**kw) - rt_content = OrderedDict() - rt_content["id"] = self.op.rt_id - rt_content["Queue"] = self.op.rt_queue - rt_content["Requestor"] = self.op.rt_requestor - rt_content["Subject"] = self.op.rt_subject - rt_content["Cc"] = self.op.rt_cc + rt_content = [ + ("id", self.op.rt_id), + ("Queue", self.op.rt_queue), + ("Requestor", self.op.rt_requestor), + ("Subject", self.op.rt_subject), + ("Cc", self.op.rt_cc), + ] ctx["rt_content"] = rt_content ctx["text"] = make_rt_ticket_text(self.request, self.visitor, self.process) return ctx