From 75a0393a758f607f1d4ef00192499145a2b46d01 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Laur=C3=A9line=20Gu=C3=A9rin?= <zebuline@entrouvert.com>
Date: Tue, 8 Feb 2022 12:02:37 +0100
Subject: [PATCH 1/3] misc: split workflow notification tests

---
 tests/workflow/test_all.py          |  99 ---------------------
 tests/workflow/test_notification.py | 132 ++++++++++++++++++++++++++++
 2 files changed, 132 insertions(+), 99 deletions(-)
 create mode 100644 tests/workflow/test_notification.py

diff --git a/tests/workflow/test_all.py b/tests/workflow/test_all.py
index 3bca5bf8..6c48782b 100644
--- a/tests/workflow/test_all.py
+++ b/tests/workflow/test_all.py
@@ -64,7 +64,6 @@ from wcs.wf.external_workflow import ManyExternalCallsPart
 from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormFieldsFormDef
 from wcs.wf.geolocate import GeolocateWorkflowStatusItem
 from wcs.wf.jump import JumpWorkflowStatusItem, _apply_timeouts
-from wcs.wf.notification import SendNotificationWorkflowStatusItem
 from wcs.wf.profile import UpdateUserProfileStatusItem
 from wcs.wf.redirect_to_url import RedirectToUrlWorkflowStatusItem
 from wcs.wf.register_comment import JournalEvolutionPart, RegisterCommenterWorkflowStatusItem
@@ -5948,104 +5947,6 @@ def test_workflow_action_condition(two_pubs):
         assert logged_error.expression_type == 'python'
 
 
-def test_notifications(pub, http_requests):
-    formdef = FormDef()
-    formdef.name = 'baz'
-    formdef.fields = []
-    formdef.store()
-
-    formdata = formdef.data_class()()
-    formdata.just_created()
-    formdata.store()
-
-    assert not SendNotificationWorkflowStatusItem.is_available()
-
-    if not pub.site_options.has_section('variables'):
-        pub.site_options.add_section('variables')
-    pub.site_options.set('variables', 'portal_url', 'https://portal/')
-    assert SendNotificationWorkflowStatusItem.is_available()
-
-    item = SendNotificationWorkflowStatusItem()
-    assert item.to == ['_submitter']
-    item.title = 'xxx'
-    item.body = 'XXX'
-
-    # no user
-    http_requests.empty()
-    item.perform(formdata)
-    assert http_requests.count() == 0
-
-    # user
-    http_requests.empty()
-    user = pub.user_class()
-    user.name_identifiers = ['xxx']
-    user.store()
-    formdata.user_id = user.id
-    formdata.store()
-
-    item.perform(formdata)
-    assert http_requests.count() == 1
-    assert http_requests.get_last('url') == 'https://portal/api/notification/add/?NameID=xxx'
-    assert json.loads(http_requests.get_last('body')) == {
-        'body': 'XXX',
-        'url': formdata.get_url(),
-        'id': 'formdata:%s' % formdata.get_display_id(),
-        'origin': '',
-        'summary': 'xxx',
-    }
-
-    # deleted user
-    http_requests.empty()
-    user.deleted_timestamp = datetime.datetime.now()
-    user.store()
-    item.perform(formdata)
-    assert http_requests.count() == 0
-
-    # roles (not exposed in current UI)
-    http_requests.empty()
-    user.deleted_timestamp = datetime.datetime.now()
-    user.store()
-
-    role = pub.role_class(name='blah')
-    role.store()
-
-    user1 = pub.user_class()
-    user1.roles = [role.id]
-    user1.name_identifiers = ['xxy1']
-    user1.store()
-    user2 = pub.user_class()
-    user2.roles = [role.id]
-    user2.name_identifiers = ['xxy2']
-    user2.store()
-
-    formdef.workflow_roles = {'_receiver': role.id}
-
-    item.to = ['_receiver']
-    item.perform(formdata)
-    assert http_requests.count() == 2
-    assert {x['url'] for x in http_requests.requests} == {
-        'https://portal/api/notification/add/?NameID=xxy1',
-        'https://portal/api/notification/add/?NameID=xxy2',
-    }
-
-    # test inactive users are ignored
-    user2.is_active = False
-    user2.store()
-    http_requests.empty()
-    item.perform(formdata)
-    assert http_requests.count() == 1
-    assert {x['url'] for x in http_requests.requests} == {'https://portal/api/notification/add/?NameID=xxy1'}
-
-    # check notifications are sent to interco portal if it exists
-    pub.site_options.set('variables', '_interco_portal_url', 'https://interco-portal/')
-    http_requests.empty()
-    item.perform(formdata)
-    assert http_requests.count() == 1
-    assert {x['url'] for x in http_requests.requests} == {
-        'https://interco-portal/api/notification/add/?NameID=xxy1'
-    }
-
-
 def test_workflow_field_migration(pub):
     Workflow.wipe()
     wf = Workflow(name='wf with backoffice field')
