check_v1_parsing.py

import os
import django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", 'ods.core.settings')
django.setup()

from _pytest.monkeypatch import MonkeyPatch
from django.test import RequestFactory
from django.core.handlers.base import BaseHandler
from ods.core.utils import platform_utils

import pyparsing

from ods.explore.backends.query_parser import QueryParser

original_query_parser_es_parse = QueryParser.parse
original_pyparsing_forward_parse_string = pyparsing.Forward.parseString


class MockedQuery:

    def __init__(self, domain_id):
        self.factory = RequestFactory()

        self.domain_id = domain_id

        self.handler = BaseHandler()
        self.handler.load_middleware()
        self._res = None
        self._error = None

    def _check_query(self, version, params):
        self._res = None
        self._error = None
        with MonkeyPatch().context() as m:
            def _mocked_pyparsing_forward_parse_string(this, query, *args, **kwargs):
                kwargs["parseAll"] = version == "new"
                try:
                    return original_pyparsing_forward_parse_string(this, query, *args, **kwargs)
                except Exception as e:
                    self._error = e

            def _mocked_query_parser_es_parse(this, query):
                from pyparsing import ParseException
                if version == "old":
                    if isinstance(query, list):
                        query = ' AND '.join(['(%s)' % q for q in query if q])

                    if not query.strip():
                        return
                    try:
                        self._res = this._parser(query)[0]
                    except ParseException as e:
                        self._error = e
                else:  # new version
                    if not isinstance(query, list):
                        query = [query]
                    try:
                        parsed_queries = [this._parser(q)[0] for q in query if q and q.strip()]
                        self._res = this.searchAndExpr(parsed_queries)
                    except ParseException as e:
                        self._error = e

            m.setattr(
                'ods.explore.backends.query_parser.QueryParser.parse',
                _mocked_query_parser_es_parse)
            m.setattr(
                'pyparsing.Forward.parseString',
                _mocked_pyparsing_forward_parse_string)

            request = self.factory.get('/api/records/1.0/search',
                                       params,
                                       HTTP_HOST="{}.{}".format(self.domain_id, platform_utils.full_suffix()))

            self.handler.get_response(request)
            return self._res, self._error



mq = MockedQuery("testdomain")
params = {"dataset": "arbres",
          "q": '#exact(espece,"excelsior"))) AND #exact(espece, "truc")'}
print(mq._check_query("old", params))
print(mq._check_query("new", params))

# does not work
params = {"dataset": "ods-api-monitoring",
          "source": "monitoring",
          "q": 'timestamp>#now(days=-7))'}
print(mq._check_query("old", params))
print(mq._check_query("new", params))