diff --git a/testing/test_renderer.py b/testing/test_renderer.py new file mode 100644 index 0000000000000000000000000000000000000000..10a6c440e13fce0cb44f522b5db03f5455089f1d --- /dev/null +++ b/testing/test_renderer.py @@ -0,0 +1,96 @@ + +import requests +import pytest +import csv +import re +from xml.etree import ElementTree +from osgeo import gdal + +from urllib.parse import unquote + + +@pytest.fixture +def identifiers(): + with open('./product_list.csv') as f: + yield csv.reader(f) + + +def get_requests(url, service_list, service, request): + response = requests.get(url=url) + catalog = ElementTree.fromstring(response.content) + entries = catalog.findall('{http://www.w3.org/2005/Atom}entry') + + for entry in entries: + + offers = entry.findall( + '{http://www.opengis.net/owc/1.0}offering[@code="http://www.opengis.net/spec/owc-atom/1.0/req/%s"]' + % service) + for offer in offers: + services = offer.findall( + '{http://www.opengis.net/owc/1.0}operation[@code="%s"]' + % request) + if len(services) > 0: + service_list.append(services[0].get('href')) + + return service_list + + +def test_renderer(identifiers): + wms_items = get_requests( + 'http://127.0.0.1:81/opensearch/collections/Emergency/atom/', + [], 'wms', 'GetMap') + + for row in identifiers: + identifier = row[0].split('/')[4] + for item in wms_items: + if identifier in unquote(unquote(item)): + wms_response = requests.get(url=item) + + # wms succsess + assert wms_response.status_code == 200 + + # check if there are registered products + assert len(wms_items) > 0 + + # spell check test: + if len(wms_items) > 0: + + spelled_item = wms_items[0] + spelled_request = re.sub( + 'layers=.*?&format', 'layers=wrong_layer&format', spelled_item) + spelled_response = requests.get(url=spelled_request) + assert spelled_response.status_code == 400 + + xml_response = ElementTree.fromstring(spelled_response.content) + entry = xml_response.findall( + '{http://www.opengis.net/ogc}ServiceException')[0].text + + assert entry == "Layer 'wrong_layer' does not exist" + + +def test_wcs(identifiers): + wcs_items = get_requests( + 'http://127.0.0.1:81/opensearch/collections/Emergency/atom/', + [], 'wcs', 'GetCoverage') + + for row in identifiers: + identifier = row[0].split('/')[4] + for item in wcs_items: + if identifier in unquote(unquote(item)): + + wcs_response = requests.get( + url=item + '&scalesize=x(50),y(50)') + + data = wcs_response.content + + with open('temp.tif', 'wb') as f: + f.write(data) + + image = gdal.Open('temp.tif', gdal.GA_ReadOnly) + srcband = image.GetRasterBand(1) + # wcs success + assert wcs_response.status_code == 200 + assert srcband.Checksum() is not None + assert srcband.Checksum() > 0 + # check if there are registered products + assert len(wcs_items) > 0