applification: add test users (#89640)

tests/api/test_export_import.py

@@ -7,6 +7,7 @@ import xml.etree.ElementTree as ET
 
 import pytest
 
+from wcs import workflow_tests
 from wcs.api_export_import import BundleDeclareJob, BundleImportJob, klass_to_slug
 from wcs.applications import Application, ApplicationElement
 from wcs.backoffice.deprecations import DeprecationsScan
@@ -66,6 +67,8 @@ coucou = 1234
     NamedDataSource.wipe()
     NamedWsCall.wipe()
     pub.custom_view_class.wipe()
+    TestDef.wipe()
+    pub.user_class.wipe()
 
     return pub
 
@@ -138,6 +141,8 @@ def test_export_import_dependencies(pub):
     role4 = pub.role_class(name='Fourth role')
     role4.uuid = str(uuid.uuid4())
     role4.store()
+    role5 = pub.role_class(name='Fifth role')
+    role5.store()
 
     wscall = NamedWsCall(name='Test')
     wscall.store()
@@ -345,6 +350,21 @@ def test_export_import_dependencies(pub):
     formdef.lateral_template = 'x{{ webservice.test_in_lateral_template.blah }}y'
     formdef.store()
 
+    user = pub.user_class(name='test user')
+    user.test_uuid = '42'
+    user.roles = [role5.id]
+    user.store()
+    user2 = pub.user_class(name='test user 2')
+    user2.test_uuid = '43'
+    user2.store()
+
+    testdef = TestDef.create_from_formdata(formdef, formdef.data_class()())
+    testdef.user_uuid = user.test_uuid
+    testdef.workflow_tests.actions = [
+        workflow_tests.ButtonClick(button_name='Name', who='other', who_id=user2.test_uuid),
+    ]
+    testdef.store()
+
     resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
     form_data = [d for d in resp.json['data'] if d['id'] == 'test']
     resp = get_app(pub).get(sign_uri(form_data[0]['urls']['dependencies']))
@@ -364,6 +384,9 @@ def test_export_import_dependencies(pub):
         ('second-role', 'roles'),
         ('third-role', 'roles'),
         ('fourth-role', 'roles'),
+        ('fifth-role', 'roles'),
+        ('42', 'users'),
+        ('43', 'users'),
     }
     for dependency in resp.json['data']:
         if dependency['type'] == 'roles':
@@ -550,6 +573,10 @@ def test_export_import_redirect_url(pub):
     comment_template_category = CommentTemplateCategory(name='Test')
     comment_template_category.store()
 
+    user = pub.user_class(name='test user')
+    user.test_uuid = 'test'
+    user.store()
+
     elements = [
         ('forms', '/backoffice/forms/%s/' % formdef.id),
         ('cards', '/backoffice/cards/%s/' % carddef.id),
@@ -571,6 +598,7 @@ def test_export_import_redirect_url(pub):
             'comment-templates-categories',
             '/backoffice/workflows/comment-templates/categories/%s/' % comment_template_category.id,
         ),
+        ('users', '/backoffice/forms/test-users/%s/' % user.id),
     ]
     for object_type, obj_url in elements:
         resp = get_app(pub).get(sign_uri('/api/export-import/%s/' % object_type))
@@ -680,6 +708,10 @@ def test_export_import_bundle_import(pub):
     testdef = TestDef.create_from_formdata(formdef, formdef.data_class()())
     testdef.store()
 
+    user = pub.user_class(name='test user')
+    user.test_uuid = '42'
+    user.store()
+
     custom_view = pub.custom_view_class()
     custom_view.title = 'shared formdef custom view'
     custom_view.formdef = formdef
@@ -738,6 +770,7 @@ def test_export_import_bundle_import(pub):
                     {'type': 'data-sources-categories', 'slug': 'test', 'name': 'test'},
                     {'type': 'data-sources', 'slug': 'test', 'name': 'test'},
                     {'type': 'wscalls', 'slug': 'test', 'name': 'test'},
+                    {'type': 'users', 'slug': '42', 'name': '42'},
                     {'type': 'foobar', 'slug': 'test', 'name': 'test'},
                 ],
                 ('forms-categories/test', category),
@@ -756,6 +789,7 @@ def test_export_import_bundle_import(pub):
                 ('comment-templates/test', comment_template),
                 ('roles/test', role),
                 ('wscalls/test', wscall),
+                ('users/42', user),
                 version_number=version_number,
             )
         )
@@ -778,6 +812,7 @@ def test_export_import_bundle_import(pub):
         pub.role_class,
         NamedWsCall,
         TestDef,
