EOX GitLab Instance

Skip to content
Snippets Groups Projects
renderer_test.py 2.29 KiB
Newer Older
Mussab Abdalla's avatar
Mussab Abdalla committed

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')
Mussab Abdalla's avatar
Mussab Abdalla committed
    
    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
Mussab Abdalla's avatar
Mussab Abdalla committed

def test_wcs(identifiers):
    wcs_items = get_requests('http://docker:81/opensearch/collections/Emergency/atom/', [], 'wcs', 'GetCoverage')
Mussab Abdalla's avatar
Mussab Abdalla committed
    
    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