diff --git a/tests/workflow/test_notification.py b/tests/workflow/test_notification.py
new file mode 100644
index 00000000..9a2cb726
--- /dev/null
+++ b/tests/workflow/test_notification.py
@@ -0,0 +1,132 @@
+import datetime
+import json
+
+import pytest
+from quixote import cleanup
+
+from wcs import sessions
+from wcs.formdef import FormDef
+from wcs.qommon.http_request import HTTPRequest
+from wcs.wf.notification import SendNotificationWorkflowStatusItem
+
+from ..utilities import clean_temporary_pub, create_temporary_pub
+
+
+def setup_module(module):
+    cleanup()
+
+
+def teardown_module(module):
+    clean_temporary_pub()
+
+
+@pytest.fixture
+def pub(request):
+    pub = create_temporary_pub()
+    pub.cfg['language'] = {'language': 'en'}
+    pub.write_cfg()
+    req = HTTPRequest(None, {'SERVER_NAME': 'example.net', 'SCRIPT_NAME': ''})
+    req.response.filter = {}
+    req._user = None
+    pub._set_request(req)
+    req.session = sessions.BasicSession(id=1)
+    pub.set_config(req)
+    return pub
+
+
+def test_notifications(pub, http_requests):
+    formdef = FormDef()
+    formdef.name = 'baz'
+    formdef.fields = []
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+    formdata.store()
+
+    assert not SendNotificationWorkflowStatusItem.is_available()
+
+    if not pub.site_options.has_section('variables'):
+        pub.site_options.add_section('variables')
+    pub.site_options.set('variables', 'portal_url', 'https://portal/')
+    assert SendNotificationWorkflowStatusItem.is_available()
+
+    item = SendNotificationWorkflowStatusItem()
+    assert item.to == ['_submitter']
+    item.title = 'xxx'
+    item.body = 'XXX'
+
+    # no user
+    http_requests.empty()
+    item.perform(formdata)
+    assert http_requests.count() == 0
+
+    # user
+    http_requests.empty()
+    user = pub.user_class()
+    user.name_identifiers = ['xxx']
+    user.store()
+    formdata.user_id = user.id
+    formdata.store()
+
+    item.perform(formdata)
+    assert http_requests.count() == 1
+    assert http_requests.get_last('url') == 'https://portal/api/notification/add/?NameID=xxx'
+    assert json.loads(http_requests.get_last('body')) == {
+        'body': 'XXX',
+        'url': formdata.get_url(),
+        'id': 'formdata:%s' % formdata.get_display_id(),
+        'origin': '',
+        'summary': 'xxx',
+    }
+
+    # deleted user
+    http_requests.empty()
+    user.deleted_timestamp = datetime.datetime.now()
+    user.store()
+    item.perform(formdata)
+    assert http_requests.count() == 0
+
+    # roles (not exposed in current UI)
+    http_requests.empty()
+    user.deleted_timestamp = datetime.datetime.now()
+    user.store()
+
+    role = pub.role_class(name='blah')
+    role.store()
+
+    user1 = pub.user_class()
+    user1.roles = [role.id]
+    user1.name_identifiers = ['xxy1']
+    user1.store()
+    user2 = pub.user_class()
+    user2.roles = [role.id]
+    user2.name_identifiers = ['xxy2']
+    user2.store()
+
+    formdef.workflow_roles = {'_receiver': role.id}
+
+    item.to = ['_receiver']
+    item.perform(formdata)
+    assert http_requests.count() == 2
+    assert {x['url'] for x in http_requests.requests} == {
+        'https://portal/api/notification/add/?NameID=xxy1',
+        'https://portal/api/notification/add/?NameID=xxy2',
+    }
+
+    # test inactive users are ignored
+    user2.is_active = False
+    user2.store()
+    http_requests.empty()
+    item.perform(formdata)
+    assert http_requests.count() == 1
+    assert {x['url'] for x in http_requests.requests} == {'https://portal/api/notification/add/?NameID=xxy1'}
+
+    # check notifications are sent to interco portal if it exists
+    pub.site_options.set('variables', '_interco_portal_url', 'https://interco-portal/')
+    http_requests.empty()
+    item.perform(formdata)
+    assert http_requests.count() == 1
+    assert {x['url'] for x in http_requests.requests} == {
+        'https://interco-portal/api/notification/add/?NameID=xxy1'
+    }
-- 
2.34.1