+        pub.test_user_class,
     ]
     for object_class in object_classes:
         object_class.wipe()
@@ -794,7 +829,7 @@ def test_export_import_bundle_import(pub):
     afterjob_url = resp.json['url']
     resp = get_app(pub).put(sign_uri(afterjob_url))
     assert resp.json['data']['status'] == 'completed'
-    assert resp.json['data']['completion_status'] == '34/34 (100%)'
+    assert resp.json['data']['completion_status'] == '36/36 (100%)'
 
     assert Category.count() == 1
     assert FormDef.count() == 1
@@ -822,6 +857,7 @@ def test_export_import_bundle_import(pub):
     assert NamedDataSource.select()[0].category_id == DataSourceCategory.select()[0].id
     assert NamedWsCall.count() == 1
     assert TestDef.count() == 1
+    assert pub.test_user_class.count() == 1
     assert pub.custom_view_class().count() == 1
     assert Application.count() == 1
     application = Application.select()[0]
@@ -834,7 +870,7 @@ def test_export_import_bundle_import(pub):
     assert application.icon.base_filename == 'foo.png'
     assert application.editable is False
     assert application.visible is True
-    assert ApplicationElement.count() == 15
+    assert ApplicationElement.count() == 16
 
     # check backoffice field have been added to table
     # (a sql error would happen if it was missing)
@@ -910,8 +946,9 @@ def test_export_import_bundle_import(pub):
     assert pub.custom_view_class().count() == 1
     assert NamedWsCall.count() == 1
     assert TestDef.count() == 1
+    assert pub.test_user_class.count() == 1
     assert Application.count() == 1
-    assert ApplicationElement.count() == 15
+    assert ApplicationElement.count() == 16
     assert (
         ApplicationElement.select(
             [
@@ -1605,6 +1642,10 @@ def test_export_import_bundle_check(pub):
     wscall = NamedWsCall(name='Test')
     wscall.store()
 
+    user = pub.user_class(name='Test')
+    user.test_uuid = 'test'
+    user.store()
+
     bundles = []
     for version in ['1.0', '2.0']:
         bundles.append(
@@ -1626,6 +1667,7 @@ def test_export_import_bundle_check(pub):
                     {'type': 'data-sources-categories', 'slug': 'test', 'name': 'test'},
                     {'type': 'data-sources', 'slug': 'test', 'name': 'test'},
                     {'type': 'wscalls', 'slug': 'test', 'name': 'test'},
+                    {'type': 'users', 'slug': 'test', 'name': 'test'},
                     {'type': 'foobar', 'slug': 'test', 'name': 'test'},
                 ],
                 ('forms-categories/test', category),
@@ -1644,6 +1686,7 @@ def test_export_import_bundle_check(pub):
                 ('comment-templates/test', comment_template),
                 ('roles/test', role),
                 ('wscalls/test', wscall),
+                ('users/test', user),
                 visible=False,
                 version=version,
             )
@@ -1667,6 +1710,7 @@ def test_export_import_bundle_check(pub):
         pub.custom_view_class,
         pub.role_class,
         NamedWsCall,
+        pub.test_user_class,
     ]
     for object_class in object_classes:
         object_class.wipe()
@@ -1718,6 +1762,7 @@ def test_export_import_bundle_check(pub):
                 {'slug': 'test', 'type': 'data-sources-categories'},
                 {'slug': 'test', 'type': 'data-sources'},
                 {'slug': 'test', 'type': 'wscalls'},
+                {'slug': 'test', 'type': 'users'},
             ],
             'legacy_elements': [],
         },
@@ -1730,9 +1775,9 @@ def test_export_import_bundle_check(pub):
     afterjob_url = resp.json['url']
     resp = get_app(pub).put(sign_uri(afterjob_url))
     assert resp.json['data']['status'] == 'completed'
-    assert resp.json['data']['completion_status'] == '34/34 (100%)'
+    assert resp.json['data']['completion_status'] == '36/36 (100%)'
     assert Application.count() == 1
-    assert ApplicationElement.count() == 15
+    assert ApplicationElement.count() == 16
 
     # remove application links
     Application.wipe()
@@ -1836,6 +1881,12 @@ def test_export_import_bundle_check(pub):
                     'type': 'wscalls',
                     'url': 'http://example.net/api/export-import/wscalls/test/redirect/',
                 },
+                {
+                    'slug': 'test',
+                    'text': 'Test',
+                    'type': 'users',
+                    'url': 'http://example.net/api/export-import/users/test/redirect/',
+                },
             ],
         }
     }
