Ported emeritus tests to op-based tests

process/ops.py

@@ -447,6 +447,19 @@ class RequestEmeritus(op.Operation):
         from .email import notify_new_statement
         return notify_new_statement(self._statement, request=request, cc_nm=False, notify_ml="private", mia="in, retired; emeritus via nm.d.o")
 
+    def _mock_execute(self, request=None):
+        try:
+            process = pmodels.Process.objects.get(person=self.person, applying_for__in=(const.STATUS_EMERITUS_DD, const.STATUS_REMOVED_DD))
+            requirement = process.requirements.get(type="intent")
+        except pmodels.Process.DoesNotExist:
+            process = pmodels.Process(self.person, const.STATUS_EMERITUS_DD)
+            process.pk = 1
+            requirement = pmodels.Requirement(process=process, type="ident")
+            requirement.pk = 2
+
+        self._statement = pmodels.Statement(requirement=requirement, uploaded_by=self.audit_author, uploaded_time=self.audit_time, statement=self.statement)
+        self._statement.pk = 3
+
 
 @op.Operation.register
 class ProcessCancel(op.Operation):

process/tests/test_emeritus.py

@@ -8,66 +8,45 @@ from unittest.mock import patch
 import time
 import process.models as pmodels
 import process.views as pviews
-from .common import (ProcessFixtureMixin,
-                     test_fingerprint1, test_fpr1_signed_valid_text,
-                     test_fingerprint2, test_fpr2_signed_valid_text)
+from process.unittest import ProcessFixtureMixin
+from process import ops as pops
 
 
 class TestEmeritus(ProcessFixtureMixin, TestCase):
-    @classmethod
-    def __add_extra_tests__(cls):
-        for visitor in "dd_nu", "dd_u", "fd", "dam":
-            cls._add_method(cls._test_success, visitor)
-            cls._add_method(cls._test_nonsso_success, visitor)
-            cls._add_method(cls._test_existing_success, visitor)
-            cls._add_method(cls._test_existing_nonsso_success, visitor)
+    def test_base_op(self):
+        o = pops.RequestEmeritus(audit_author=self.persons.fd, person=self.persons.dd_u, statement="test bye")
+        @self.assertOperationSerializes(o)
+        def _(o):
+            self.assertEqual(o.audit_author, self.persons.fd)
+            self.assertEqual(o.audit_notes, "Requested to become emeritus")
+            self.assertEqual(o.person, self.persons.dd_u)
+            self.assertEqual(o.statement, "test bye")
 
-        for visitor in "pending", "dc", "dc_ga", "dm", "dm_ga", "dd_e", "dd_r":
-            cls._add_method(cls._test_forbidden, visitor)
-            cls._add_method(cls._test_nonsso_forbidden, visitor)
+    def test_op_new_process(self):
+        self._test_op_common()
 
-    def _test_success(self, visitor):
-        mail.outbox = []
-        client = self.make_test_client(visitor)
-        self._test_success_common(visitor, client, reverse("process_emeritus"))
+    def test_op_existing_process(self):
+        self.processes.create("dd_u", person=self.persons.dd_u, applying_for=const.STATUS_EMERITUS_DD, fd_comment="test")
+        self._test_op_common()
 
-    def _test_existing_success(self, visitor):
-        mail.outbox = []
-        client = self.make_test_client(visitor)
-        self.processes.create("dd_u", person=self.persons[visitor], applying_for=const.STATUS_EMERITUS_DD, fd_comment="test")
-        self._test_success_common(visitor, client, reverse("process_emeritus"))
+    def _test_op_common(self):
+        o = pops.RequestEmeritus(audit_author=self.persons.fd, person=self.persons.dd_u, statement="test bye")
+        o.execute()
 
-    def _test_nonsso_success(self, visitor):
-        mail.outbox = []
-        url = pviews.Emeritus.get_nonauth_url(self.persons[visitor])
-        client = self.make_test_client(None)
-        self._test_success_common(visitor, client, url)
-        self._test_expired_token_common(visitor, client, url)
+        # No creation of duplicate processes
+        self.assertEquals(pmodels.Process.objects.filter(person=self.persons.dd_u).count(), 1)
 
-    def _test_existing_nonsso_success(self, visitor):
-        mail.outbox = []
-        url = pviews.Emeritus.get_nonauth_url(self.persons[visitor])
-        client = self.make_test_client(None)
-        self.processes.create("dd_u", person=self.persons[visitor], applying_for=const.STATUS_EMERITUS_DD, fd_comment="test")
-        self._test_success_common(visitor, client, url)
-        self._test_expired_token_common(visitor, client, url)
+        stm = o._statement
+        req = stm.requirement
+        process = req.process
 
