Produits Entr'ouvert » w.c.s.

from wcs.mail_templates import MailTemplate

from wcs.wscalls import NamedWsCall

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()

# 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)

def get_formdefs_of_all_kinds(**kwargs):

    select_kwargs = {

    select_kwargs.update(kwargs)

    formdefs += FormDef.select(**select_kwargs)

    formdefs += BlockDef.select(**select_kwargs)

    formdefs += CardDef.select(**select_kwargs)

    for workflow in Workflow.select(**select_kwargs):