Skip to content

Commit

Permalink
formatted tests and adding missing __init__.py
Browse files Browse the repository at this point in the history
  • Loading branch information
@abodeuis
abodeuis committed Jul 9, 2024
1 parent e0c4263 commit f2ae63e
Show file tree
Hide file tree
Showing 5 changed files with 119 additions and 94 deletions.
Empty file added tests/__init__.py
View file
Empty file.
36 changes: 23 additions & 13 deletions tests/test_cdrhook/test_connector.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,29 +6,39 @@

load_dotenv()


class TestCDRConnector:
system_name = 'ncsa_test'
system_version = '0.1'
token = os.getenv('CDR_TOKEN')
callback_url = 'https://criticalmaas.ncsa.illinois.edu/'
system_name = "ncsa_test"
system_version = "0.1"
token = os.getenv("CDR_TOKEN")
callback_url = "https://criticalmaas.ncsa.illinois.edu/"

def test_constructor(self):
log = init_test_log('TestCDRConnector/test_constructor')
con = CdrConnector(system_name=self.system_name, system_version=self.system_version, token=self.token, callback_url=self.callback_url)

log.info(f'Connector : {con.__dict__}')
log = init_test_log("TestCDRConnector/test_constructor")
con = CdrConnector(
system_name=self.system_name,
system_version=self.system_version,
token=self.token,
callback_url=self.callback_url,
)

log.info(f"Connector : {con.__dict__}")
assert con.system_name == self.system_name
assert con.system_version == self.system_version
assert con.token == self.token
assert str(con.callback_url) == self.callback_url
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_registration(self):
log = init_test_log('TestCDRConnector/test_registration')
con = CdrConnector(system_name=self.system_name, system_version=self.system_version, token=self.token, callback_url=self.callback_url)
log = init_test_log("TestCDRConnector/test_registration")
con = CdrConnector(
system_name=self.system_name,
system_version=self.system_version,
token=self.token,
callback_url=self.callback_url,
)
con.register()
assert con.registration is not None
con.unregister()
assert con.registration is None
log.info('Test passed successfully')

log.info("Test passed successfully")
31 changes: 18 additions & 13 deletions tests/test_cdrhook/test_convert.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,14 @@
import json

import cdrhook.convert as convert
from cdr_schemas.map_results import MapResults
from cdrhook.cdr_endpoint_schemas import AreaExtractionsEndpoint, LegendItemsEndpoint, MetadataEndpoint
from cdrhook.retrieve import validate_endpoint
from tests.utilities import init_test_log
from cdrhook.retrieve import validate_endpoint
# from cdr_schemas.map_results import MapResults
from cdrhook.cdr_endpoint_schemas import (
AreaExtractionsEndpoint,
LegendItemsEndpoint,
MetadataEndpoint,
)

class TestConvert:
# def test_convert_cdr_schema_map_to_cmass_map(self):
Expand All @@ -18,29 +23,29 @@ class TestConvert:
# log.info('Test passed successfully')

