registrar_test.py

import os
import csv

import pytest
import psycopg2
import paramiko
from dotenv import load_dotenv
from xml.etree import ElementTree


@pytest.fixture(scope="session")
def connection():
    load_dotenv(dotenv_path='../env/emg_db.env')

    connect_args = dict(
        dbname=os.environ['DB_NAME'],
        user=os.environ['DB_USER'],
        password=f"\"{os.environ['DB_PW']}\"",
        host='docker',
        port=os.environ['DB_PORT'],
    )

    with psycopg2.connect(**connect_args) as connection:
        yield connection


@pytest.fixture
def identifiers():
    with open('./product_list.csv') as f:
        yield csv.reader(f)


@pytest.fixture
def sftp_connection():
    username = os.environ['sftp_users_emg'].split(':')[0]
    password = os.environ['sftp_users_emg'].split(':')[1]
    transport = paramiko.Transport(('docker', 2223))
    transport.connect(username=username, password=password)
    with paramiko.SFTPClient.from_transport(transport) as sftp:
        yield sftp

def query_eo_object(connection, eo_id):
    query = f"SELECT identifier FROM coverages_eoobject WHERE identifier = '{eo_id}';"
    with connection.cursor() as cursor:
        cursor.execute(query)
        return cursor.fetchone()[0]


def test_db_name(connection, identifiers):
    for row in identifiers:
        identifier = row[0].split('/')[4]
        coverage_id = query_eo_object(connection, identifier)
        db_name = connection.get_dsn_parameters()["dbname"]

        assert coverage_id == identifier
        assert db_name == os.environ['DB_NAME']


def compare_links(sftp, product_xml, product):

    report= sftp.file("data/to/panda/%s" % product_xml)


    xml_file= report.read()
    root = ElementTree.fromstring(xml_file.decode('utf-8').strip())

    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, product)
    expected_wcs_link = '%s"%s"' % (wcs_capabilities, product)
    
    assert expected_wms_link.replace('&amp&', '&') == wms_link

    assert expected_wcs_link.replace('&amp&', '&') == wcs_link

def test_reporting(sftp_connection, identifiers):
    report_list = sftp_connection.listdir('data/to/panda/')
    assert len(report_list) > 0
    for item in report_list:
        for row in identifiers:
            identifier = row[0].split('/')[4]

            if identifier in item:
                compare_links(sftp_connection, item, identifier)