<commit_before>from owslib.wfs import WebFeatureService

from GeoHealthCheck.probe import Probe
from GeoHealthCheck.result import Result


class WFS3Drilldown(Probe):
    """
    Probe for WFS3 endpoint "drilldown": starting
    with top endpoint: get Collections and do
    GetItems on them etc. Using OWSLib.WebFeatureService.

    TODO: needs finalization.
    """

    NAME = 'WFS3 Drilldown'
    DESCRIPTION = 'Traverses a OGC WFS3 (REST) API endpoint by drilling down'
    RESOURCE_TYPE = 'OGC:WFS3'

    REQUEST_METHOD = 'GET'

    # PARAM_DEFS = {
    #     'drilldown_level': {
    #         'type': 'string',
    #         'description': 'How heavy the drilldown should be.',
    #         'default': 'minor',
    #         'required': True,
    #         'range': ['minor', 'moderate', 'full']
    #     }
    # }
    """Param defs"""

    def __init__(self):
        Probe.__init__(self)

    def perform_request(self):
        """
        Perform the drilldown.
        See https://github.com/geopython/OWSLib/blob/
        master/tests/doctests/wfs3_GeoServerCapabilities.txt
        """
        wfs3 = None
        collections = None

        # 1. Test top endpoints existence
        result = Result(True, 'Test Top Endpoints')
        result.start()
        try:
            wfs3 = WebFeatureService(self._resource.url, version='3.0')
            wfs3.conformance()
            collections = wfs3.collections()
        except Exception as err:
            result.set(False, str(err))

        result.stop()
        self.result.add_result(result)

        # 2. Test layers
        # TODO: use parameters to work on less/more drilling
        # "full" could be all layers.
        result = Result(True, 'Test Collections')
        result.start()
        coll_name = ''
        try:
            for collection in collections:
                coll_name = collection['name']
                coll_name = coll_name.encode('utf-8')

                try:
                    items = wfs3.collection_items(coll_name, limit=2)
                except Exception as e:
                    msg = 'GetItems %s: OWSLib err: %s ' % (str(e), coll_name)
                    result = self.add_result(result,
                                             False, msg, 'Test GetItems')
                    continue

                features = items.get('features', None)
                if not features:
                    msg = 'GetItems %s: No features attr' % coll_name
                    result = self.add_result(result,
                                             False, msg, 'Test GetItems')
                    continue

                if len(items['features']) > 0:

                    fid = items['features'][0]['id']
                    try:
                        item = wfs3.collection_item(coll_name, fid)
                    except Exception as e:
                        msg = 'GetItem %s: OWSLib err: %s' \
                              % (str(e), coll_name)
                        result = self.add_result(result,
                                                 False, msg, 'Test GetItem')
                        continue

                    for attr in ['id', 'links', 'properties', 'type']:
                        val = item.get(attr, None)
                        if not val:
                            msg = '%s:%s no attr=%s' \
                                  % (coll_name, str(fid), attr)
                            result = self.add_result(
                                result, False, msg, 'Test GetItem')
                            continue

        except Exception as err:
            result.set(False, 'Collection err: %s : e=%s'
                       % (coll_name, str(err)))

        result.stop()<commit_msg>#221 refine WFS3 drilldown, basic and full option<commit_after>import requests
from owslib.wfs import WebFeatureService

from GeoHealthCheck.probe import Probe
from GeoHealthCheck.result import Result