-    def _test_success_common(self, visitor, client, url):
-        response = client.get(url)
-        self.assertEqual(response.status_code, 200)
-        self.assertContains(response, url)
-        response = client.post(url, data={"statement": "test statement"})
-        self.assertRedirectMatches(response, r"/process/\d+$")
-        visitor = self.persons[visitor]
-        process = pmodels.Process.objects.get(person=visitor, applying_for=const.STATUS_EMERITUS_DD, closed__isnull=True)
-        req = process.requirements.get(type="intent")
         status = req.compute_status()
         self.assertTrue(status["satisfied"])
         self.assertEqual(req.statements.count(), 1)
-        stm = req.statements.get()
+
         self.assertIsNone(stm.fpr)
-        self.assertEqual(stm.statement, "test statement")
-        self.assertEqual(stm.uploaded_by, visitor)
+        self.assertEqual(stm.statement, "test bye")
+        self.assertEqual(stm.uploaded_by, self.persons.fd)
         self.assertEqual(len(mail.outbox), 1)
         self.assertEqual(mail.outbox[0].to, ["debian-private@lists.debian.org"])
         self.assertCountEqual(mail.outbox[0].cc, ["{} <{}>".format(process.person.fullname, process.person.email), process.archive_email, "mia-{}@qa.debian.org".format(process.person.uid)])
@@ -76,68 +55,128 @@ class TestEmeritus(ProcessFixtureMixin, TestCase):
 
         # Submit again, it adds a new one
         mail.outbox = []
-        response = client.post(url, data={"statement": "test statement1"})
-        self.assertRedirectMatches(response, r"/process/\d+$")
+        o = pops.RequestEmeritus(audit_author=self.persons.fd, person=self.persons.dd_u, statement="test bye 1")
+        o.execute()
         req.refresh_from_db()
         status = req.compute_status()
         self.assertTrue(status["satisfied"])
         self.assertEqual(req.statements.count(), 2)
 
-        stm = req.statements.get(statement="test statement1")
+        stm = req.statements.get(statement="test bye 1")
         self.assertIsNone(stm.fpr)
-        self.assertEqual(stm.uploaded_by, visitor)
+        self.assertEqual(stm.uploaded_by, self.persons.fd)
         self.assertEqual(len(mail.outbox), 1)
         self.assertEqual(mail.outbox[0].to, ["debian-private@lists.debian.org"])
         self.assertCountEqual(mail.outbox[0].cc, ["{} <{}>".format(process.person.fullname, process.person.email), process.archive_email, "mia-{}@qa.debian.org".format(process.person.uid)])
         self.assertEqual(mail.outbox[0].extra_headers["X-MIA-Summary"], "in, retired; emeritus via nm.d.o")
         self.assertIn(stm.statement, mail.outbox[0].body)
 
+    @classmethod
+    def __add_extra_tests__(cls):
+        for visitor in "dd_nu", "dd_u", "fd", "dam":
+            cls._add_method(cls._test_success, visitor)
+            cls._add_method(cls._test_nonsso_success, visitor)
+            cls._add_method(cls._test_blocked, visitor)
+            cls._add_method(cls._test_nonsso_blocked, visitor)
+            cls._add_method(cls._test_expired_token, visitor)
+
+        for visitor in "pending", "dc", "dc_ga", "dm", "dm_ga", "dd_e", "dd_r":
+            cls._add_method(cls._test_forbidden, visitor)
+            cls._add_method(cls._test_nonsso_forbidden, visitor)
+            cls._add_method(cls._test_expired_token, visitor)
+
+    def _test_success(self, visitor):
+        client = self.make_test_client(visitor)
+        self._test_success_common(visitor, client, reverse("process_emeritus"))
+
+    def _test_nonsso_success(self, visitor):
+        url = pviews.Emeritus.get_nonauth_url(self.persons[visitor])
+        client = self.make_test_client(None)
+        self._test_success_common(visitor, client, url)
+
+    def _test_success_common(self, visitor, client, url):
+        visitor = self.persons[visitor]
+        with self.collect_operations() as ops:
+            response = client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertContains(response, url)
+            self.assertEqual(len(ops), 0)
+
+            response = client.post(url, data={"statement": "test statement"})
+            self.assertRedirectMatches(response, r"/process/1$")
+            self.assertEqual(len(ops), 1)
+
+        op = ops[0]
+        self.assertEqual(op.audit_author, visitor)
+        self.assertEqual(op.audit_notes, "Requested to become emeritus")
+        self.assertEqual(op.person, visitor)
+        self.assertEqual(op.statement, "test statement")
+
+    def _test_blocked(self, visitor):
+        client = self.make_test_client(visitor)
+        self.processes.create("dd_u", person=self.persons[visitor], applying_for=const.STATUS_EMERITUS_DD, fd_comment="test")
+        self._test_blocked_common(visitor, client, reverse("process_emeritus"))
+
+    def _test_nonsso_blocked(self, visitor):
+        url = pviews.Emeritus.get_nonauth_url(self.persons[visitor])
+        client = self.make_test_client(None)
+        self.processes.create("dd_u", person=self.persons[visitor], applying_for=const.STATUS_EMERITUS_DD, fd_comment="test")
+        self._test_blocked_common(visitor, client, url)
+
+    def _test_blocked_common(self, visitor, client, url):
+        process = self.processes.dd_u
+
         # check for closed EMERITUS_DD process
         process.closed = now()
         process.save()
