From aa3511611acc76eb4cf7217f2271d84b2b5e2ee3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Fr=C3=A9d=C3=A9ric=20P=C3=A9ters?= <fpeters@entrouvert.com>
Date: Sat, 24 Sep 2022 16:54:45 +0200
Subject: [PATCH] general: add new grep command (#53994)

---
 tests/test_ctl.py                   |  83 ++++++++++++++++
 wcs/ctl/management/commands/grep.py | 144 ++++++++++++++++++++++++++++
 wcs/formdef.py                      |  13 +--
 3 files changed, 234 insertions(+), 6 deletions(-)
 create mode 100644 wcs/ctl/management/commands/grep.py

diff --git a/tests/test_ctl.py b/tests/test_ctl.py
index edb038cf2..60ea35971 100644
--- a/tests/test_ctl.py
+++ b/tests/test_ctl.py
@@ -16,12 +16,14 @@ from wcs.ctl.rebuild_indexes import rebuild_vhost_indexes
 from wcs.ctl.wipe_data import CmdWipeData
 from wcs.fields import EmailField, StringField
 from wcs.formdef import FormDef
+from wcs.mail_templates import MailTemplate
 from wcs.qommon.afterjobs import AfterJob
 from wcs.qommon.management.commands.collectstatic import Command as CmdCollectStatic
 from wcs.qommon.management.commands.migrate import Command as CmdMigrate
 from wcs.qommon.management.commands.migrate_schemas import Command as CmdMigrateSchemas
 from wcs.sql import cleanup_connection, get_connection_and_cursor
 from wcs.workflows import Workflow, WorkflowStatusItem
+from wcs.wscalls import NamedWsCall
 
 from .utilities import clean_temporary_pub, create_temporary_pub
 
@@ -530,3 +532,84 @@ def test_makemessages(pub):
     # just make sure it loads correctly
     with pytest.raises(SystemExit):
         call_command('makemessages', '--help')
+
+
+def test_grep(pub):
+    FormDef.wipe()
+    Workflow.wipe()
+    NamedWsCall.wipe()
+    MailTemplate.wipe()
+
+    formdef = FormDef()
+    formdef.name = 'test'
+    formdef.fields = [StringField(id='1', label='Your Name'), EmailField(id='2', label='Email')]
+    formdef.options = {'x': 'Name'}
+    formdef.store()
+
+    workflow = Workflow()
+    workflow.name = 'test'
+    st = workflow.add_status('status')
+    st.add_action('aggregationemail')
+    workflow.store()
+
+    wscall = NamedWsCall()
+    wscall.name = 'Hello'
+    wscall.request = {'url': 'http://example.org/api/test', 'qs_data': {'a': 'b'}}
+    wscall.store()
+
+    mail_template = MailTemplate(name='test mail template')
+    mail_template.subject = 'test subject'
+    mail_template.body = 'test body'
+    mail_template.attachments = ['form_var_file1_raw']
+    mail_template.store()
+
+    with pytest.raises(CommandError):
+        call_command('grep')
+
+    with pytest.raises(CommandError):
+        call_command('grep', 'xxx')
+
+    with pytest.raises(CommandError):
+        call_command('grep', '--all-tenants', '--domain', 'example.net', 'xxx')
+
+    with pytest.raises(CommandError):
+        call_command('grep', '--domain', 'example.net', '--type', 'foo', 'xxx')
+
+    with mock.patch('wcs.ctl.management.commands.grep.Command.print_hit') as print_hit:
+        call_command('grep', '--domain', 'example.net', '--type', 'action-types', 'email')
+        assert print_hit.call_args[0] == ('http://example.net/backoffice/workflows/1/status/1/items/1/',)
+        print_hit.reset_mock()
+
+        call_command('grep', '--domain', 'example.net', '--type', 'field-types', 'email')
+        assert print_hit.call_args[0] == ('http://example.net/backoffice/forms/1/fields/2/',)
+        print_hit.reset_mock()
+
+        call_command('grep', '--domain', 'example.net', 'Name')
+        assert print_hit.call_count == 2
+        assert print_hit.call_args_list[0].args == (
+            'http://example.net/backoffice/forms/1/fields/1/',
+            'Your Name',
+        )
+        assert print_hit.call_args_list[1].args == (
+            'http://example.net/backoffice/forms/1/workflow-variables',
+            'Name',
+        )
+        print_hit.reset_mock()
+
+        call_command('grep', '--domain', 'example.net', '/api/test')
+        assert print_hit.call_args[0] == (
+            'http://example.net/backoffice/settings/wscalls/hello/',
+            'http://example.org/api/test',
+        )
+        print_hit.reset_mock()
+
+        call_command('grep', '--domain', 'example.net', 'form_var_file1_raw')
+        assert print_hit.call_args[0] == (
+            'http://example.net/backoffice/workflows/mail-templates/1/',
+            'form_var_file1_raw',
+        )
+        print_hit.reset_mock()
+
+        call_command('grep', '--domain', 'example.net', 'xxx')
+        assert print_hit.call_count == 0
+        print_hit.reset_mock()
diff --git a/wcs/ctl/management/commands/grep.py b/wcs/ctl/management/commands/grep.py
new file mode 100644
index 000000000..9afda6cbc
--- /dev/null
+++ b/wcs/ctl/management/commands/grep.py
@@ -0,0 +1,144 @@
+# w.c.s. - web application for online forms
+# Copyright (C) 2005-2022  Entr'ouvert
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <http://www.gnu.org/licenses/>.
+
+import itertools
+
+from django.core.management import CommandError
+
+from wcs.data_sources import NamedDataSource
+from wcs.formdef import get_formdefs_of_all_kinds
+from wcs.mail_templates import MailTemplate
+from wcs.qommon.publisher import get_publisher_class
+from wcs.workflows import Workflow
+from wcs.wscalls import NamedWsCall
+
+from . import TenantCommand
+
+
+class Command(TenantCommand):
+    help = '''Grep tenant(s) for a given pattern'''
+
+    def add_arguments(self, parser):
+        parser.add_argument('-d', '--domain', '--vhost', metavar='DOMAIN')
+        parser.add_argument('--all-tenants', action='store_true')
+        parser.add_argument(
+            '--type',
+            metavar='TYPE',
+            default='strings',
+            choices=['strings', 'action-types', 'field-types'],
+            help='strings (default), action-types, field-types',
+        )
+        parser.add_argument('--urls', action='store_true', help='display URLs to hits')
+        parser.add_argument('pattern', nargs='+')
+
+    def handle(self, *args, **options):
+        domain = options.get('domain')
+        all_tenants = options.get('all_tenants')
+        if domain and all_tenants:
+            raise CommandError('--domain and --all-tenants are exclusive')
+        if not (domain or all_tenants):
+            raise CommandError('--domain or --all-tenants are required')
+        if domain:
+            domains = [domain]
+        else:
+            domains = [x.hostname for x in get_publisher_class().get_tenants()]
+
+        self.pattern = ' '.join(options.get('pattern'))
+        self.urls = options.get('urls')
+        try:
+            method = getattr(self, 'grep_' + options.get('type').replace('-', '_'))
+        except AttributeError:
+            raise CommandError('unknown grep type')
+        for domain in domains:
+            self.init_tenant_publisher(domain, register_tld_names=False)
+            method()
+
+    def grep_strings(self):
+        for formdef in get_formdefs_of_all_kinds(order_by='id'):
+            url = formdef.get_admin_url()
+            for attr in formdef.TEXT_ATTRIBUTES:
+                if self.check_string(getattr(formdef, attr, None), url=url):
+                    break
+
+            for field in formdef.fields or []:
+                url = formdef.get_field_admin_url(field)
+                for attr in field.get_admin_attributes():
+                    if self.check_string(getattr(field, attr, None), url=url):
+                        break
+            if hasattr(formdef, 'options'):
+                url = formdef.get_admin_url() + 'workflow-variables'
+                for option_value in (formdef.options or {}).values():
+                    if self.check_string(option_value, url=url):
+                        break
+
+        select_kwargs = {'ignore_errors': True, 'ignore_migration': True, 'order_by': 'id'}
+        for workflow in Workflow.select(**select_kwargs):
+            for action in workflow.get_all_items():
+                url = action.get_admin_url()
+                for parameter in action.get_parameters():
+                    if self.check_string(getattr(action, parameter, None), url=url):
+                        break
+
+        for obj in itertools.chain(
+            NamedDataSource.select(**select_kwargs),
+            NamedWsCall.select(**select_kwargs),
+            MailTemplate.select(**select_kwargs),
+        ):
+            url = obj.get_admin_url()
+            for attr, attr_type in obj.XML_NODES:
+                attr_value = getattr(obj, attr, None)
+                if attr_type == 'str':
+                    if self.check_string(attr_value, url=url):
+                        break
+                elif attr_type == 'str_list':
+                    for str_item in attr_value or []:
+                        if self.check_string(str_item, url=url):
+                            break
+                    else:
+                        continue
+                    break
+                elif isinstance(attr_value, dict):
+                    for str_item in attr_value.values():
+                        if self.check_string(str_item, url=url):
+                            break
+                    else:
+                        continue
+                    break
+
+    def check_string(self, value, url):
+        if isinstance(value, str) and self.pattern in value:
+            self.print_hit(url, value)
+            return True
+
+    def grep_action_types(self):
+        for workflow in Workflow.select(ignore_errors=True, ignore_migration=True, order_by='id'):
+            for action in workflow.get_all_items():
+                url = action.get_admin_url()
+                if self.pattern in action.key:
+                    self.print_hit(url)
+
+    def grep_field_types(self):
+        for formdef in get_formdefs_of_all_kinds(order_by='id'):
+            for field in formdef.fields or []:
+                url = formdef.get_field_admin_url(field)
+                if self.pattern in field.key:
+                    self.print_hit(url)
+
+    def print_hit(self, url, *args):
+        if self.urls:
+            print(url)
+        else:
+            print(url, *args)
diff --git a/wcs/formdef.py b/wcs/formdef.py
index 441d15321..da08e8aa0 100644
--- a/wcs/formdef.py
+++ b/wcs/formdef.py
@@ -2183,22 +2183,23 @@ def clean_unused_files(publisher, **kwargs):
             pass
 
 
-def get_formdefs_of_all_kinds():
+def get_formdefs_of_all_kinds(**kwargs):
     from wcs.admin.settings import UserFieldsFormDef
     from wcs.blocks import BlockDef
     from wcs.carddef import CardDef
     from wcs.wf.form import FormWorkflowStatusItem
     from wcs.workflows import Workflow
 
-    kwargs = {
+    select_kwargs = {
         'ignore_errors': True,
         'ignore_migration': True,
     }
+    select_kwargs.update(kwargs)
     formdefs = [UserFieldsFormDef()]
-    formdefs += FormDef.select(**kwargs)
-    formdefs += BlockDef.select(**kwargs)
-    formdefs += CardDef.select(**kwargs)
-    for workflow in Workflow.select(**kwargs):
+    formdefs += FormDef.select(**select_kwargs)
+    formdefs += BlockDef.select(**select_kwargs)
+    formdefs += CardDef.select(**select_kwargs)
+    for workflow in Workflow.select(**select_kwargs):
         for status in workflow.possible_status:
             for item in status.items:
                 if isinstance(item, FormWorkflowStatusItem) and item.formdef:
-- 
2.38.1