class WFS3Drilldown(Probe):
    """
    Probe for WFS3 endpoint "drilldown": starting
    with top endpoint: get Collections and do
    GetItems on them etc. Using OWSLib.WebFeatureService.

    TODO: needs finalization.
    """

    NAME = 'WFS3 Drilldown'
    DESCRIPTION = 'Traverses a OGC WFS3 (REST) API endpoint by drilling down'
    RESOURCE_TYPE = 'OGC:WFS3'

    REQUEST_METHOD = 'GET'

    PARAM_DEFS = {
        'drilldown_level': {
            'type': 'string',
            'description': 'How heavy the drilldown should be.\
                            basic: test presence endpoints, \
                            full: go through collections, fetch items',
            'default': 'basic',
            'required': True,
            'range': ['basic', 'full']
        }
    }
    """Param defs"""

    def __init__(self):
        Probe.__init__(self)

    def perform_request(self):
        """
        Perform the drilldown.
        See https://github.com/geopython/OWSLib/blob/
        master/tests/doctests/wfs3_GeoServerCapabilities.txt
        """
        wfs3 = None
        collections = None

        # 1. Test top endpoints existence
        result = Result(True, 'Test Top Endpoints')
        result.start()
        try:
            wfs3 = WebFeatureService(self._resource.url, version='3.0')
            wfs3.conformance()

            # TODO: OWSLib 0.17.1 has no call to '/api yet.
            url = wfs3._build_url('api')
            api = requests.get(url).json()
            for attr in ['components', 'paths', 'openapi']:
                val = api.get(attr, None)
                if val is None:
                    msg = '/api: missing attr: %s' % attr
                    result = self.add_result(
                        result, False, msg, 'Test Collection')
                    continue

            collections = wfs3.collections()
        except Exception as err:
            result.set(False, str(err))

        result.stop()
        self.result.add_result(result)

        if self._parameters['drilldown_level'] == 'basic':
            return

        # ASSERTION: will do full drilldown from here

        # 2. Test layers
        # TODO: use parameters to work on less/more drilling
        # "full" could be all layers.
        result = Result(True, 'Test Collections')
        result.start()
        coll_name = ''
        try:
            for collection in collections:
                coll_name = collection['name']
                coll_name = coll_name.encode('utf-8')

                try:
                    coll = wfs3.collection(coll_name)

                    # TODO: Maybe also add crs
                    for attr in ['name', 'links']:
                        val = coll.get(attr, None)
                        if val is None:
                            msg = '%s: missing attr: %s' \
                                  % (coll_name, attr)
                            result = self.add_result(
                                result, False, msg, 'Test Collection')
                            continue
                except Exception as e:
                    msg = 'GetCollection %s: OWSLib err: %s ' \
                          % (str(e), coll_name)
                    result = self.add_result(result,
                                             False, msg, 'Test GetCollection')
                    continue

                try:
                    items = wfs3.collection_items(coll_name, limit=1)
                except Exception as e:
                    msg = 'GetItems %s: OWSLib err: %s ' % (str(e), coll_name)
                    result = self.add_result(result,
                                             False, msg, 'Test GetItems')
                    continue

                features = items.get('features', None)
                if features is None:
                    msg = 'GetItems %s: No features attr' % coll_name
                    result = self.add_result(result,
                                             False, msg, 'Test GetItems')
                    continue

                type = items.get('type', '')
                if type != 'FeatureCollection':
                    msg = '%s:%s type not FeatureCollection: %s' \
                          % (coll_name, type, val)
                    result = self.add_result(
                        result, False, msg, 'Test GetItems')
                    continue

                if len(items['features']) > 0:

                    fid = items['features'][0]['id']
                    try:
                        item = wfs3.collection_item(coll_name, fid)
                    except Exception as e:
                        msg = 'GetItem %s: OWSLib err: %s' \
                              % (str(e), coll_name)
                        result = self.add_result(result,
                                                 False, msg, 'Test GetItem')
                        continue

                    for attr in \
                            ['id', 'links', 'properties', 'geometry', 'type']:
                        val = item.get(attr, None)
                        if val is None:
                            msg = '%s:%s missing attr: %s' \
                                  % (coll_name, str(fid), attr)
                            result = self.add_result(
                                result, False, msg, 'Test GetItem')
                            continue

                        if attr == 'type' and val != 'Feature':
                            msg = '%s:%s type not Feature: %s' \
                                  % (coll_name, str(fid), val)
                            result = self.add_result(
                                result, False, msg, 'Test GetItem')
                            continue

        except Exception as err:
            result.set(False, 'Collection err: %s : e=%s'
                       % (coll_name, str(err)))

        result.stop()
{'fraction_non_alphanumeric': 0.09629404617253949, 'fraction_numerical': 0.005062778452814905, 'mean_word_length': 1.2185534591194969, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7302670', 'n_tokens_mistral': 2629, 'n_tokens_neox': 2466, 'n_words': 687}