@@ -1847,9 +1898,9 @@ def test_export_import_bundle_check(pub):
     afterjob_url = resp.json['url']
     resp = get_app(pub).put(sign_uri(afterjob_url))
     assert resp.json['data']['status'] == 'completed'
-    assert resp.json['data']['completion_status'] == '34/34 (100%)'
+    assert resp.json['data']['completion_status'] == '36/36 (100%)'
     assert Application.count() == 1
-    assert ApplicationElement.count() == 15
+    assert ApplicationElement.count() == 16
 
     # no changes since last import
     resp = get_app(pub).post(
@@ -1978,6 +2029,12 @@ def test_export_import_bundle_check(pub):
                     'url': 'http://example.net/backoffice/settings/wscalls/test/history/compare?version1=%s&version2=%s'
                     % snapshots['wscall:test'],
                 },
+                {
+                    'slug': 'test',
+                    'type': 'users',
+                    'url': 'http://example.net/backoffice/forms/test-users/%s/history/compare?version1=%s&version2=%s'
+                    % (pub.test_user_class.select()[0].id, *snapshots['user:test']),
+                },
             ],
             'unknown_elements': [],
             'no_history_elements': [],
@@ -1992,7 +2049,7 @@ def test_export_import_bundle_check(pub):
     afterjob_url = resp.json['url']
     resp = get_app(pub).put(sign_uri(afterjob_url))
     assert resp.json['data']['status'] == 'completed'
-    assert resp.json['data']['completion_status'] == '34/34 (100%)'
+    assert resp.json['data']['completion_status'] == '36/36 (100%)'
 
     # and check
     resp = get_app(pub).post(
@@ -2034,6 +2091,7 @@ def test_export_import_bundle_check(pub):
                 {'slug': 'test', 'type': 'data-sources-categories'},
                 {'slug': 'test', 'type': 'data-sources'},
                 {'slug': 'test', 'type': 'wscalls'},
+                {'slug': 'test', 'type': 'users'},
             ],
             'unknown_elements': [],
             'legacy_elements': [],

wcs/api_export_import.py

@@ -41,7 +41,7 @@ from wcs.comment_templates import CommentTemplate
 from wcs.data_sources import NamedDataSource, NamedDataSourceImportError
 from wcs.formdef import FormDef, FormdefImportError
 from wcs.mail_templates import MailTemplate
-from wcs.sql import Equal, Role
+from wcs.sql import Equal, Role, TestUser
 from wcs.workflows import Workflow, WorkflowImportError
 from wcs.wscalls import NamedWsCall, NamedWsCallImportError
 
@@ -66,6 +66,7 @@ klasses = {
     'workflows-categories': WorkflowCategory,
     'workflows': Workflow,
     'wscalls': NamedWsCall,
+    'users': TestUser,
 }
 
 klass_to_slug = {y: x for x, y in klasses.items()}
@@ -154,6 +155,7 @@ def index(request):
             'singular': _('Role'),
             'minor': True,
         },
+        {'id': 'users', 'text': _('Test users'), 'singular': _('Test user'), 'minor': True},
     ]
     for obj in response:
         obj['urls'] = {
@@ -304,6 +306,8 @@ def bundle_check(request):
                 tree = ET.fromstring(element_content)
                 if hasattr(element_klass, 'url_name'):
                     slug = xml_node_text(tree.find('url_name'))
+                elif hasattr(element_klass, 'test_uuid'):
+                    slug = xml_node_text(tree.find('test_uuid'))
                 else:
                     slug = xml_node_text(tree.find('slug'))
                 try:
@@ -329,7 +333,7 @@ def bundle_check(request):
                         [
                             Equal('application_id', application.id),
                             Equal('object_type', obj.xml_root_node),
-                            Equal('object_id', obj.id),
+                            Equal('object_id', str(obj.id)),
                         ]
                     )
                     if not elements:
@@ -354,7 +358,7 @@ def bundle_check(request):
                 snapshots_for_app = get_publisher().snapshot_class.select(
                     [
                         Equal('object_type', obj.xml_root_node),
-                        Equal('object_id', obj.id),
+                        Equal('object_id', str(obj.id)),
                         Equal('application_slug', application_slug),
                         Equal('application_version', application_version),
                     ],
@@ -623,7 +627,7 @@ class BundleImportJob(AfterJob):
 
     def link_object(self, obj):
         element = ApplicationElement.update_or_create_for_object(self.application, obj)
-        self.application_elements.add((element.object_type, element.object_id))
+        self.application_elements.add((element.object_type, str(element.object_id)))
 
     def unlink_obsolete_objects(self):
         known_elements = ApplicationElement.select([Equal('application_id', self.application.id)])

wcs/applications.py

@@ -154,7 +154,7 @@ class ApplicationElement(sql.ApplicationElement):
             [
                 sql.Equal('application_id', application.id),
                 sql.Equal('object_type', obj.xml_root_node),
-                sql.Equal('object_id', obj.id),
+                sql.Equal('object_id', str(obj.id)),
             ]
         )
         if elements:

