import requests import pytest import csv import subprocess import json 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://docker: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 registerd products assert len(wms_items) > 0 def test_wcs(identifiers): wcs_items = get_requests('http://docker: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 succsess assert wcs_response.status_code == 200 assert srcband.Checksum() != None assert srcband.Checksum() > 0 # check if there are registerd products assert len(wcs_items) > 0