class BoringSearch(AbstractSearch):
    """Search class to retrieve information about boreholes (Boring)."""

    def __init__(self, objecttype=Boring):
        """Initialisation.

        Parameters
        ----------
        objecttype : subclass of pydov.types.abstract.AbstractDovType
            Reference to a class representing the Boring type.
            Optional: defaults to the Boring type containing the fields
            described in the documentation.

        """
        super(BoringSearch, self).__init__('dov-pub:Boringen', objecttype) class GrondwaterFilterSearch(AbstractSearch):
    """Search class to retrieve information about groundwater screens
    (GrondwaterFilter).
    """

    def __init__(self, objecttype=GrondwaterFilter):
        """Initialisation.

        Parameters
        ----------
        objecttype : subclass of pydov.types.abstract.AbstractDovType
            Reference to a class representing the GrondwaterFilter type.
            Optional: defaults to the GrondwaterFilter type containing the
            fields described in the documentation.

        """
        super(GrondwaterFilterSearch,
              self).__init__('gw_meetnetten:meetnetten', objecttype)

    def search(self, location=None, query=None, sort_by=None,
               return_fields=None, max_features=None):
        """Search for objects of this type. Provide `location` and/or
        `query` and/or `max_features`.
        When `return_fields` is None, all fields are returned.

        Excludes 'empty' filters (i.e. Putten without Filters) by extending
        the `query` with a not-null check on pkey_filter.

        Parameters
        ----------
        location : pydov.util.location.AbstractLocationFilter or \
                   owslib.fes.BinaryLogicOpType<AbstractLocationFilter> or \
                   owslib.fes.UnaryLogicOpType<AbstractLocationFilter>
            Location filter limiting the features to retrieve. Can either be a
            single instance of a subclass of AbstractLocationFilter, or a
            combination using And, Or, Not of AbstractLocationFilters.
        query : owslib.fes.OgcExpression
            OGC filter expression to use for searching. This can contain any
            combination of filter elements defined in owslib.fes. The query
            should use the fields provided in `get_fields()`. Note that not
            all fields are currently supported as a search parameter.
        sort_by : owslib.fes.SortBy, optional
            List of properties to sort by.
        return_fields : list<str> or tuple<str> or set<str>
            A list of fields to be returned in the output data. This should
            be a subset of the fields provided in `get_fields()`. Note that
            not all fields are currently supported as return fields.
        max_features : int
            Limit the maximum number of features to request.

        Returns
        -------
        pandas.core.frame.DataFrame
            DataFrame containing the output of the search query.

        Raises
        ------
        pydov.util.errors.InvalidSearchParameterError
            When not one of `location` or `query` or `max_features` is
            provided.

        pydov.util.errors.InvalidFieldError
            When at least one of the fields in `return_fields` is unknown.

            When a field that is only accessible as return field is used as
            a query parameter.

            When a field that can only be used as a query parameter is used as
            a return field.

        pydov.util.errors.FeatureOverflowError
            When the number of features to be returned is equal to the
            maxFeatures limit of the WFS server.

        AttributeError
            When the argument supplied as return_fields is not a list,
            tuple or set.

        NotImplementedError
            This is an abstract method that should be implemented in a
            subclass.

        """
        self._pre_search_validation(location, query, sort_by, return_fields,
                                    max_features)

        exclude_empty_filters = Not([PropertyIsNull(
                                     propertyname='pkey_filter')])

        if query is not None:
            query = And([query, exclude_empty_filters])
        else:
            query = exclude_empty_filters

        fts = self._search(location=location, query=query, sort_by=sort_by,
                           return_fields=return_fields,
                           max_features=max_features)

        gw_filters = self._type.from_wfs(fts, self._wfs_namespace)

        df = pd.DataFrame(
            data=self._type.to_df_array(gw_filters, return_fields),
            columns=self._type.get_field_names(return_fields))

        return df class GzipTextFileCache(AbstractFileCache):
    """Class for GZipped text caching of downloaded XML files from DOV."""

    def _get_filepath(self, datatype, key):
        return os.path.join(self.cachedir, datatype, key + '.xml.gz')

    def _get_type_key_from_path(self, path):
        key = os.path.basename(path).rstrip('.xml.gz')
        datatype = os.path.dirname(path).split()[-1]
        return datatype, key

    def _save(self, datatype, key, content):
        filepath = self._get_filepath(datatype, key)
        folder = os.path.dirname(filepath)

        if not os.path.exists(folder):
            os.makedirs(folder)

        with gzip.open(filepath, 'wb') as f:
            f.write(content)

    def _load(self, datatype, key):
        filepath = self._get_filepath(datatype, key)
        with gzip.open(filepath, 'rb') as f:
            return f.read().decode('utf-8') def build_dov_url(path):
    """Build the DOV url consisting of the fixed DOV base url, appended with
    the given path.

    Returns
    -------
    str
        The absolute DOV url.

    """
    if 'PYDOV_BASE_URL' in os.environ:
        base_url = os.environ['PYDOV_BASE_URL']
    else:
        base_url = 'https://www.dov.vlaanderen.be/'

    return base_url + path.lstrip('/') class XmlFetchWarning(DOVWarning):
    """Emitted when an XML document fails to be fetched from the DOV
    webservice, resulting in an incomplete dataframe.""" class XmlStaleWarning(DOVWarning):
    """Emitted when an XML document fails to be fetched from the DOV
    webservice and an older stale version is used from the cache, resulting
    in an out-of-date dataframe.""" class XsdFetchWarning(DOVWarning):
    """Emitted when an XSD document fails to be fetched from the DOV
    webservice, resulting in the fields metadata to be incomplete.""" class Hooks(list):
    """Runtime representation of registered pydov hooks, i.e. a list of
    instances of AbstractReadHook and/or AbstractInjectHook."""

    def get_read_hooks(self):
        """Get the registered read hooks (i.e. hooks that are subclasses of
        AbstractReadHook), in the order they are defined in the list.

        Returns
        -------
        tuple of AbstractReadHook
            A tuple with the registered read hooks.

        """
        return (h for h in self if isinstance(h, AbstractReadHook))

    def get_inject_hooks(self):
        """Get the registered inject hooks (i.e. hooks that are subclasses of
        AbstractInjectHook), in the order they are defined in the list.

        Returns
        -------
        tuple of AbstractInjectHook
            A tuple with the registered inject hooks.

        """
        return (h for h in self if isinstance(h, AbstractInjectHook)) class ServiceCheck:

    is_service_ok = None

    @staticmethod
    def service_ok(timeout=5):
        """Check whether DOV services are accessible.

        Used to skip online tests when the service is unavailable or
        unreachable.

        Parameters
        ----------
        timeout : int, optional
            Timeout in seconds. Defaults to 5.

        Returns
        -------
        bool
            True if the DOV services are reachable, False otherwise.

        """
        def check_url(url, timeout):
            try:
                ok = pydov.session.head(
                    url, allow_redirects=True, timeout=timeout).ok
            except requests.exceptions.ReadTimeout:
                ok = False
            except requests.exceptions.ConnectTimeout:
                ok = False
            except Exception:
                ok = False
            return ok

        if ServiceCheck.is_service_ok is None:
            ServiceCheck.is_service_ok = (
                check_url(build_dov_url('geoserver'), timeout) and
                check_url(build_dov_url('geonetwork'), timeout) and
                check_url(
                    build_dov_url('xdov/schema/latest/xsd/kern/dov.xsd'),
                    timeout))

        return ServiceCheck.is_service_ok class HookCounter(AbstractReadHook, AbstractInjectHook):
    """Hook implementation for testing purposes, counting all event calls."""

    def __init__(self):
        self.count_meta_received = 0
        self.count_inject_meta_response = 0
        self.count_wfs_search_init = 0
        self.count_wfs_search_result = 0
        self.count_wfs_search_result_received = 0
        self.count_inject_wfs_getfeature_response = 0
        self.count_xml_received = 0
        self.count_inject_xml_response = 0
        self.count_xml_cache_hit = 0
        self.count_xml_stale_hit = 0
        self.count_xml_fetch_error = 0
        self.count_xml_downloaded = 0

    def meta_received(self, url, response):
        self.count_meta_received += 1

    def inject_meta_response(self, url):
        self.count_inject_meta_response += 1

    def wfs_search_init(self, typename):
        self.count_wfs_search_init += 1

    def wfs_search_result(self, number_of_results):
        self.count_wfs_search_result += 1

    def wfs_search_result_received(self, query, features):
        self.count_wfs_search_result_received += 1

    def inject_wfs_getfeature_response(self, query):
        self.count_inject_wfs_getfeature_response += 1

    def xml_received(self, pkey_object, xml):
        self.count_xml_received += 1

    def inject_xml_response(self, pkey_object):
        self.count_inject_xml_response += 1

    def xml_cache_hit(self, pkey_object):
        self.count_xml_cache_hit += 1

    def xml_stale_hit(self, pkey_object):
        self.count_xml_stale_hit += 1

    def xml_downloaded(self, pkey_object):
        self.count_xml_downloaded += 1

    def xml_fetch_error(self, pkey_object):
        self.count_xml_fetch_error += 1