def test_convert_cdr_schema_metadata_to_cmass_map_metadata(self):
log = init_test_log('TestConvert/test_convert_cdr_schema_metadata_to_cmass_map_metadata')
with open('tests/data/sample_cog_metadata.json', 'r') as fh:
cdr_metadata = (validate_endpoint(json.load(fh), MetadataEndpoint))
log = init_test_log("TestConvert/test_convert_cdr_schema_metadata_to_cmass_map_metadata")
with open("tests/data/sample_cog_metadata.json", "r") as fh:
cdr_metadata = validate_endpoint(json.load(fh), MetadataEndpoint)
cmass_map_metadata = convert.convert_cdr_schema_metadata_to_cmass_map_metadata(cdr_metadata)
assert cmass_map_metadata
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_convert_cdr_schema_legend_items_to_cmass_legend(self):
log = init_test_log('TestConvert/test_convert_cdr_schema_legend_items_to_cmass_legend')
with open('tests/data/sample_cog_legend.json', 'r') as fh:
log = init_test_log("TestConvert/test_convert_cdr_schema_legend_items_to_cmass_legend")
with open("tests/data/sample_cog_legend.json", "r") as fh:
cdr_legend = []
for map_unit in json.load(fh):
cdr_legend.append(validate_endpoint(map_unit, LegendItemsEndpoint))
cmass_legend = convert.convert_cdr_schema_legend_items_to_cmass_legend(cdr_legend)
assert cmass_legend
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_convert_cdr_schema_area_extraction_to_layout(self):
log = init_test_log('TestConvert/test_convert_cdr_schema_area_extraction_to_layout')
with open('tests/data/sample_cog_area_extraction.json', 'r') as fh:
log = init_test_log("TestConvert/test_convert_cdr_schema_area_extraction_to_layout")
with open("tests/data/sample_cog_area_extraction.json", "r") as fh:
cdr_area_extraction = []
for area in json.load(fh):
cdr_area_extraction.append(validate_endpoint(area, AreaExtractionsEndpoint))
layout = convert.convert_cdr_schema_area_extraction_to_layout(cdr_area_extraction)
assert layout
log.info('Test passed successfully')
log.info("Test passed successfully")
133 changes: 71 additions & 62 deletions tests/test_cdrhook/test_retrieve.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,97 +5,106 @@
import cdrhook.retrieve as rt
from cdrhook.connector import CdrConnector
from tests.utilities import init_test_log
from cdrhook.cdr_endpoint_schemas import AreaExtractionsEndpoint, LegendItemsEndpoint, SystemId, MetadataEndpoint, SystemVersionsEndpoint
from cdr_schemas.map_results import MapResults
from cdr_schemas.feature_results import FeatureResults
from cdrhook.cdr_endpoint_schemas import (
AreaExtractionsEndpoint,
LegendItemsEndpoint,
SystemId,
MetadataEndpoint,
)

class TestRetrieve:
cog_id = '78c274e9575d1ac948d55a55265546d711551cdd5cdd53592c9928d502d50700'
#cog_id = '5a06544690b6611f419f0c6f244776a536ad52915555555555515545c9b1ddb9'
event_id = 'uncharted_0.0.1_f2090dc52547330f2b1f0bc7163f19f730ff107f135f226708cf070f250fcca0'
cog_id = "78c274e9575d1ac948d55a55265546d711551cdd5cdd53592c9928d502d50700"
event_id = "uncharted_0.0.1_f2090dc52547330f2b1f0bc7163f19f730ff107f135f226708cf070f250fcca0"

def setup_class(self):
load_dotenv()
system_name = 'ncsa_test'
system_version = '0.1'
token = os.getenv('CDR_TOKEN')
callback_url = 'https://criticalmaas.ncsa.illinois.edu'
self.con = CdrConnector(system_name=system_name, system_version=system_version, token=token, callback_url=callback_url)
system_name = "ncsa_test"
system_version = "0.1"
token = os.getenv("CDR_TOKEN")
callback_url = "https://criticalmaas.ncsa.illinois.edu"
self.con = CdrConnector(
system_name=system_name,
system_version=system_version,
token=token,
callback_url=callback_url,
)
self.con.register()