-        self._text_blocked(client, url)
+        self._test_blocked_request(client, url)
+
         # check that if the process is turned into REMOVED_DD, the visitor can
         # no longer insert statements
         process.closed = None
         process.applying_for = const.STATUS_REMOVED_DD
         process.save()
-        self._text_blocked(client, url)
+        self._test_blocked_request(client, url)
+
         # check for closed REMOVED_DD process
         process.closed = now()
         process.save()
-        self._text_blocked(client, url)
+        self._test_blocked_request(client, url)
 
-    def _text_blocked(self, client, url):
+    def _test_blocked_request(self, client, url):
         response = client.get(url)
         self.assertEqual(response.status_code, 200)
         self.assertTrue(response.context["expired"])
         self.assertContains(response, "expired")  # XXX
         self.assertNotContains(response, "<textarea")
-        response = client.post(url, data={"statement": "test statement"})
-        self.assertPermissionDenied(response)
+        with self.collect_operations() as ops:
+            response = client.post(url, data={"statement": "test statement"})
+            self.assertPermissionDenied(response)
+            self.assertEquals(len(ops), 0)
 
     def _test_forbidden(self, visitor):
-        mail.outbox = []
         client = self.make_test_client(visitor)
         self._test_forbidden_common(visitor, client, reverse("process_emeritus"))
 
     def _test_nonsso_forbidden(self, visitor):
-        mail.outbox = []
         url = pviews.Emeritus.get_nonauth_url(self.persons[visitor])
         client = self.make_test_client(None)
         self._test_forbidden_common(visitor, client, url)
-        self._test_expired_token_common(visitor, client, url)
 
-    def _test_forbidden_common(self, visitor, client, url):
-        response = client.get(url)
-        self.assertPermissionDenied(response)
-        response = client.post(url, data={"statement": "test statement"})
-        self.assertPermissionDenied(response)
-        self.assertEqual(len(mail.outbox), 0)
-
-    def _test_expired_token_common(self, visitor, client, url):
+    def _test_expired_token(self, visitor):
+        url = pviews.Emeritus.get_nonauth_url(self.persons[visitor])
+        client = self.make_test_client(None)
         expired = time.time() + 92 * 3600 * 24
         with patch('time.time', return_value=expired) as mock_time:
             assert time.time() == expired
+
+            self._test_forbidden_common(visitor, client, url)
+
+    def _test_forbidden_common(self, visitor, client, url):
+        with self.collect_operations() as ops:
             response = client.get(url)
             self.assertPermissionDenied(response)
+            response = client.post(url, data={"statement": "test statement"})
+            self.assertPermissionDenied(response)
+            self.assertEqual(len(ops), 0)
+

process/tests/test_ops.py

@@ -193,16 +193,6 @@ class TestOps(ProcessFixtureMixin, TestCase):
         o = pops.ProcessApproveRT(audit_author=self.persons.fd, process=self.processes.app)
         self.check_op(o, check_contents)
 
-    def test_request_emeritus(self):
-        def check_contents(o):
-            self.assertEqual(o.audit_author, self.persons.fd)
-            self.assertEqual(o.audit_notes, "Requested to become emeritus")
-            self.assertIsInstance(o.audit_time, datetime.datetime)
-            self.assertEqual(o.person, self.persons.dd_u)
-            self.assertEqual(o.statement, "test bye")
-
-        o = pops.RequestEmeritus(audit_author=self.persons.fd, person=self.persons.dd_u, statement="test bye")
-        self.check_op(o, check_contents)
 
 
 class TestProcessClose(ProcessFixtureMixin, TestCase):