import datetime
import gzip
import os
import sys
import tempfile
import time
import numpy as np
import pytest
import pydov
from importlib import reload
from subprocess import Popen
from owslib.fes import PropertyIsEqualTo
from pydov.search.boring import BoringSearch
from pydov.search.grondwaterfilter import GrondwaterFilterSearch
from pydov.util.caching import GzipTextFileCache
from pydov.util.dovutil import build_dov_url
from pydov.util.errors import XmlFetchWarning, XmlStaleWarning, XsdFetchWarning
from pydov.util.hooks import Hooks
from tests.abstract import ServiceCheck
from tests.test_util_hooks import HookCounter




@pytest.fixture(scope="module", autouse=True)
def dov_proxy_no_xdov():
    """Fixture to start the DOV proxy and set PYDOV_BASE_URL to route
    traffic through it.

    The DOV proxy behaves as the XDOV server would be unavailable.

    """
    process = Popen([sys.executable,
                     os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                  'stub', 'dov_proxy.py'),
                     '--dov-base-url', build_dov_url('/'),
                     '--no-xdov'])
    time.sleep(2)

    orig_base_url = os.environ.get('PYDOV_BASE_URL', None)
    os.environ['PYDOV_BASE_URL'] = 'http://localhost:1337/'

    yield

    if orig_base_url is not None:
        os.environ['PYDOV_BASE_URL'] = orig_base_url
    else:
        del(os.environ['PYDOV_BASE_URL'])

    process.terminate()
    process.communicate()


