registrar_test.py

import psycopg2
import os
import csv
from xml.etree import ElementTree


with open('./env/emg_db.env', 'r') as f:
    env = dict(
        line.split('=', 1)
        for line in f
    )
database= env['DB_NAME'].replace('\n','')
port = env['DB_PORT'].replace('\n','')
host = env['DB_HOST'].replace('\n','')
database_password= env['DB_PW'].replace('\n','')
database_user = env['DB_USER'].replace('\n','')


def connect_to_db(eo_id):
    global db_name, coverage_id
    connection= None
    try:
        connection = psycopg2.connect(dbname=database, user=database_user, password=database_password, host='docker', port=port)
        cursor = connection.cursor()
        db_name = connection.get_dsn_parameters()["dbname"]
        postgreSQL_select_Query = "SELECT identifier FROM coverages_eoobject WHERE identifier = '%s';" % eo_id
        cursor.execute(postgreSQL_select_Query)
        coverage_id = cursor.fetchone()[0]

    except (Exception, psycopg2.Error) as error :
        print ("Error while connecting to PostgreSQL", error)
    finally:
        #closing database connection.
        if connection:
            cursor.close()
            connection.close()
            print("PostgreSQL connection is closed")
                


def test_db_name(name):
    with open(name, newline='') as csvfile:
        spamreader = csv.reader(csvfile)
        for row in spamreader:
            identifier = row[0].split('/')[4]

            xml_file= "%s.xml"% identifier
            tree = ElementTree.parse(xml_file)
            root = tree.getroot()
            urls = root.findall('{http://www.telespazio.com/CSCDA/CDD/PDAS}URL')
            wms_link = urls[0].find('{http://www.telespazio.com/CSCDA/CDD/PDAS}URL').text
            wcs_link = urls[1].find('{http://www.telespazio.com/CSCDA/CDD/PDAS}URL').text

            wms_capabilities = 'emg.pass.copernicus.eu/ows?service=wms&request=GetCapabilities&amp&cql=identifier='
            wcs_capabilities = 'emg.pass.copernicus.eu/ows?service=wcs&request=GetCapabilities&amp&cql=identifier='

            expected_wms_link = '%s"%s"' % (wms_capabilities, identifier)
            expected_wcs_link = '%s"%s"' % (wcs_capabilities, identifier)
            assert expected_wms_link.replace('&amp&', '&') == wms_link
            assert expected_wcs_link.replace('&amp&', '&') == wcs_link

            connect_to_db(identifier)
            assert coverage_id == identifier
            assert db_name == database