wcs/formdef.py

@@ -728,6 +728,11 @@ class FormDef(StorableObject):
         ]:
             yield from get_dependencies_from_template(template)
 
+        from .testdef import TestDef
+
+        for testdef in TestDef.select_for_objectdef(self):
+            yield from testdef.get_dependencies()
+
     @property
     def keywords_list(self):
         if not self.keywords:

wcs/sql.py

@@ -2944,7 +2944,7 @@ class SqlUser(SqlMixin, wcs.users.User):
         return super().count(clause=clause)
 
     @invalidate_substitution_cache
-    def store(self, comment=None):
+    def store(self, comment=None, application=None):
         sql_dict = {
             'name': self.name,
             'ascii_name': self.ascii_name,
@@ -3039,7 +3039,7 @@ class SqlUser(SqlMixin, wcs.users.User):
         cur.close()
 
         if self.test_uuid and get_publisher().snapshot_class:
-            get_publisher().snapshot_class.snap(instance=self, comment=comment)
+            get_publisher().snapshot_class.snap(instance=self, comment=comment, application=application)
 
     @classmethod
     def _row2ob(cls, row, **kwargs):

wcs/testdef.py

@@ -604,6 +604,25 @@ class TestDef(sql.TestDef):
 
         return testdef
 
+    def get_dependencies(self):
+        if self.user_uuid:
+            try:
+                user = get_publisher().test_user_class.select([Equal('test_uuid', self.user_uuid)])[0]
+            except IndexError:
+                pass
+            else:
+                yield user
+                yield from user.get_dependencies()
+        if self.agent_id:
+            try:
+                user = get_publisher().test_user_class.select([Equal('test_uuid', self.agent_id)])[0]
+            except IndexError:
+                pass
+            else:
+                yield user
+                yield from user.get_dependencies()
+        yield from self.workflow_tests.get_dependencies()
+
 
 class TestResult(sql.TestResult):
     _names = 'test_result'

wcs/users.py

@@ -23,6 +23,7 @@ from quixote import get_publisher
 
 from wcs.api_utils import get_secret_and_orig, sign_url
 from wcs.sql_criterias import ArrayContains, Equal, Intersects, Not, Null, Or
+from wcs.workflows import get_role_dependencies
 
 from .qommon import _, get_cfg
 from .qommon.misc import get_formatted_phone, http_post_request, simplify
@@ -414,6 +415,20 @@ class User(XmlStorableObject):
     def get_admin_url(self):
         return '%s/forms/test-users/%s/' % (get_publisher().get_backoffice_url(), self.id)
 
+    @property
+    def slug(self):
+        return self.test_uuid
+
+    @classmethod
+    def get_by_slug(cls, slug, **kwargs):
+        try:
+            return cls.select([Equal('test_uuid', slug)])[0]
+        except IndexError:
+            return None
+
+    def get_dependencies(self):
+        yield from get_role_dependencies(self.roles)
+
 
 Substitutions.register(
     'session_user_display_name',

wcs/workflow_tests.py

@@ -207,6 +207,10 @@ class WorkflowTests(XmlStorableObject):
 
         return actions
 
+    def get_dependencies(self):
+        for action in self.actions:
+            yield from action.get_dependencies()
+
 
 class WorkflowTestAction(XmlStorableObject):
     xml_root_node = 'test-action'
@@ -251,6 +255,9 @@ class WorkflowTestAction(XmlStorableObject):
 
         return self.details_label
 
+    def get_dependencies(self):
+        return []
+
 
 class ButtonClick(WorkflowTestAction):
     label = _('Simulate click on action button')
@@ -410,6 +417,13 @@ class ButtonClick(WorkflowTestAction):
             **{'data-autocomplete': 'true'},
         )
 
+    def get_dependencies(self):
+        if self.who == 'other' and self.who_id:
+            try:
+                yield get_publisher().test_user_class.select([Equal('test_uuid', self.who_id)])[0]
+            except IndexError:
+                pass
+
 
 class AssertStatus(WorkflowTestAction):
     label = _('Assert form status')