@pytest.fixture(scope="module", autouse=True)
def reload_modules(dov_proxy_no_xdov):
    """Reload the boring and grondwaterfilter modules after setting
    PYDOV_BASE_URL.

    These need to be reloaded because they use the PYDOV_BASE_URL at import
    time to set the location of XSD schemas.

    Parameters
    ----------
    dov_proxy_no_xdov : pytest.fixture
        Fixture starting the DOV proxy and setting PYDOV_BASE_URL accordingly.
    """
    reload(pydov.types.boring)
    reload(pydov.types.grondwaterfilter)

    yield

    reload(pydov.types.boring)
    reload(pydov.types.grondwaterfilter)


@pytest.fixture(scope="function", autouse=True)
def reset_cache(dov_proxy_no_xdov):
    """Reset the cache to a temporary folder to remove influence from other
    tests.

    The cache needs to be reset after setting the PYDOV_BASE_URL variable
    because at initialisation this URL is used to construct a regex for
    determining the datatype of an XML request.

    Parameters
    ----------
    dov_proxy_no_xdov : pytest.fixture
        Fixture starting the DOV proxy and setting PYDOV_BASE_URL accordingly.
    """
    gziptext_cache = GzipTextFileCache(
        cachedir=os.path.join(tempfile.gettempdir(), 'pydov_tests_error'),
        max_age=datetime.timedelta(seconds=0.1))
    gziptext_cache.remove()

    orig_cache = pydov.cache
    pydov.cache = gziptext_cache

    yield

    gziptext_cache.remove()
    pydov.cache = orig_cache


@pytest.fixture
def test_hook_count():
    """PyTest fixture temporarily disabling default hooks and installing
    HookCounter."""
    orig_hooks = pydov.hooks

    pydov.hooks = Hooks(
        (HookCounter(),)
    )
    yield

    pydov.hooks = orig_hooks


class TestNoXDOV(object):
    """Class grouping tests related failing DOV services."""

    @pytest.mark.online
    @pytest.mark.skipif(not ServiceCheck.service_ok(),
                        reason="DOV service is unreachable")
    def test_do_not_cache_error(self):
        """Test whether the 404 error page does not end up being cached."""

        bs = BoringSearch(objecttype=pydov.types.boring.Boring)