How to use OWSLib - 10 common examples

def test_wps_request7():
    # Process input/ouutput arguments
    processid = "wordcount"
    textdoc = ComplexDataInput("http://emu.readthedocs.org/en/latest/index.html")
    inputs = [("text", textdoc), ]
    outputs = [("output", False)]

    # Build XML request for WPS process execution, sync request
    execution = WPSExecution()
    requestElement = execution.buildRequest(processid, inputs, output=outputs, mode=SYNC, lineage=False)
    request = etree.tostring(requestElement)

    # Compare to cached XML request
    _request = open(resource_file('wps_EmuExecuteRequest7.xml'), 'rb').read()
    print(request)
    assert compare_xml(request, _request) is True

("STATISTICS", "MEAN"),
              ("STATISTICS", "MINIMUM"),
              ("STATISTICS", "MAXIMUM"),
              ("STATISTICS", "WEIGHT_SUM"),
              ("STATISTICS", "VARIANCE"),
              ("STATISTICS", "STD_DEV"),
              ("STATISTICS", "COUNT"),
              ("GROUP_BY", "STATISTIC"),
              ("SUMMARIZE_TIMESTEP", "false"),
              ("SUMMARIZE_FEATURE_ATTRIBUTE", "false"),
              ("FEATURE_COLLECTION", featureCollection)]
    output = "OUTPUT"
    # build XML request for WPS process execution
    execution = WPSExecution()
    requestElement = execution.buildRequest(processid, inputs, output=[(output, True)])
    request = etree.tostring(requestElement)
    # Compare to cached XML request
    _request = open(resource_file('wps_USGSExecuteRequest3.xml'), 'rb').read()
    assert compare_xml(request, _request) is True

("STATISTICS", "MAXIMUM"),
              ("STATISTICS", "WEIGHT_SUM"),
              ("STATISTICS", "VARIANCE"),
              ("STATISTICS", "STD_DEV"),
              ("STATISTICS", "COUNT"),
              ("GROUP_BY", "STATISTIC"),
              ("SUMMARIZE_TIMESTEP", "true"),
              ("SUMMARIZE_FEATURE_ATTRIBUTE", "true"),
              ("FEATURE_COLLECTION", featureCollection)
              ]
    output = "OUTPUT"

    # build XML request for WPS process execution
    execution = WPSExecution()
    requestElement = execution.buildRequest(processid, inputs, output=[(output, True)])
    request = etree.tostring(requestElement)

    # Compare to cached XML request
    _request = open(resource_file('wps_USGSExecuteRequest4.xml'), 'rb').read()
    assert compare_xml(request, _request) is True

def test_wps_process_with_invalid_identifer():
    p = Process(etree.Element('invalid'))
    assert repr(p) == ''
    assert str(p) == 'WPS Process: , title='

("STATISTICS", "MAXIMUM"),
              ("STATISTICS", "WEIGHT_SUM"),
              ("STATISTICS", "VARIANCE"),
              ("STATISTICS", "STD_DEV"),
              ("STATISTICS", "COUNT"),
              ("GROUP_BY", "STATISTIC"),
              ("SUMMARIZE_TIMESTEP", "true"),
              ("SUMMARIZE_FEATURE_ATTRIBUTE", "true"),
              ("FEATURE_COLLECTION", featureCollection)
              ]
    output = "OUTPUT"

    # build XML request for WPS process execution
    execution = WPSExecution()
    requestElement = execution.buildRequest(processid, inputs, output=[(output, True)])
    request = etree.tostring(requestElement)

    # Compare to cached XML request
    _request = open(resource_file('wps_USGSExecuteRequest2.xml'), 'rb').read()
    assert compare_xml(request, _request) is True

def test_wms_getmap():
    # Lastly, test the getcapabilities and getmap methods
    wms = WebMapService(SERVICE_URL, version='1.1.1')

def test_wms_getmap_111_service_exception():
    """GetMap 1.1.1 ServiceException for an invalid CRS"""
    wms = WebMapService(SERVICE_URL, version='1.1.1')
    try:
        wms.getmap(
            layers=['nexrad_base_reflect'],
            styles=['default'],
            srs='EPSG:4328',
            bbox=(-126, 24, -66, 50),
            size=(250, 250),
            format='image/jpeg',
            transparent=True)
    except ServiceException as e:
        assert "msWMSLoadGetMapParams(): WMS server error. Invalid SRS given : SRS must be valid for all requested layers." in str(e)  # noqa
    else:
        assert False

def setUp(self):
        super(WmsServiceHandlerTestCase, self).setUp()

        self.phony_url = ("http://a-really-long-and-fake-name-here-so-that-"
                          "we-use-it-in-tests")
        self.phony_title = "a generic title"
        self.phony_version = "some.version"
        self.phony_layer_name = "phony_name"
        self.phony_keywords = ["first", "second"]
        mock_parsed_wms = mock.MagicMock(OwsWebMapService).return_value
        (url, mock_parsed_wms) = mock.MagicMock(WebMapService,
                                                return_value=(self.phony_url,
                                                              mock_parsed_wms)).return_value
        mock_parsed_wms.url = self.phony_url
        mock_parsed_wms.identification.title = self.phony_title
        mock_parsed_wms.identification.version = self.phony_version
        mock_parsed_wms.identification.keywords = self.phony_keywords
        mock_layer_meta = mock.MagicMock(ContentMetadata)
        mock_layer_meta.name = self.phony_layer_name
        mock_layer_meta.children = []
        mock_layer_meta.crsOptions = ["EPSG:3857"]
        mock_parsed_wms.contents = {
            mock_layer_meta.name: mock_layer_meta,
        }
        self.parsed_wms = mock_parsed_wms

def test_getmap_130_national_map():
    """National Map"""
    # TODO: use flaky tests or fix it: https://pypi.python.org/pypi/pytest-ignore-flaky
    url = SERVICE_URL_NATIONAL_MAP
    wms = WebMapService(url, version='1.3.0')
    rsp = wms.getmap(
        layers=['3'],
        styles=['default'],
        srs='CRS:84',
        bbox=(-176.646, 17.7016, -64.8017, 71.2854),
        size=(500, 300),
        format='image/png',
        transparent=True)
    assert type(rsp) is ResponseWrapper
    assert "service=WMS" in wms.request
    assert "version=1.3.0" in wms.request
    assert "request=GetMap" in wms.request
    assert "layers=3" in wms.request
    assert "styles=default" in wms.request
    assert "crs=CRS%3A84" in wms.request
    assert "box=-176.646%2C17.7016%2C-64.8017%2C71.2854" in wms.request

def get_md_resource(file_path):
    """Read the file and parse into an XML tree.

    Parameters
    ----------
    file_path : str
        Path of the file to read.

    Returns
    -------
    etree.ElementTree
        XML tree of the resource on disk.

    """
    namespaces = Namespaces().get_namespaces(keys=('gmd', 'gmi'))

    with io.open(file_path, mode='r', encoding='utf-8') as f:
        data = f.read().encode('utf-8')
        data = etree.fromstring(data)
        mdelem = data.find('.//' + util.nspath_eval(
            'gmd:MD_Metadata', namespaces)) or data.find(
            './/' + util.nspath_eval('gmi:MI_Metadata', namespaces))

        if mdelem is None and data.tag == '{http://www.isotc211.org/2005/gmd}MD_Metadata':
            mdelem = data

    return mdelem