def test_retrieve_cog_metadata(self):
log = init_test_log('TestRetrieve/test_retrieve_cog_metadata')
log = init_test_log("TestRetrieve/test_retrieve_cog_metadata")
response_data = rt.retrieve_cog_metadata(self.con, self.cog_id)
json_path = 'tests/logs/TestRetrieve/test_cog_metadata.json'
log.info(f'Saving result to {json_path}')
with open(json_path, 'w') as fh:
json_path = "tests/logs/TestRetrieve/test_cog_metadata.json"
log.info(f"Saving result to {json_path}")
with open(json_path, "w") as fh:
fh.write(json.dumps(response_data))
assert response_data
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_validate_cog_metadata(self):
log = init_test_log('TestRetrieve/test_validate_cog_metadata')
log = init_test_log("TestRetrieve/test_validate_cog_metadata")
response_data = rt.retrieve_cog_metadata(self.con, self.cog_id)
cog_metadata = rt.validate_endpoint(response_data, MetadataEndpoint)
assert cog_metadata
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_retrieve_cog_results(self):
log = init_test_log('TestRetrieve/test_retrieve_cog_results')
log = init_test_log("TestRetrieve/test_retrieve_cog_results")
response_data = rt.retrieve_cog_results(self.con, self.cog_id)
json_path = 'tests/logs/TestRetrieve/test_cog_results.json'
log.info(f'Saving result to {json_path}')
with open(json_path, 'w') as fh:
json_path = "tests/logs/TestRetrieve/test_cog_results.json"
log.info(f"Saving result to {json_path}")
with open(json_path, "w") as fh:
fh.write(json.dumps(response_data))
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_validate_cog_results(self):
log = init_test_log('TestRetrieve/test_validate_cog_results')
log = init_test_log("TestRetrieve/test_validate_cog_results")
# response_data = rt.retrieve_cog_results(self.con, self.cog_id)
with open('tests/logs/TestRetrieve/test_cog_results.json', 'r') as fh:
with open("tests/logs/TestRetrieve/test_cog_results.json", "r") as fh:
response_data = json.load(fh)
for er in response_data['extraction_results']:
er['point_feature_results'] = []
for er in response_data["extraction_results"]:
er["point_feature_results"] = []
cog_results = rt.validate_endpoint(response_data, MapResults)
assert cog_results
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_retrieve_cog_system_versions(self):
log = init_test_log('TestRetrieve/test_retrieve_cog_system_versions')
response_data = rt.retrieve_cog_system_versions(self.con, self.cog_id)
json_path = 'tests/logs/TestRetrieve/test_cog_system_versions.json'
log.info(f'Saving result to {json_path}')
with open(json_path, 'w') as fh:
log = init_test_log("TestRetrieve/test_retrieve_cog_system_versions")
response_data = rt.retrieve_cog_system_versions(self.con, self.cog_id)
json_path = "tests/logs/TestRetrieve/test_cog_system_versions.json"
log.info(f"Saving result to {json_path}")
with open(json_path, "w") as fh:
fh.write(json.dumps(response_data))
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_validate_cog_system_versions(self):
log = init_test_log('TestRetrieve/test_validate_cog_system_versions')
log = init_test_log("TestRetrieve/test_validate_cog_system_versions")
response_data = rt.retrieve_cog_system_versions(self.con, self.cog_id)
for r in response_data:
cog_system_versions = SystemId(name=r[0], version=r[1])
#cog_system_versions = rt.validate_endpoint(r, SystemVersionsEndpoint)
# cog_system_versions = rt.validate_endpoint(r, SystemVersionsEndpoint)
assert cog_system_versions
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_retrieve_cog_area_extraction(self):
log = init_test_log('TestRetrieve/test_retrieve_cog_area_extraction')
cog_id = '5a06544690b6611f419f0c6f244776a536ad52915555555555515545c9b1ddb9'
log = init_test_log("TestRetrieve/test_retrieve_cog_area_extraction")
cog_id = "5a06544690b6611f419f0c6f244776a536ad52915555555555515545c9b1ddb9"
response_data = rt.retrieve_cog_area_extraction(self.con, cog_id)
json_path = 'tests/logs/TestRetrieve/test_cog_area_extraction.json'
log.info(f'Saving result to {json_path}')
with open(json_path, 'w') as fh:
json_path = "tests/logs/TestRetrieve/test_cog_area_extraction.json"
log.info(f"Saving result to {json_path}")
with open(json_path, "w") as fh:
fh.write(json.dumps(response_data))
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_validate_cog_area_extraction(self):
log = init_test_log('TestRetrieve/test_validate_cog_area_extraction')
cog_id = '5a06544690b6611f419f0c6f244776a536ad52915555555555515545c9b1ddb9'
log = init_test_log("TestRetrieve/test_validate_cog_area_extraction")
cog_id = "5a06544690b6611f419f0c6f244776a536ad52915555555555515545c9b1ddb9"
response_data = rt.retrieve_cog_area_extraction(self.con, cog_id)
for r in response_data:
cog_area_extraction = rt.validate_endpoint(r, AreaExtractionsEndpoint)
assert cog_area_extraction
log.info('Test passed successfully')
log.info("Test passed successfully")

# def test_retrieve_cog_area_extraction_by_system_id(self):
# log = init_test_log('TestRetrieve/test_retrieve_cog_area_extraction_by_system_id')
Expand All @@ -105,23 +114,23 @@ def test_validate_cog_area_extraction(self):
# with open(json_path, 'w') as fh:
# fh.write(json.dumps(json_data))
# log.info('Test passed successfully')

def test_retrieve_cog_legend_items(self):
log = init_test_log('TestRetrieve/test_retrieve_cog_legend_items')
log = init_test_log("TestRetrieve/test_retrieve_cog_legend_items")
response_data = rt.retrieve_cog_legend_items(self.con, self.cog_id)
json_path = 'tests/logs/TestRetrieve/test_cog_legend_items.json'
log.info(f'Saving result to {json_path}')
with open(json_path, 'w') as fh:
json_path = "tests/logs/TestRetrieve/test_cog_legend_items.json"
log.info(f"Saving result to {json_path}")
with open(json_path, "w") as fh:
fh.write(json.dumps(response_data))
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_validate_cog_legend_items(self):
log = init_test_log('TestRetrieve/test_validate_cog_legend_items')
log = init_test_log("TestRetrieve/test_validate_cog_legend_items")
response_data = rt.retrieve_cog_legend_items(self.con, self.cog_id)
for r in response_data:
cog_legend_items = rt.validate_endpoint(r, LegendItemsEndpoint)
assert cog_legend_items
log.info('Test passed successfully')
log.info("Test passed successfully")

# def test_retrieve_cog_legend_items_by_system_id(self):
# log = init_test_log('TestRetrieve/test_retrieve_cog_legend_items_by_system_id')
Expand All @@ -133,21 +142,21 @@ def test_validate_cog_legend_items(self):
# log.info('Test passed successfully')

def test_retrieve_area_extraction_event(self):
log = init_test_log('TestRetrieve/test_retrieve_area_extraction_event')
log = init_test_log("TestRetrieve/test_retrieve_area_extraction_event")
response_data = rt.retrieve_area_extraction_event(self.con, self.event_id)
json_path = 'tests/logs/TestRetrieve/test_area_extraction_event.json'
log.info(f'Saving result to {json_path}')
with open(json_path, 'w') as fh:
json_path = "tests/logs/TestRetrieve/test_area_extraction_event.json"
log.info(f"Saving result to {json_path}")
with open(json_path, "w") as fh:
fh.write(json.dumps(response_data))
log.info('Test passed successfully')
log.info("Test passed successfully")

def test_validate_area_extraction_event(self):
log = init_test_log('TestRetrieve/test_validate_area_extraction_event')
log = init_test_log("TestRetrieve/test_validate_area_extraction_event")
response_data = rt.retrieve_area_extraction_event(self.con, self.event_id)
for r in response_data:
area_extraction_event = rt.validate_endpoint(r, FeatureResults)
assert area_extraction_event
log.info('Test passed successfully')
log.info("Test passed successfully")

def teardown_class(self):
self.con.unregister()
self.con.unregister()
13 changes: 7 additions & 6 deletions tests/utilities.py
View file
Original file line number Diff line number Diff line change
@@ -1,14 +1,15 @@
import os
import logging

def init_test_log(name, level=logging.DEBUG, writemode='w'):
log_dir = os.path.join('tests/logs', os.path.dirname(name))

def init_test_log(name, level=logging.DEBUG, writemode="w"):
log_dir = os.path.join("tests/logs", os.path.dirname(name))
os.makedirs(log_dir, exist_ok=True)
log = logging.getLogger(name)
handler = logging.FileHandler(os.path.join(log_dir, f'{os.path.basename(name)}.log'), mode=writemode)
handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s - %(message)s', datefmt='%d/%m/%Y %H:%M:%S'))
handler = logging.FileHandler(os.path.join(log_dir, f"{os.path.basename(name)}.log"), mode=writemode)
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s - %(message)s", datefmt="%d/%m/%Y %H:%M:%S"))
handler.setLevel(level)
log.addHandler(handler)
log.setLevel(level)
log.info(f'Starting {name}')
return log
log.info(f"Starting {name}")
return log

0 comments on commit f2ae63e

Please sign in to comment.