From 87a75c08f9eea1f9f3e14344411ddd451f6fca6b Mon Sep 17 00:00:00 2001 From: charlien Date: Thu, 30 Jan 2025 15:51:08 +0100 Subject: [PATCH 01/12] wip, check for pre-existing parent reference --- py_mmd_tools/nc_to_mmd.py | 320 ++- tests/data/reference_nc_withparent.nc | Bin 0 -> 34401 bytes .../data/reference_nc_withparentmalformed.nc | Bin 0 -> 32078 bytes tests/test_nc_to_mmd.py | 2433 +++++++++-------- 4 files changed, 1582 insertions(+), 1171 deletions(-) create mode 100644 tests/data/reference_nc_withparent.nc create mode 100644 tests/data/reference_nc_withparentmalformed.nc diff --git a/py_mmd_tools/nc_to_mmd.py b/py_mmd_tools/nc_to_mmd.py index 9c6814c8..b445a3ef 100644 --- a/py_mmd_tools/nc_to_mmd.py +++ b/py_mmd_tools/nc_to_mmd.py @@ -16,26 +16,22 @@ """ import os -import re -import yaml -import jinja2 import pathlib +import re import warnings -import shapely.wkt - -import numpy as np - -from filehash import FileHash from itertools import zip_longest -from pkg_resources import resource_string -from dateutil.parser import isoparse from uuid import UUID -from metvocab.mmdgroup import MMDGroup +import jinja2 +import numpy as np +import shapely.wkt +import yaml +from dateutil.parser import isoparse +from filehash import FileHash from metvocab.cfstd import CFStandard - +from metvocab.mmdgroup import MMDGroup from netCDF4 import Dataset - +from pkg_resources import resource_string from shapely.errors import ShapelyError @@ -86,7 +82,11 @@ def normalize_iso8601(s): secs = int(utc_offset.total_seconds()) tz_hours = secs // 3600 tz_mins = (secs % 3600) // 60 - tz = "Z" if (tz_hours == 0 and tz_mins == 0) else "+{:02d}:{:02d}".format(tz_hours, tz_mins) + tz = ( + "Z" + if (tz_hours == 0 and tz_mins == 0) + else "+{:02d}:{:02d}".format(tz_hours, tz_mins) + ) return dt.strftime("%Y-%m-%dT%H:%M:%S{}{}".format(sec_frac, tz)), None @@ -226,7 +226,6 @@ def getncattr(self, attr): class Nc_to_mmd(object): - # Some constants: # add others when needed. See #198 ACDD_ID = "id" @@ -235,8 +234,15 @@ class Nc_to_mmd(object): VALID_NAMING_AUTHORITIES = None LANDING_PAGE_BASE = None - def __init__(self, netcdf_file, opendap_url=None, output_file=None, check_only=False, - json_input=False, checksum_calculation=False): + def __init__( + self, + netcdf_file, + opendap_url=None, + output_file=None, + check_only=False, + json_input=False, + checksum_calculation=False, + ): """Class for creating an MMD XML file based on the discovery metadata provided in the global attributes of NetCDF files that are compliant with the CF-conventions and ACDD. @@ -264,7 +270,7 @@ def __init__(self, netcdf_file, opendap_url=None, output_file=None, check_only=F self.LANDING_PAGE_BASE = { "no.met": "https://data.met.no/dataset", "dummy": "https://data.fake.no", # used if naming_authority is - # missing from the nc file + # missing from the nc file } self.HASH_ALGORITHM = "md5" self.checksum_calculation = checksum_calculation @@ -285,7 +291,9 @@ def __init__(self, netcdf_file, opendap_url=None, output_file=None, check_only=F self.HASH_ALGORITHM = netcdf_file["file_checksum_type"] + "sum" else: self.netcdf_file = os.path.abspath(netcdf_file) - self.file_size = np.round(pathlib.Path(self.netcdf_file).stat().st_size/(1024*1024), 2) + self.file_size = np.round( + pathlib.Path(self.netcdf_file).stat().st_size / (1024 * 1024), 2 + ) if self.checksum_calculation: # we may have to base it on the complete file - @amundi.. hasher = FileHash(self.HASH_ALGORITHM, chunk_size=1048576) @@ -302,10 +310,14 @@ def __init__(self, netcdf_file, opendap_url=None, output_file=None, check_only=F self.instrument_group = MMDGroup("mmd", "https://vocab.met.no/mmd/Instrument") self.instrument_group.init_vocab() - self.operational_status = MMDGroup("mmd", "https://vocab.met.no/mmd/Operational_Status") + self.operational_status = MMDGroup( + "mmd", "https://vocab.met.no/mmd/Operational_Status" + ) self.operational_status.init_vocab() - self.iso_topic_category = MMDGroup("mmd", "https://vocab.met.no/mmd/ISO_Topic_Category") + self.iso_topic_category = MMDGroup( + "mmd", "https://vocab.met.no/mmd/ISO_Topic_Category" + ) self.iso_topic_category.init_vocab() self.contact_roles = MMDGroup("mmd", "https://vocab.met.no/mmd/Contact_Roles") @@ -319,7 +331,9 @@ def __init__(self, netcdf_file, opendap_url=None, output_file=None, check_only=F ) self.dataset_production_status.init_vocab() - self.quality_control = MMDGroup("mmd", "https://vocab.met.no/mmd/Quality_Control") + self.quality_control = MMDGroup( + "mmd", "https://vocab.met.no/mmd/Quality_Control" + ) self.quality_control.init_vocab() self.cfstdn_keyword = CFStandard() @@ -327,7 +341,9 @@ def __init__(self, netcdf_file, opendap_url=None, output_file=None, check_only=F self.json_input = json_input - if not (self.platform_group.is_initialised and self.instrument_group.is_initialised): + if not ( + self.platform_group.is_initialised and self.instrument_group.is_initialised + ): raise ValueError("Instrument or Platform group were not initialised") if self.json_input: @@ -444,7 +460,9 @@ def get_acdd_metadata(self, mmd_element, ncin, mmd_element_name): # self.missing_attributes['warnings'].append( # 'Using default value %s for %s' %(str(default), acdd)) # else: - self.missing_attributes["errors"].append("%s is a required attribute" % acdd_key) + self.missing_attributes["errors"].append( + "%s is a required attribute" % acdd_key + ) if mmd_element_name != "metadata_status" and required and data == default: self.missing_attributes["warnings"].append( @@ -485,7 +503,9 @@ def get_data_centers(self, mmd_element, ncin): institutions = [] try: - institutions = self.separate_repeated(True, getattr(ncin, acdd_institution_key)) + institutions = self.separate_repeated( + True, getattr(ncin, acdd_institution_key) + ) except AttributeError: self.missing_attributes["errors"].append( "%s is a required attribute" % acdd_institution_key @@ -540,7 +560,8 @@ def get_metadata_updates(self, mmd_element, ncin): # Check that DATE_CREATED attribute is present if DATE_CREATED not in ncin.ncattrs(): self.missing_attributes["errors"].append( - "ACDD attribute %s is required" % DATE_CREATED) + "ACDD attribute %s is required" % DATE_CREATED + ) return times = [] @@ -602,7 +623,9 @@ def get_title_or_abstract(self, elem_name, mmd_element, ncin): if acdd_key in ncin.ncattrs(): contents.append(getattr(ncin, acdd_key)) else: - self.missing_attributes["errors"].append("%s is a required ACDD attribute" % acdd_key) + self.missing_attributes["errors"].append( + "%s is a required ACDD attribute" % acdd_key + ) return data acdd_ext_lang_key = list(acdd_ext_lang.keys())[0] if acdd_ext_lang_key in ncin.ncattrs(): @@ -615,7 +638,9 @@ def get_title_or_abstract(self, elem_name, mmd_element, ncin): contents.append(getattr(ncin, lang_key)) content_lang.append(lang_key[-2:]) else: - self.missing_attributes["errors"].append("%s is a required attribute" % lang_key) + self.missing_attributes["errors"].append( + "%s is a required attribute" % lang_key + ) for i in range(len(contents)): data.append({elem_name: contents[i], "lang": content_lang[i]}) return data @@ -665,7 +690,8 @@ def convert_to_normalized_iso8601(dts): if ndt is None: ndts.append(dt) # keep original self.missing_attributes["errors"].append( - "ACDD start/end datetime %s is not valid ISO8601: %s." % (dt, reason) + "ACDD start/end datetime %s is not valid ISO8601: %s." + % (dt, reason) ) else: ndts.append(ndt) # replace with normalized form @@ -720,7 +746,9 @@ def get_personnel(self, mmd_element, ncin): roles.extend([acdd_roles[acdd_role]["default"]]) # Get emails - acdd_email = [email for email in acdd_emails.keys() if acdd_main in email][0] + acdd_email = [email for email in acdd_emails.keys() if acdd_main in email][ + 0 + ] if acdd_email and acdd_email in ncin.ncattrs(): emails.extend(self.separate_repeated(True, getattr(ncin, acdd_email))) else: @@ -734,12 +762,14 @@ def get_personnel(self, mmd_element, ncin): these_orgs = [] for org_elem in acdd_organisations_list: if org_elem and org_elem in ncin.ncattrs(): - these_orgs.extend(self.separate_repeated(True, getattr(ncin, org_elem))) + these_orgs.extend( + self.separate_repeated(True, getattr(ncin, org_elem)) + ) if not these_orgs: for org in acdd_organisations.keys(): if ( - type(acdd_organisations[org] - ) is dict and "default" in acdd_organisations[org].keys() + type(acdd_organisations[org]) is dict + and "default" in acdd_organisations[org].keys() ): these_orgs.append(acdd_organisations[org]["default"]) if not len(these_orgs) == len(these_names): @@ -812,10 +842,19 @@ def get_CFSTDN_keywords(self, ncin): varlist = [] for key in ncin.variables.keys(): if "standard_name" in ncin.variables[key].ncattrs(): - if all([ncin.variables[key].standard_name not in ["longitude", "latitude", "time", - "projection_x_coordinate", - "projection_y_coordinate"], - ncin.variables[key].standard_name not in varlist]): + if all( + [ + ncin.variables[key].standard_name + not in [ + "longitude", + "latitude", + "time", + "projection_x_coordinate", + "projection_y_coordinate", + ], + ncin.variables[key].standard_name not in varlist, + ] + ): varlist.append(ncin.variables[key].standard_name) return varlist @@ -830,7 +869,9 @@ def get_keywords(self, mmd_element, ncin): cfstd_names = self.get_CFSTDN_keywords(ncin) if acdd_vocabulary_key in ncin.ncattrs(): - vocabularies = self.separate_repeated(True, getattr(ncin, acdd_vocabulary_key)) + vocabularies = self.separate_repeated( + True, getattr(ncin, acdd_vocabulary_key) + ) else: ok_formatting = False self.missing_attributes["errors"].append( @@ -840,7 +881,8 @@ def get_keywords(self, mmd_element, ncin): # add vocabulary CFSTDN if len(cfstd_names) != 0: vocabularies.append( - "CFSTDN:CF Standard Names:https://vocab.met.no/mmd" "/Keywords_Vocabulary/CFSTDN" + "CFSTDN:CF Standard Names:https://vocab.met.no/mmd" + "/Keywords_Vocabulary/CFSTDN" ) resources = [] @@ -851,7 +893,8 @@ def get_keywords(self, mmd_element, ncin): # note that the url contains a ":" ok_formatting = False self.missing_attributes["errors"].append( - "%s must be formatted as ::" % acdd_vocabulary_key + "%s must be formatted as ::" + % acdd_vocabulary_key ) else: resources.append(voc_elems[0] + ":" + voc_elems[2] + ":" + voc_elems[3]) @@ -870,7 +913,9 @@ def get_keywords(self, mmd_element, ncin): if len(cfstd_names) != 0: for cfstd_name in cfstd_names: # Verify whether the standard name is a cf-standard name from CFSTDN - cfstdn_search_result = self.cfstdn_keyword.check_standard_name(cfstd_name, True) + cfstdn_search_result = self.cfstdn_keyword.check_standard_name( + cfstd_name, True + ) if cfstdn_search_result is not True: self.missing_attributes["errors"].append( "The standard name %s is not a CF standard name (see " @@ -894,15 +939,25 @@ def get_keywords(self, mmd_element, ncin): if ok_formatting: for vocabulary in vocabularies: prefix = vocabulary.split(":")[0] - resource = [r.replace(prefix + ":", "") for r in resources if prefix in r][0] + resource = [ + r.replace(prefix + ":", "") for r in resources if prefix in r + ][0] if not valid_url(resource): self.missing_attributes["errors"].append( - "%s in %s attribute is not a valid url" % (resource, acdd_vocabulary_key) + "%s in %s attribute is not a valid url" + % (resource, acdd_vocabulary_key) ) continue - keywords_this = [k.replace(prefix + ":", "").strip() - for k in keywords if prefix in k] - data.append({"resource": resource, "keyword": keywords_this, "vocabulary": prefix}) + keywords_this = [ + k.replace(prefix + ":", "").strip() for k in keywords if prefix in k + ] + data.append( + { + "resource": resource, + "keyword": keywords_this, + "vocabulary": prefix, + } + ) return data def get_projects(self, mmd_element, ncin): @@ -946,7 +1001,9 @@ def get_platforms(self, mmd_element, ncin): instruments = [] acdd_instrument_key = list(acdd_instrument.keys())[0] if acdd_instrument_key in ncin.ncattrs(): - instruments = self.separate_repeated(True, getattr(ncin, acdd_instrument_key)) + instruments = self.separate_repeated( + True, getattr(ncin, acdd_instrument_key) + ) resources = [] acdd_resource = mmd_element["resource"].pop("acdd") @@ -958,15 +1015,18 @@ def get_platforms(self, mmd_element, ncin): acdd_instrument_resource = mmd_element["instrument"]["resource"].pop("acdd") acdd_instrument_resource_key = list(acdd_instrument_resource.keys())[0] if acdd_instrument_resource_key in ncin.ncattrs(): - iresources = self.separate_repeated(True, getattr(ncin, acdd_instrument_resource_key)) + iresources = self.separate_repeated( + True, getattr(ncin, acdd_instrument_resource_key) + ) data = [] for platform, instrument, resource, iresource in zip_longest( platforms, instruments, resources, iresources, fillvalue="" ): - - platform_dict = get_vocab_dict(platform, self.platform_group, resource, False) + platform_dict = get_vocab_dict( + platform, self.platform_group, resource, False + ) if not bool(platform_dict): self.missing_attributes["errors"].append( "%s must be formed as (). " @@ -976,7 +1036,9 @@ def get_platforms(self, mmd_element, ncin): ) continue - instrument_dict = get_vocab_dict(instrument, self.instrument_group, iresource, False) + instrument_dict = get_vocab_dict( + instrument, self.instrument_group, iresource, False + ) if not bool(instrument_dict): self.missing_attributes["warnings"].append( "%s must be formed as (). " @@ -1101,15 +1163,21 @@ def get_metadata_identifier(self, mmd_element, ncin, **kwargs): # id and naming_authority are required, and both should be in # the acdd list acdd_key = list(acdd.keys()) - if any([len(acdd_key) != 2, - self.ACDD_ID not in acdd_key, - self.ACDD_NAMING_AUTH not in acdd_key]): + if any( + [ + len(acdd_key) != 2, + self.ACDD_ID not in acdd_key, + self.ACDD_NAMING_AUTH not in acdd_key, + ] + ): raise AttributeError( "ACDD attribute inconsistency in mmd_elements.yaml. Expected %s and %s but " "received %s." % (self.ACDD_ID, self.ACDD_NAMING_AUTH, str(acdd_key)) ) if self.ACDD_ID not in ncin.ncattrs(): - self.missing_attributes["errors"].append("%s is a required attribute." % self.ACDD_ID) + self.missing_attributes["errors"].append( + "%s is a required attribute." % self.ACDD_ID + ) if self.ACDD_NAMING_AUTH not in ncin.ncattrs(): self.missing_attributes["errors"].append( "%s is a required attribute." % self.ACDD_NAMING_AUTH @@ -1151,7 +1219,9 @@ def get_related_dataset(self, mmd_element, ncin): relations = [] acdd_ext_relation_key = list(acdd_ext_relation.keys())[0] if acdd_ext_relation_key in ncin.ncattrs(): - relations = self.separate_repeated(True, getattr(ncin, acdd_ext_relation_key)) + relations = self.separate_repeated( + True, getattr(ncin, acdd_ext_relation_key) + ) # Initialise returned list data = [] @@ -1175,7 +1245,8 @@ def get_related_dataset(self, mmd_element, ncin): 'type of relationship must be either "parent" ' "(this dataset is a child dataset of the " 'referenced dataset) or "auxiliary" (this dataset' - "is auxiliary data for the referenced dataset)." % acdd_ext_relation_key + "is auxiliary data for the referenced dataset)." + % acdd_ext_relation_key ) else: # Get rid of remaining empty space(s) @@ -1194,7 +1265,8 @@ def get_related_dataset(self, mmd_element, ncin): if re.search(ns_re_pattern, identifier) is None: self.missing_attributes["errors"].append( "%s ACDD attribute is missing " - "naming_authority in the identifier." % acdd_ext_relation_key + "naming_authority in the identifier." + % acdd_ext_relation_key ) else: # If everything is ok, append the relation id and type @@ -1243,7 +1315,9 @@ def get_geographic_extent_rectangle(self, mmd_element, ncin): acdd = mmd_element[dir]["acdd"] acdd_key = list(acdd.keys())[0] if acdd_key not in ncin.ncattrs(): - self.missing_attributes["errors"].append("%s is a required attribute" % acdd_key) + self.missing_attributes["errors"].append( + "%s is a required attribute" % acdd_key + ) else: data[dir] = getattr(ncin, acdd_key) try: @@ -1300,7 +1374,9 @@ def get_iso_topic_category(self, mmd_element, ncin): data = [] for category in categories: # If not given, search for Not available will return Not available - categories_search_result = self.iso_topic_category.search_lowercase(category) + categories_search_result = self.iso_topic_category.search_lowercase( + category + ) iso_topic_category = categories_search_result.get("Short_Name", "") if iso_topic_category == "": @@ -1417,7 +1493,10 @@ def get_related_information(self, mmd_element, ncin): data = [] # Add dataset landing page from rule data.append( - {"resource": self.get_dataset_landing_page_url(), "type": "Dataset landing page"} + { + "resource": self.get_dataset_landing_page_url(), + "type": "Dataset landing page", + } ) repetition_allowed = mmd_element.pop("maxOccurs", "") not in ["0", "1"] @@ -1426,7 +1505,9 @@ def get_related_information(self, mmd_element, ncin): acdd_key = list(acdd.keys())[0] refs = [] if acdd_key in ncin.ncattrs(): - refs = self.separate_repeated(repetition_allowed, getattr(ncin, acdd_key), separator) + refs = self.separate_repeated( + repetition_allowed, getattr(ncin, acdd_key), separator + ) for ref in refs: ri = ref.split("(") if len(ri) != 2: @@ -1436,7 +1517,9 @@ def get_related_information(self, mmd_element, ncin): continue uri = ri[0].strip() if not valid_url(uri): - self.missing_attributes["errors"].append("%s must contain valid uris" % acdd_key) + self.missing_attributes["errors"].append( + "%s must contain valid uris" % acdd_key + ) continue ref_type = ri[1][:-1] valid_ref_types = [vt.lower() for vt in VALID_REF_TYPES] @@ -1456,7 +1539,9 @@ def get_related_information(self, mmd_element, ncin): xx = [[ref_type, tt] for tt in VALID_REF_TYPES] x = filter(lambda a: a[0].lower() == a[1].lower(), xx) ri = {"resource": uri, "type": list(x)[0][1]} - ri["description"] = "" # not easily available in acdd - needs to be discussed + ri["description"] = ( + "" # not easily available in acdd - needs to be discussed + ) if ri["type"] == "Dataset landing page": # The landing page is given by a rule in py-mmd-tools # see get_dataset_landing_page_url @@ -1469,7 +1554,8 @@ def check_attributes_not_empty(self, ncin): for attr in ncin.ncattrs(): if ncin.getncattr(attr) == "": raise ValueError( - "%s: Global attribute %s is empty - please correct." % (self.netcdf_file, attr) + "%s: Global attribute %s is empty - please correct." + % (self.netcdf_file, attr) ) def check_conventions(self, ncin): @@ -1566,7 +1652,9 @@ def get_license(self, mmd_element, ncin): if "license_resource" in ncin.ncattrs(): license_url = ncin.license_resource if not valid_url(license_url): - self.missing_attributes["errors"].append('"%s" is not a valid url' % license_url) + self.missing_attributes["errors"].append( + '"%s" is not a valid url' % license_url + ) return data else: data = {"resource": license_url} @@ -1585,7 +1673,9 @@ def get_license(self, mmd_element, ncin): data["identifier"] = ncin.license else: data["identifier"] = ncin.license.split("/")[-1] - if not bool(get_vocab_dict(data["identifier"], license_group, data["resource"])): + if not bool( + get_vocab_dict(data["identifier"], license_group, data["resource"]) + ): data.pop("identifier") self.missing_attributes["errors"].append( "license should be provided as ()" @@ -1595,12 +1685,28 @@ def get_license(self, mmd_element, ncin): # and rewrite data dict if necessary if data is not None: if "identifier" in data.keys(): - license_dict = get_vocab_dict(data["identifier"], license_group, data["resource"]) + license_dict = get_vocab_dict( + data["identifier"], license_group, data["resource"] + ) if not bool(license_dict): data = {"license_text": ncin.license} return data + def well_formed_parent(self, parent): + if ":" not in parent: + raise ValueError( + "parent must be composed as <%s>:" % self.ACDD_NAMING_AUTH + ) + nauth, uuid = parent.split(":") + if nauth not in self.VALID_NAMING_AUTHORITIES: + raise ValueError( + "%s ACDD attribute %s is not valid" % (self.ACDD_NAMING_AUTH, nauth) + ) + if not Nc_to_mmd.is_valid_uuid(uuid): + raise ValueError("UUID part of the parent ID is not valid") + return True + def to_mmd( self, collection=None, @@ -1675,7 +1781,9 @@ def to_mmd( geographic_extent_rectangle = overrides.pop("geographic_extent_rectangle", None) dataset_citation = overrides.pop("dataset_citation", None) platform = overrides.pop("platform", None) - file_location = overrides.pop("file_location", os.path.dirname(self.netcdf_file)) + file_location = overrides.pop( + "file_location", os.path.dirname(self.netcdf_file) + ) # Get ncin object from instance ncin = self.ncin @@ -1707,7 +1815,9 @@ def to_mmd( self.metadata["metadata_identifier"] = self.get_metadata_identifier( mmd_yaml.pop("metadata_identifier"), ncin, **kwargs ) - self.metadata["data_center"] = self.get_data_centers(mmd_yaml.pop("data_center"), ncin) + self.metadata["data_center"] = self.get_data_centers( + mmd_yaml.pop("data_center"), ncin + ) self.metadata["last_metadata_update"] = self.get_metadata_updates( mmd_yaml.pop("last_metadata_update"), ncin ) @@ -1739,7 +1849,9 @@ def to_mmd( self.metadata["keywords"] = self.get_keywords(mmd_yaml.pop("keywords"), ncin) self.metadata["project"] = self.get_projects(mmd_yaml.pop("project"), ncin) if platform is None: - self.metadata["platform"] = self.get_platforms(mmd_yaml.pop("platform"), ncin) + self.metadata["platform"] = self.get_platforms( + mmd_yaml.pop("platform"), ncin + ) else: mmd_yaml.pop("platform") self.metadata["platform"] = [platform] @@ -1747,26 +1859,39 @@ def to_mmd( self.metadata["dataset_citation"] = self.get_dataset_citations( mmd_yaml.pop("dataset_citation"), ncin, dataset_citation=dataset_citation ) + self.metadata["related_dataset"] = self.get_related_dataset( mmd_yaml.pop("related_dataset"), ncin ) # Add parent from function kwarg if parent is not None: - if ":" not in parent: - raise ValueError("parent must be composed as <%s>:" % self.ACDD_NAMING_AUTH) - nauth, uuid = parent.split(":") - if nauth not in self.VALID_NAMING_AUTHORITIES: - raise ValueError( - "%s ACDD attribute %s is not valid" % (self.ACDD_NAMING_AUTH, nauth) - ) - if not Nc_to_mmd.is_valid_uuid(uuid): - raise ValueError("UUID part of the parent ID is not valid") - self.metadata["related_dataset"].append( - { - "id": parent, - "relation_type": "parent", - } - ) + if self.well_formed_parent(parent): + if ( + self.metadata["related_dataset"] + and "parent" == self.metadata["related_dataset"][0]["relation_type"] + ): + parentinplace = self.metadata["related_dataset"][0]["id"] + self.missing_attributes["warnings"].append( + "parent reference already in place: %s" % (parentinplace) + ) + if self.well_formed_parent(parentinplace): + if parentinplace.split(":")[1] != parent.split(":")[1]: + self.missing_attributes["warnings"].append( + "passed parent reference %s does not match existing parent reference %s, " + "not updating" % (parent, parentinplace) + ) + else: + self.missing_attributes["warnings"].append( + "passed parent reference %s already in place, " + "no need to update it" % (parent) + ) + else: + self.metadata["related_dataset"].append( + { + "id": parent, + "relation_type": "parent", + } + ) self.metadata["related_information"] = self.get_related_information( mmd_yaml.pop("related_information"), ncin @@ -1783,8 +1908,10 @@ def to_mmd( } mmd_yaml["geographic_extent"].pop("rectangle") else: - self.metadata["geographic_extent"]["rectangle"] = self.get_geographic_extent_rectangle( - mmd_yaml["geographic_extent"].pop("rectangle"), ncin + self.metadata["geographic_extent"]["rectangle"] = ( + self.get_geographic_extent_rectangle( + mmd_yaml["geographic_extent"].pop("rectangle"), ncin + ) ) # Check for geographic_extent/polygon polygon = self.get_geographic_extent_polygon( @@ -1795,7 +1922,9 @@ def to_mmd( mmd_yaml.pop("geographic_extent") # Get use_constraint data - self.metadata["use_constraint"] = self.get_license(mmd_yaml.pop("use_constraint"), ncin) + self.metadata["use_constraint"] = self.get_license( + mmd_yaml.pop("use_constraint"), ncin + ) # Data access should not be read from the netCDF-CF file mmd_yaml.pop("data_access") @@ -1818,7 +1947,8 @@ def to_mmd( # Set Activity_Type self.metadata["activity_type"] = self.get_activity_type( - mmd_yaml.pop("activity_type"), ncin) + mmd_yaml.pop("activity_type"), ncin + ) # Set dataset_production_status self.metadata["dataset_production_status"] = self.get_dataset_production_status( @@ -1849,7 +1979,9 @@ def to_mmd( if self.checksum_calculation: self.metadata["storage_information"]["checksum"] = self.file_checksum - self.metadata["storage_information"]["checksum_type"] = self.HASH_ALGORITHM + "sum" + self.metadata["storage_information"]["checksum_type"] = ( + self.HASH_ALGORITHM + "sum" + ) self.check_conventions(ncin) self.check_feature_type(ncin) @@ -1857,8 +1989,10 @@ def to_mmd( if len(self.missing_attributes["warnings"]) > 0: warnings.warn("\n\t" + "\n\t".join(self.missing_attributes["warnings"])) if len(self.missing_attributes["errors"]) > 0: - raise AttributeError("Errors in %s:\n\t" % self.netcdf_file + "\n\t".join( - self.missing_attributes["errors"])) + raise AttributeError( + "Errors in %s:\n\t" % self.netcdf_file + + "\n\t".join(self.missing_attributes["errors"]) + ) env = jinja2.Environment( loader=jinja2.PackageLoader(self.__module__.split(".")[0], "templates"), diff --git a/tests/data/reference_nc_withparent.nc b/tests/data/reference_nc_withparent.nc new file mode 100644 index 0000000000000000000000000000000000000000..de416b84784c5dc74f7963982b014961a536e240 GIT binary patch literal 34401 zcmeHQ3zS?%nf~wOH3<_)0!$u8x#493)9>druX!f}$;>b_BpcD~^mN~u=}b@Gu^*F6 z3?{JXk%tI~t4CQlvLGrP6(7iEH6k9B6 z59~VWse9{I{q@)XSN*T5dvn=_wvNS%RxKh zabd>H4v891p(&^N7KnNBmTCb$o;1fM9?4rJ!wM(szM36D3cR!Uo6Q})8+xb~oz&PT zwwiSt2W0pA71Q@x=tQl&N9K*z?V&PE4K+LDG^Vx48{Q||&8553Bd@mfZ)gKXn1W03 zEVXtCV~3=X>gCnCzL!OMDwij6Ut`<#p(W+(GtK z=)zM7%abV1e~2=H11?&?P0k}~K(_0K{ze*Lp#Mi05z*y$U2`U7cy$IeN+A1o*R^L+ zegUq6MCT*>s_%(b>RE^@FnSHL-{02VPR;0&wh-lz73x=Y&<~Ho6^O{?&tGt5Cp9r; zFf#6J&l8{SrLQuKGl)(^_S{RC_S3yAG0#PNWUsX?8K6_RAWU>JvYSWRlC+zF0`=38 z9eDPJB#rVU0|`M*djDmq6y3qY2Z6X^=u>HfZs*Z%Ac`XUX7j!Y+IK9j0CXF&W0&@Q zfc}dS3lK2}2V+}y(^cGUjObKk_a*%o)AzX-6tKkk{lL|i&`PGqdx*M`oqx&qFa6dV zcVQsYCBgrN4n~O1xKr+%e#U z5s$(_4{~z{sKz95kdErPH_(wMAZ4u^A^Po;l=`gsxSXrmCnw#?6j?_kq_%x_IN0pq zC|$@Hf(EBBG&|AhWM&^`4zI6MbFcq5SNPjaE&ZKcSdVtb{^Wy&k1i1$CR)8ja0(ZB z<5#>Gs_E@T0Zab(^0LDk%OFdpF2B~X&K_QI%crTHg*N0rmMGO;_=V3n1j7TKZU0K2 zUvERruI+Kuez z#*_a>H=Ka0W}-B*??30+=MYZUUb%fA_40Z=P5h(lv$yS|J9(W$A#6tWqpv^tU8--w z)p#vkglyMEC;iYNYd(9?L-*1{j5?YC73vv1{zr5&BeoYZ7unN8@4b)eSsz2B0u{RS z1`uGuk_Sk^9(9alf?R8=pf<} zA;Y2Am$2qS(%n55uacM5zi%@NSo+^?yf}<7hppsT(GKgyF)tmm!m)B3HloAmbSxi- z4dXBY9oC*h;yA2LhYWD6ZigM~u(uqRr9%KX#FRrKkzH;&>`aGU=CB?eHl0I^IHaFL zYQ5f;+k0dtp7hsOYuXBiuW4+?S;n3{KaP#TrE5;G+#Q{k*V6Jms4=gWpVZXUh=%Nu z&!u}j41?$%nU#F|`K0`?$$1^C2mqQ7Q>2z!H#YaN5;+z>w(}%TFr7O$wRiP(c6Y`5 zTAMevQ{6Uxron3GR$a}T+EH1wYC^7B^$(}>I{qc}tWjuf>(Eo_Q6rmAn^}FJSkSX( zq5Nn(kxM5AGKO1M2__&t`@5rFuS_TQ7dWYW&SBB%QZeec2c1Ho!ojwdW=W@S)M#s& zE}EPc(s{7ci60_OqMKe}e$_D}d45TY^o8AX#f(#TwIZ4`N88^iA;Py1NyU?v(3<0A z$=9s>VJoM_?2A-Wl>#LSoPF1`SIithx|5Afq&L(gV)?|0jXzFSu8L*h^nWpppoTcG3&AnmpAImN1=@)tjG!3xWPIG3>x%N4z0*Sy7a0vDVvhqv+b`Vu;t zJKZK7{Mu)qSuSG^8&Qb#(&ZB<=aE%tJpW+LT z*-CeCFKEIZ{=LgzHYv$XP{3~MFXrE|i}rF2pA1ZVcgu#W=nq_g5iqBIu+{SwdXAf{ zA!37`(l^|}u_NL8bH&dTR_~)MqmCE|+r#(8FZvH@$#0V(iR8zxd75rx7%eii^W?`D zycE6i={HdFcaSUtQp`zyn`Q32HU5hgRu^# zMpy5=`|p5xZ^00y!oNcMb18he>2s!KVt&{ffl~MLrZ+Lg8pg$Jx{wEKuFU}()kn%0${9vJo;7oYVWR8%K6JNa%awm-wq!Yc zgY*|=3nIOh*|HFrDUy};^6H=)X68%fW)0IZhSiAbMKQ)msz}wil>%V zRyBFba8i4ktWP*iF1Ivysa)-JDl1}g{dp2hslI}5#du2P3Jxn$PLtuNdzC9VOdL+C zT*0xz#-(yq-?E>klrz;hO@<>&i5|3KQR=>$S1C}XK$QYj3REdjr9hPeRSHxoP^CbX z0#yoBDe%`!fsVeu_R{WwhVv$$7aqNK`5YP-KK%AGyw`SY=xizNTZz%^56MC8N3?>| z1Fp5do&O?lwsJl-g0$Z{d76H9oAUSNgSYx`ck71p)DB(vTnSz6 z(DCl=TtrHa!3M<&+urWR&c1%4m8i@i5=@1V-rvA zU-)G~b9LlbL(A92s5Lx4D5-x8&X|hGg!U1low!5#($jZ~y!^*Lnm=~R@*_7Ij@;Ns z#}S(+u3nY3W1*&jik{S6&pR-4KvOTmoYdRi(catM)!H5xMCFs1Ow&}NkjNVdNGal$ zcEqWcPbbS*l1?=sx{52QT%DemqsIx;Ne6?}(RlT|bP`?_tgKsBX#~Jlq(9#XROC&K zK;5emsCzX6b^qCO@3}|~rc!RFAU{_#S4Hkrt|+PC9tw|t`S$m2@7U006Aa9gFNeSR z^|x{j@w#G+)EZLQqNtjx6sS_*FG>OD`e#(KH|4H+D>ti&pLf#N#!pQUHj@nscD2zM{ zG6eyzX{oBCW-2|HHgdeJ#)+%)R)eaVFJ^b)m9AV4gEdlW`)&-odV^+eRCU6w*xFF7 z`ikag+SS$5!>nQ2U9t(G<3id%*~@D#sW^6pN#qR+zfhSUccMx zYmE9Lm3BRO2x=IosCmC5F>Yjx>=5sLa$2gNgRVI>QSc}VzF{dbH35}QQ~C3|&k6nG zzK74b<~hFjCEs+`)jV=eX)Ty$E|tzE3PxT)kN5agRXE$8-}rS zEMw*jH($EV+>j?_CW|dl*?b`V_%#ee8zy@c1>!!B8^J48-w!^w0T+-0b z)|(5XW`1lKoV4G7pgCYJ$l0K;Z%5CuO&EolGc)E8=$z4q409-#7#kKhgBU{*LQK^y zK?R_s2@W&*rKjCVBav0Drvrq6_HRJPeX%V74g__~6g`rV03;v&g~|~^~3=9l^7{tDmMlLA@E7+tyVKT&Qk{1fHsm!?DMFp(Awehxg!mF zb8A~0uJ9tEoa{+6!}<*EL^HwvrM7k;P*-bTgO20J_}kpps^(zQVy}&30s^!7L>quE zGq=MSN++`VBszCm{KvYJsXPm;f^z0u|19HP-1eQ9HN2%M)LuFX^I6*=?8(a}V10w+==^4x&kYBfes1N>A>#9CP<`b#7sZO{q02kxe`%9P;d z?Cq%L-D5^p;mCxSD>t4W#sL_+PPVH^u`V+itkJw@JZPloZL6TZR)9sj=VoLI%PDKPUPcW+z6Ig;+-gzT84o<=hki--cGmnJ@dKR+< zYe;fX986Gfq{Pt>8eo=}TN$H?bVl@QM{vx{7n-C7Q5+~YaV`Sp8lMAE9hE0Gd#ujR z}G07=|JvhjJWEuQl|oP`Hyq`^NM)j$9rDY&^I4y zv#nMbBJt6*CCd(F52-P#oclZD+NP#TaLsIdG_g}+^d{gww!PIqE!;l=U)}ev^Y7MR z+VKwV5IWv(;;#8N;29FL)+DMPXDo7Oc@+{Z8^BL2v_sFG>rvnqZvd#6fK+yB% zU||Q03cX;mvdlmR!Pk)L&~n3YJI<#S@g_@FFNqmFzXO_+i%JNTr6J0c&*9d^OE3_t z1qju=4sEM%aLWAw*yqj3s5m-XPB-!I!)lFPLNLWSj>$ zWGaa?xeI9nupu&0)|C?_lf!5*vg7HTnPpqC0b(_gh0K#DB*lsXwhS_6!@4rtEFOTk z6{rekZ1k}iC%sa+R+T=JJ+J|;CYRJ`s09_ECXuCB1vYkdY16aQTOW2=g*gT5gUGr5 z(K{WrvmTO3+W+ycR?IahkV4wX;~5hnf}cW>%_kT*XZd#HO{ipQ5FWlN3cAB?@9yE$ zV8~YJbNjq*FPa}O*au%9hh|x;vj~Kb0euGOq8qz+-PPN#%z#Q(mA&fv);>k6q&dp` z6wvJc;Y>4E4qI7ZQeR1Yrqx(Zo7{eAH)x|&Tj%CZe$h779btEmvI|abP3;Q!D5KD6@mHdIlIUF$>btWmt@h!Uq>KHdGFoK*B!m zRyoE>>W|gxpQ8T$=c2wQWof;?2;p9sU~9L4$s*9OhZVX^-KxSamZ-wV&fo0spZ>U_ zWeP5hu`J+eGtD-M1I|L$7c>SolOUt zjx&DMwDCueG*R@|P3K&Hpy^hW4WsPeQT8>IJ%F-i?LgCOC=1;8NRxB-4PwyRbHf4l zR|KmxI0n}48`jDHvYS8}eD4RDC!{5@k0r}a)#OXQq@AbHH<(m(#m{$noI;VEz2E5{ zxmd{+y!{F9dqPlm4=g%0JNk-Qy~7yLbwm-vF`qBy3+kA`zL=hcc4GbOO6Y2R@l-F1iLp}s@vq?cvrd~%bt#_H@I>&h-PrxL>A<c9A74*tS#9VV{jn=!jD= zTotL75;5uyhpl*pG%J+n$&;BA3$bGY~Ch;}6-jgu*^w8I&fc+5rX(!fM?Tsp^4!|WI* ziUm0E5Q&U8Cnkoyr>yCQ4iS%Re-8iL!_Tav2n|~$m}0-o557<4Td!I&No)aY@qG`{ zV7|P@CJbpTp7<>s;)4X@J5{7XUmqid)y(4UG>9%)?{pjiVIWjf5_u2uxJkPTry!&V ze7uIHQ|M!^Fl5fn{iTm`m(vstbqgBWs*)xU2m3OLt4 zmf{p~R9gQvrS(5wu63;LDVEl3L=gf$;uA@vm2I%=H2K98(jM06-QxYvxt{biEh#qo>y`qlauZP`!pS+IXIvNny0lyw`2ZQoDqDKKtTnBx) z4oBs6NN;opWZRGz*NxafK)?3ufKO*j0N~WgXxE zzQ{3zqqvSp+@cX&hq(@u9@in}SJdqf;W{9R77XCp&-j2P%wIv~s5|Pz9B7n$wC2OQ z(%xW61302t(x#XL@o(1v*fwKDP}(&+&)RBpXa={zJ8SGim(>F61Ndf1V+<~UbrH`e zS^9hOymy&26dp=#bawUiboRC*B8N(O(F68kX4E3xYNKquVM3)08TQ7!p{q27LxWV( z8kJSWG#W(t?rH}pXN)1Fh!~A9MyFbILpc?Aoh)yiBCBk&lm*pn%oCheym7Ljg7EI+ z9EBsTWJgsrSpp8#A^KV3Xb6N^Hfg%^gKjiz&|6{fmY?9`+rz~HHzLJciGS~u!zG70P=6@LQk~a+WQ3tjJ;t9HuUxL$^*?^~=1Kx=faGWC zNP9sEFby9%sb`r*$6Wr*qfSw_8l}HG>038c5@o}%C>8Ujh_aA~DpeL`N`7(p<#4Gz zMXF8dxhh0C?l`+hoiprM4F>)V1Os_{)Opfl4c4eyhl1@DPI~BTW$aBzD_-i;CZ`1k zYvqeFK!k|gtJ3l*TV-V@u~LwzY=r^CNBQf#_Q28M+hb9D`Bwy{w6Zv#<<-Jako#jmEvc#l48(#5=u9I2ip)u#l1m3&V%m@^rui>8A2P77YaWSw!yNj}q< zrs#{Rrm~KlDiT+>t*YBr*mJWs+1|XlFOS()+GT)#p^=M4OJ2Jp17V#>zOcQpA4angEg=u6VrDS2@f zErIrVRIf9X8?Qe2MW>I)2Es2Mx&TpnX-pg5F|6%2UCm^#Jek^Fy$n>7H0P6f~lVm9KEmUnskgt@)Cx5jeO@oDk|1W`0Dv5;1j!0AP@t;*x`tI?(ML6Yj+Fa4* zh28x5mN}XVK%Z&BI(`|ykEKdgUf0g=Ss#a2@ep{9cZoyh!B*Sbv-WRe@Yh8+cEL&I ze9g104m)!E6_X_lsQll4h_9He#B+{=aM+69fU5+DVW=-ya3sfm&)bD{O$?upk7E?- zl??gf7)P7@Ni?0C z@|_?KuF4xMH7Kf~J3esgB~B}}36<9@e(k4~tkBKio&-n4CgQnxHxfq%d?6h4jNxno z?6Z(#({z#mv71nDLUv4NMa6eun@P$SK80$>EXek;JcW9=n9b(hBPG?|6lT6?EFgAg zjs#|U2{rrn^Nf`8MPeag0Ur^34uMA1p%KT|*?g-yG}UM|jv-m6MW)2btNW@|hvseT a(6FgeO;rk1DNv$`?R4H)eDe(U;(B~`w literal 0 HcmV?d00001 diff --git a/tests/data/reference_nc_withparentmalformed.nc b/tests/data/reference_nc_withparentmalformed.nc new file mode 100644 index 0000000000000000000000000000000000000000..ca895956d1964232f08b6eff7216bab4ca006d33 GIT binary patch literal 32078 zcmeHQ33MFAnXZ-(WFv#^fPBP(b}+;S&3#G3C21tt0$Y+b!d?fmi-J(KW-Ftb3IrHs$5>92lu#PEI<-}D$g|Ru4?D0C$BtJ*w*W3C06X;%{ zT0AL}i*3~D$uW4s0pXdlt9gm2Grv(hu|cvqoBV<8OYLXx7SC+(tUjj|vV~$s-@MYU zuu4>rJub+Z#xSq(G@4SHZ-ba8ZdpCRr_$#5B-_XbE z(Mg?iV#`_Q%ZTm|%0b_6qvLnyE%xn^GE5CQJH#|*Ymqm7K(w37jLU93v3+1e2Qb1E zT!LqrwTlTmB!$UdUT#SKj@Md1wHAq54VASnzo~gyR}!$;*ZE3d43nEDncKHOsgLaf zCe#O|Uk&n%JtMMGe$B71{!;tE#x7vjxE#;6)Ui#-o_^{$#=cK<{SD!aMer({$F4jc zSN)7#glzryZ=c8Rq@|r^j9R{T|92O#lV}+Ly#%sFtKYqly|f5dzhi6w*~d2B(#Yy* z9eN>mkUbr_@KixT3U^4k&>;^yhMJtuSQE0{Hx9J0K?3?e3=}bT#a-8&#d5SdLkiQ7 zeW&Nzvzc`auELCMM)tM9lL^+h09PP+HL~B|*4xS2&?Vc-SRPrSab*|#!LhhHouk5D zxZtX8)=G>5Wz^ZeCqL8AzD6+41YyXYf9a9|b{|R1^UxmID;XxuSO{*i4_=i$fOlU6eo#V`+ zMD@9h5l2~(Zq8zisBrOvd8q?bBbK;W#~P_O@R24UGwd5K`kj-MCGGjRjH)>&C%ct6 zvW_t>wVkuWHO#INWfu~L;K8W`&B=7OhK3I_iq_X@`B#5M75;Kl`#^UO)}wP^fBfOX z#}@MuW^C1BK2o@_^k2|osHJxn2W+y0O2b5tY7SQ6Pi{eOQ8TS+SxvW=AMQ-05FE>SSJwEK4Ud0MRKb{N_7 zE$w%)29gp92KB|*&-ADMnca9YuG$#OBKzL+z5_1dbnR6;4zPY&kEip0>;~hu1ME&( z=THb+kp1u*Pu;^BRb1URpY24pd+#YfaLJm_?S1$@_AsH2CP0Ptjh*;IwuTVf51EVX znc<7>XN{zfAyR<~yW~aRLrkZFn;Dxx_QD4@{)CMX3`oEpWUHF5d)Or?vY%Z2m`jj_ zzS{PfOKA2l_@|d#QfKsxg@;{2t^JobA7=f;Q}l8+vJ20D;;>7i-uB|sKe?pcE7v^! zr~A%)Vjn6`llb~y;9}xpF2kYN7n9~fQoVf_uN0THzwa;(*!tgTyts@om#ySl(Jt%7 zH7{MV!nJZ-HloYubS)p34dXHaUDlpU;<&6$mke;NZkHYFvbS88rAq*~#FR@SF{j*g z*_kf8%w;{gY&w@1aS1AyDE#-f>y5W&;t79!m7**s_=-YioNeqm^ApGzT(bIP+uc#w zvU;|R#LB#SdQw+c#~YGIK9}9=BN&X`E3%SrKc7iItTJyP6#+o=VT#nV#KyKHDUo;K zM|PgT38r)Rrp}&ZcW+NBnP}VC$r`rPGX+*Vwd!fx)QQTxRV#AYYG5R5sraXfkgqNTRftL!qT^Xy3rXxVhA8@ZWK=!mm*vb{OAu~a|V!j zEiTYWfJ$6`k{{_30i^#TUmHm-0GW$YLdhLK+Hd1?s$D_m|NE&6mfH@*d4xS87qp@` z?4GxXolFI12=X?5L0!y_qfWOA2fzNgXP1eX!$uS$y>!LNYuKmgYJ)&4bN{FP>?A_0 zQ-lug+vV$Lhp0xo&`+ra$8TeIP%mgg9{zpHerK{YH9-Nnt^Yayj@@iO)d+~d#64R# zT+RMK1)zXN>igS#UuDlzlhuroLC>-`-a)Y=?)y{4&lXl4UG0-)tt^ssS5$2pgu0eB+ zv1{~P^TRdruJLdWuyY!^2G%vG&8v3Z{Z}w}Zvu%z;a{n|T?$`p`kZB(m>;z%km`Qk z^d=^soC<=kCm>{<>dNMFTeMtJ7XX%2F+J#sG>AzQP?9-gckgRhOZU?JG_)}ei;Pwh zEwZNERU#9Y_PUP4(|TqVG$W(sGby{-w5=d{=b)7^jEhFLU;#GOrhtv?Bcl)Jb=^uC zX1*}O*bT``?-s*y)!RHQh8I{N{aJ?vk>2XVLa`;uveI5&19Zd8<5Im@({uwvYDD%T z2S(Q{Y#dzrz%lCjMDIra)Yi(fCQTVia!-@=38l(%TXUDnpvHk32WlLsaiGS5|IZxgN+vr?y9Wx+n|y(L^vV@;*aY|Ccbw_J zwrfLoduiW_Q*$sPh&qU91*OUj)btbg&f%>dmw^MSb+D8EB5$*EIyHiH&^~#ZeQvw- z_r-%Z`)_Yz!})TDE_$ATE_di?_jWEKCC6ifV!2~)cVl;QfUy;*Od%3FheI&|8HkhQ z$(-UR``?2gjZ{h$X444osJEC7H|9^MZPuFs(QkB#*|+=cSYI zvS3BS(rOd{Tb|xN3S{I>QXuao1@c}}An!kS-o1MzG1YP>2kE(-xiWGmbIC~w_egmB z!*{-Wd)I~zhhU&F`AYO#-*_|E5Unf1NUkA?EsAQX#(^3K{;V8uuYW?txH*6Io4Ki_ zU>lq^sM1<4JDA5CQKmR}Mq5s#A*Hm+>q9v`ja{s?u3)ph0V-TKGAT1Zn1y2Di3B03 zWX4+gUnvMB2$}!)xA*?w1qoE<*o3NPOh1F zkA*sv1_C{w$Y%4F#ID-rBAG}MyMn2u2D6zg(S-5Uaw*qnHgJpqBwuih=WH=9kLIxg zV-BZ0o`m`IqZ#(c{Xwrk(9#@;RonF>5+oU?%6Y$2o6vK*F-&`(l$NOHqH9X6EPN~r zzh)~jIRT|kllcpK&W-%-fk)20=6SmKIo)(O)IEA`X)Ty$K9e=Hf^KoBL;@W>g`%d5QwfB5<@nfPS&t@+RcIB6p`mB;jgMz&olmo-N5jPQIPeiF}-LSfvB`+TV9 zHB4V&MEBXIpf88*=&Uj98P|rfp2f&vQs{Xd3tUjm!R@XBj$n9)v{9Xa+5pwH9m2>2 zOE_eeAg^muB@S6#*LRKQ%)IWUORt$9_GQd;5l1r%zF12j9E$|iwaIiA_gJ85b-Xy3 z%ci+NTBpFcrFBlT3X5|>&jWah&sg9m2&b0-B;jR=F#w6-oMZ2o#AXevkmtQ~2%K2# z*q1q*7p5FegZQC4$QY?<@lRGgF?ZFU4KInk6KC!7I zd49YT%XUQNj7#h4Ici&B%(TWwFp|z2Ffm2ST~3B z+V}{+83GMO2r*f=1QmeNCPtXhFFox|>zW~3PX`Eu4s1Zj$@o?P2ZAbQiki-4$FR_F zk|Exq7r;-H>jqBg=B-f`18wcVxm|5S)dn$M+Gqh&xg{KmV4Q^BN|?q3O&MSV-Uu#{ z&m*ToqRSKT#+uZ&L`Mg%@FJm@>}fMc`V8%OGd})H9PPlMo&VXt4o;XHzfIbe-rl0ZcannfdH zrP5qe5^+v_*V&g`9}kCuQOQoLI5wu`CvhT0AWsd*t(G)`8yHVHh*&F2d4F0O(gvMi z^f0`ymwqT6!@})V_t{U)iWtu^6-&!E`G!Gj!x)>I1YMyG*7XZ&8Ta~ zv4oAnx}Y4yl0Kg>)7oG$2h`D_x6i!|}L(?6Ynw&5?)G3+3 z=Lu$&Hi;#OO^qd zvMP9oOxx}63n_#P3>A+Z}>PQ6NIq%#`$7!aP zln&&6PEW}$t98mDm;boeKdp!df3)vq1%1<@Hpgnkl1PnZZCQ3Cdq~nKbL#KRYg=2Z z!8MK4n6^t`^g7@^z9SKw7VaCsR}GxB`ECV9KFzAEJtQdUgU< zfmTq_lI$UJdElCLOWOCaM8tg8>3JqBQI6|Q11HU`Ly6JDuUt5Tnd9>@TflS^nc z)Pf3-lgL)A92=*)bm-aXtq-TH!kmKj!I*pfqjz@fu0}{EVgILk5}0dHAcd@M;TbU@ zhMz={%qJK)X9xD+O{jEc2p+yH3VWko|DKV|P{dIf@CN)|KboJ&hY!9!4$YEQClLrA z1Nscmc{fh)x~G3YngP|UDreR8iKOIJ+8m?t4EkS#4Rp|1o!(`qcYP3}Cj z8+LH2qkBs?y=bO(_4cd1?a9vmEo}pIzxjf;joky=)V6AR#|;~nbjjX#T9N~Ldc@#H zHD8bE!c=7_Tm>3FCmpo3Y)3&_yOMOF;RNZOFmPf$Vbs0 zNZJ{Z^QGHRx0?A;>Byitye%?!buM5DKMLE+>Q03hj{76=NEDiGne19m0UP0;+CdjU znH`K(Gr)j}SrC>k!6IBFKDd~%p>)VN5>C1kg2rmH0J zzK*g7QP!p$YJCM|q5B_gb??4G4BC5cIKciVK2!=E1AF%k>*P1&CNKrQ_d_%$Oi5!O zOO&0ah?jhsa=ya8NvvX5{%p6;EfmSw`>lG}#Zs=|?N4~$lbpNzVA09hkt`Z&mp-Vf zh$2Sgfj~SERxyJEan*o!BK_;pRJp!rsx%Gf%t0;Jq&20R97|D!wO_D!+{0Ijn52xf z*GMm{=))fmMyKz?0n^!)!M5mr?_8Fg1vuqjFxlTD$fh-!?Fdw6)#Gd2i|!$WNXfnY2U3j6&*$898lK(3MI z3?=S$<+Ey!IiXUV7xDyy#7BsCSQ?>3+25C3GciDwZoVo8Xnc~EQ4XOmgm5yJN9aKx zq2vn$e8HISXt7;r3k$OCBy`wk`LhX-m9l<6~}fIU3Q4p&m`L$JqyRv*cZOO@c5Mg)%oJUW+3nrl!K5*%MrO=n(s1DLeXe z4?ne!A~bCiK#Bt*Kl}laZ@YSNh1det{QDkEf%)YafjAOwWfl|$pEo<)ac#ZT|rDD1Hv ziK4i&9^P@&sUy*I;w6Lq4a5jbK{L24hzk-V;X1&=trjs1J!CuP&V(J{*>L!fpP4eM~5;-8^y|p@jj{- zd_WRGQ7z{wkOgcQR7V6c1dX-7?c%)!vS1Jo5giRZSjUg+lU|6E!aRnzs~DWKqY5c zJP7cc06}W@wgl-(P~5~qbQ6m3Cy_AS@Xni2MmG_{s@WTe(oHmon;3edn;7*U^TwL_ z4G|dgM);Fv!m{#)_Y)*^Kwb-Qj*2E>q9q^za~dOrOUxUJf)6bM5@}?KU4z!4IwCBUR(%xW61Gu7D!XBCf@$b|C*ft|YP}(&+ z-`;9-X$G%OJ8R@am(>F41NdfXeH<=;eUY-#B>jCB?OmpIiHB4h-95>^?*2|h?rcmOx>8fPG)OhAQC(F`qd}zaE_ZWhv+<*G=mefAx!{Pl;~0MFv<;OU=|KIF z9Lt^}L1FCCQIX5@+q2D2y689WuO`3-nfNRnQ(lw;Ou>gv8cAlcMRPSe*26%fj+0TV-V@v09L*Y=r^C z$4;gm9f6~xx5t|Ck zQa%szxLK;4F`qTArFvg^`qGcG2UpC01oU8{?C@C+IF29AWB_f&-Uc^N;XMYhNf-CG zP^7v-s!s|2O8K5?GG{VS7fuE5pBB7?Njl?*lYAzcrs#{Brm~NmN*33)t!mp=*mJWt z*Fm`6P20G`lMwOu4^ohe?a{G8> zBjuOQ4fVJR!2f z)0BJsXBE}HdpvI;PC1=Em$!Lw4}HF6jv@opXW6h$TuO(zWU0&>I_W*@6Ywe?0?(-) ze#ktWaJ)V1{5A%CU4&v6lw?lVG|L*WBS&8`SxkV+|Lq6=ipdH*r&tJut>_K7YH$dK z{DK9=aGdwFUD(h{@agzCsF1H@h$C(kZK9W@38fidlwLCbi@Mw7Qqkmzy3KeAPK5KE zM^CS;FuBgdi-6c4$HB$`tS)X3MO-_C6|c29?S&cMXme?>)J|-eh*#alv>d&!u4m|A zWraOfGx_MngPO@V%^~I~hAoa-OfjA9v!c#fTGjyW*Aq)Y>zT~hT0F0I488i9dYT!v UW9X$b^lvbp2aQ!rub%_|2Tc+sxBvhE literal 0 HcmV?d00001 diff --git a/tests/test_nc_to_mmd.py b/tests/test_nc_to_mmd.py index dcdd925c..76d98769 100644 --- a/tests/test_nc_to_mmd.py +++ b/tests/test_nc_to_mmd.py @@ -8,34 +8,38 @@ """ +import json import os import pathlib import tempfile -import yaml -import warnings import unittest -import pytest -import json +import warnings +from unittest.mock import patch import numpy as np - +import pytest +import yaml from dateutil.parser import isoparse from filehash import FileHash from lxml import etree from netCDF4 import Dataset from pkg_resources import resource_string -from unittest.mock import patch - -from py_mmd_tools.nc_to_mmd import Nc_to_mmd, normalize_iso8601, normalize_iso8601_0 -from py_mmd_tools.nc_to_mmd import valid_url -from py_mmd_tools.nc_to_mmd import get_short_and_long_names -from py_mmd_tools.nc_to_mmd import nc_wrapper -from py_mmd_tools.yaml_to_adoc import nc_attrs_from_yaml -from py_mmd_tools.yaml_to_adoc import required -from py_mmd_tools.yaml_to_adoc import repetition_allowed -from py_mmd_tools.yaml_to_adoc import set_attribute -from py_mmd_tools.yaml_to_adoc import set_attributes +from py_mmd_tools.nc_to_mmd import ( + Nc_to_mmd, + get_short_and_long_names, + nc_wrapper, + normalize_iso8601, + normalize_iso8601_0, + valid_url, +) +from py_mmd_tools.yaml_to_adoc import ( + nc_attrs_from_yaml, + repetition_allowed, + required, + set_attribute, + set_attributes, +) from tests.test_nc2mmd_script import patchedDataset warnings.simplefilter("ignore", ResourceWarning) @@ -43,38 +47,57 @@ @pytest.mark.py_mmd_tools def test_parent_keyword_arg(dataDir): - """ Test that a parent uuid can be provided to the to_mmd + """Test that a parent uuid can be provided to the to_mmd function, and that it is verified that the uuid part of the ID is a UUID. """ # UUID is UUID - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) req, msg = md.to_mmd(parent="no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb") assert req is True # ID is not correctly composed - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) with pytest.raises(ValueError) as ve: req, msg = md.to_mmd(parent="not-a-uuid") assert str(ve.value) == "parent must be composed as :" # naming_authority is not valid - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) with pytest.raises(ValueError) as ve: req, msg = md.to_mmd(parent="no.kvet:not-a-uuid") assert str(ve.value) == "naming_authority ACDD attribute no.kvet is not valid" # UUID is not UUID - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) with pytest.raises(ValueError) as ve: req, msg = md.to_mmd(parent="no.met:not-a-uuid") assert str(ve.value) == "UUID part of the parent ID is not valid" + # pre-existing parent + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc_withparent.nc"), check_only=True) + req, msg = md.to_mmd(parent="no.met:654e8acf-77b1-4f53-b6bf-0cd6cf94e646") + # print(md.missing_attributes["warnings"]) + assert "no need to update" in md.missing_attributes["warnings"][2] + + # pre-existing parent but malformed + md = Nc_to_mmd( + os.path.join(dataDir, "reference_nc_withparentmalformed.nc"), check_only=True + ) + with pytest.raises(ValueError) as ve: + req, msg = md.to_mmd(parent="no.met:654e8acf-77b1-4f53-b6bf-0cd6cf94e646") + assert str(ve.value) == "naming_authority ACDD attribute no.kvet is not valid" + assert "parent reference already in place" in md.missing_attributes["warnings"][1] + + # pre-existing parent but different + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc_withparent.nc"), check_only=True) + req, msg = md.to_mmd(parent="no.met:bce66800-1722-495e-975b-6033abb4da7d") + assert "not updating" in md.missing_attributes["warnings"][2] + @pytest.mark.py_mmd_tools def test_file_location_in_overrides(dataDir): - """Test that over-riding the file location works as expected. - """ + """Test that over-riding the file location works as expected.""" md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) req, msg = md.to_mmd() assert md.metadata["storage_information"]["file_location"] == dataDir @@ -85,9 +108,8 @@ def test_file_location_in_overrides(dataDir): @pytest.mark.py_mmd_tools def test_platform_in_overrides(dataDir): - """Test that over-riding the platform attribute works as expected. - """ - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + """Test that over-riding the platform attribute works as expected.""" + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) req, msg = md.to_mmd( overrides={ "platform": { @@ -100,44 +122,51 @@ def test_platform_in_overrides(dataDir): "instrument": { "short_name": "AVHRR", "long_name": "Advanced Very High Resolution Radiometer", - "resource": "https://www.wmo-sat.info/oscar/instruments/view/avhrr"}}}) + "resource": "https://www.wmo-sat.info/oscar/instruments/view/avhrr", + }, + } + } + ) assert md.metadata["platform"][0]["short_name"] == "Metop-A" @pytest.mark.py_mmd_tools def test_get_landing_page_url(dataDir): md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) - assert md.get_dataset_landing_page_url() == "https://data.met.no/dataset/" \ - "b7cb7934-77ca-4439-812e-f560df3fe7eb" - md = Nc_to_mmd(os.path.join(dataDir, "reference_nc_missing_keywords_vocab.nc"), - check_only=True) + assert ( + md.get_dataset_landing_page_url() == "https://data.met.no/dataset/" + "b7cb7934-77ca-4439-812e-f560df3fe7eb" + ) + md = Nc_to_mmd( + os.path.join(dataDir, "reference_nc_missing_keywords_vocab.nc"), check_only=True + ) assert md.get_dataset_landing_page_url() == "https://data.fake.no/dummy" @pytest.mark.py_mmd_tools def test_license_missing(dataDir): - """ Test that an error is raised if the license attribute is missing. - """ + """Test that an error is raised if the license attribute is missing.""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) # nc_to_update.nc does not have license.. md = Nc_to_mmd(os.path.join(dataDir, "nc_to_update.nc"), check_only=True) - md.get_license(mmd_yaml['use_constraint'], md.ncin) + md.get_license(mmd_yaml["use_constraint"], md.ncin) assert md.missing_attributes["errors"][0] == 'ACDD attribute "license" is required' @pytest.mark.py_mmd_tools def test_separate_repeated(dataDir): - """ Test function Nc_to_mmd.separate_repeated - """ - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + """Test function Nc_to_mmd.separate_repeated""" + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.platform = ['"Basis": "Space-based Platforms"', - ' "Category": "Earth Observation Satellites"', - ' "Sub_Category": "Sentinel-1"', - ' "Short_Name": "Sentinel-1A"', - ' "Long_Name": "Sentinel-1A"'] + ncin.platform = [ + ""Basis": "Space-based Platforms"", + " "Category": "Earth Observation Satellites"", + " "Sub_Category": "Sentinel-1"", + " "Short_Name": "Sentinel-1A"", + " "Long_Name": "Sentinel-1A"", + ] with pytest.raises(AttributeError) as ee: md.separate_repeated(True, getattr(ncin, "platform")) assert str(ee.value) == "'list' object has no attribute 'split'" @@ -145,78 +174,87 @@ def test_separate_repeated(dataDir): @pytest.mark.py_mmd_tools def testNc_to_mmd_get_geographic_extent_polygon(dataDir): - """ Test that get_geographic_extent_polygon returns default crs if + """Test that get_geographic_extent_polygon returns default crs if geospatial_bounds_crs is missing. """ - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.geospatial_bounds = ("POLYGON ((59.01 1.23, 59.06 1.66, 59.10 2.09, 59.15 2.53, " - "59.19 2.96, 59.24 3.40, 59.28 3.84, 59.32 4.28, 59.35 4.72, " - "59.39 5.16, 59.43 5.61, 59.43 5.61, 59.57 5.57, 59.72 5.52, " - "59.87 5.48, 60.01 5.43, 60.16 5.39, 60.31 5.34, 60.46 5.29, " - "60.60 5.25, 60.75 5.20, 60.92 5.15, 60.92 5.15, 60.88 4.67, " - "60.84 4.21, 60.80 3.75, 60.76 3.29, 60.72 2.84, 60.68 2.38, " - "60.63 1.92, 60.59 1.47, 60.54 1.02, 60.49 0.56, 60.49 0.56, " - "60.33 0.64, 60.18 0.71, 60.03 0.77, 59.89 0.84, 59.74 0.90, " - "59.59 0.97, 59.45 1.03, 59.30 1.10, 59.15 1.16, 59.01 1.23))") + ncin.geospatial_bounds = ( + "POLYGON ((59.01 1.23, 59.06 1.66, 59.10 2.09, 59.15 2.53, " + "59.19 2.96, 59.24 3.40, 59.28 3.84, 59.32 4.28, 59.35 4.72, " + "59.39 5.16, 59.43 5.61, 59.43 5.61, 59.57 5.57, 59.72 5.52, " + "59.87 5.48, 60.01 5.43, 60.16 5.39, 60.31 5.34, 60.46 5.29, " + "60.60 5.25, 60.75 5.20, 60.92 5.15, 60.92 5.15, 60.88 4.67, " + "60.84 4.21, 60.80 3.75, 60.76 3.29, 60.72 2.84, 60.68 2.38, " + "60.63 1.92, 60.59 1.47, 60.54 1.02, 60.49 0.56, 60.49 0.56, " + "60.33 0.64, 60.18 0.71, 60.03 0.77, 59.89 0.84, 59.74 0.90, " + "59.59 0.97, 59.45 1.03, 59.30 1.10, 59.15 1.16, 59.01 1.23))" + ) mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + data = md.get_geographic_extent_polygon( + mmd_yaml["geographic_extent"].pop("polygon"), ncin ) - data = md.get_geographic_extent_polygon(mmd_yaml["geographic_extent"].pop("polygon"), ncin) assert data["srsName"] == "EPSG:4326" @pytest.mark.py_mmd_tools def test_get_related_dataset(dataDir): - """ Test get_related_dataset function. - """ + """Test get_related_dataset function.""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) # One related dataset - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.related_dataset = 'no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb (parent)' - data = md.get_related_dataset(mmd_yaml['related_dataset'], ncin) - assert data[0]['id'] == 'no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb' + ncin.related_dataset = "no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb (parent)" + data = md.get_related_dataset(mmd_yaml["related_dataset"], ncin) + assert data[0]["id"] == "no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb" ncin.close() # Two related datasets - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.related_dataset = ( - 'no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb (parent), ' - 'no.met:b7cb7934-78ca-4439-812e-f560df3fe7eb (auxiliary)') - data = md.get_related_dataset(mmd_yaml['related_dataset'], ncin) - assert data[0]['id'] == 'no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb' - assert data[1]['id'] == 'no.met:b7cb7934-78ca-4439-812e-f560df3fe7eb' + "no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb (parent), " + "no.met:b7cb7934-78ca-4439-812e-f560df3fe7eb (auxiliary)" + ) + data = md.get_related_dataset(mmd_yaml["related_dataset"], ncin) + assert data[0]["id"] == "no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb" + assert data[1]["id"] == "no.met:b7cb7934-78ca-4439-812e-f560df3fe7eb" ncin.close() # Malformed relation - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.related_dataset = 'no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb' - data = md.get_related_dataset(mmd_yaml['related_dataset'], ncin) - assert 'The global attribute "related_dataset" is malformed' in \ - md.missing_attributes['errors'][0] + ncin.related_dataset = "no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb" + data = md.get_related_dataset(mmd_yaml["related_dataset"], ncin) + assert ( + 'The global attribute "related_dataset" is malformed' + in md.missing_attributes["errors"][0] + ) ncin.close() # Invalid relation type - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.related_dataset = 'no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb (child)' - data = md.get_related_dataset(mmd_yaml['related_dataset'], ncin) - assert 'The dataset relation type must be either' in \ - md.missing_attributes['errors'][0] + ncin.related_dataset = "no.met:b7cb7934-77ca-4439-812e-f560df3fe7eb (child)" + data = md.get_related_dataset(mmd_yaml["related_dataset"], ncin) + assert ( + "The dataset relation type must be either" in md.missing_attributes["errors"][0] + ) ncin.close() # Invalid identifier pattern - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.related_dataset = 'b7cb7934-77ca-4439-812e-f560df3fe7eb (parent)' - data = md.get_related_dataset(mmd_yaml['related_dataset'], ncin) - assert 'missing naming_authority in the identifier' in \ - md.missing_attributes['errors'][0] + ncin.related_dataset = "b7cb7934-77ca-4439-812e-f560df3fe7eb (parent)" + data = md.get_related_dataset(mmd_yaml["related_dataset"], ncin) + assert ( + "missing naming_authority in the identifier" + in md.missing_attributes["errors"][0] + ) ncin.close() @@ -225,16 +263,16 @@ def test_invalid_opendap_url(dataDir): """Test that a warning is issued if the opendap url is not accessible. """ - test_in = os.path.join(dataDir, 'reference_nc.nc') - url = 'https://thredds.met.no/thredds/dodsC/reference_nc.nc' + test_in = os.path.join(dataDir, "reference_nc.nc") + url = "https://thredds.met.no/thredds/dodsC/reference_nc.nc" md = Nc_to_mmd(test_in, url, check_only=True) req, msg = md.to_mmd() - assert "Cannot access OPeNDAP stream" in md.missing_attributes['warnings'][2] + assert "Cannot access OPeNDAP stream" in md.missing_attributes["warnings"][2] @pytest.mark.py_mmd_tools def testNc_to_mmd_Get_acdd_metadata(dataDir): - """ Test that the if-check + """Test that the if-check 'default' in acdd_ext[acdd_ext_key].keys() @@ -254,17 +292,19 @@ def testNc_to_mmd_Get_acdd_metadata(dataDir): def test_checksum(monkeypatch): """Verify that the checksum created in nc_to_mmd.py is correct""" tested = tempfile.mkstemp()[1] - fn = os.path.abspath('tests/data/reference_nc.nc') - url = os.path.join('https://thredds.met.no/thredds/dodsC/', os.path.basename(fn)) + fn = os.path.abspath("tests/data/reference_nc.nc") + url = os.path.join("https://thredds.met.no/thredds/dodsC/", os.path.basename(fn)) with monkeypatch.context() as mp: - mp.setattr("py_mmd_tools.nc_to_mmd.Dataset", - lambda *args, **kwargs: patchedDataset(url, *args, **kwargs)) + mp.setattr( + "py_mmd_tools.nc_to_mmd.Dataset", + lambda *args, **kwargs: patchedDataset(url, *args, **kwargs), + ) md = Nc_to_mmd(fn, url, checksum_calculation=True, check_only=True) md.to_mmd() - checksum = md.metadata['storage_information']['checksum'] - with open(tested, 'w') as tt: - tt.write('%s *%s'%(checksum, fn)) - md5hasher = FileHash('md5') + checksum = md.metadata["storage_information"]["checksum"] + with open(tested, "w") as tt: + tt.write("%s *%s" % (checksum, fn)) + md5hasher = FileHash("md5") assert md5hasher.verify_checksums(tested)[0].hashes_match is True @@ -275,27 +315,35 @@ def test_create_mmd_1(monkeypatch): Please add new fields to test as needed.. """ tested = tempfile.mkstemp()[1] - fn = os.path.abspath('tests/data/reference_nc_with_altID_multiple.nc') - url = 'https://thredds.met.no/thredds/dodsC/reference_nc_with_altID_multiple.nc' + fn = os.path.abspath("tests/data/reference_nc_with_altID_multiple.nc") + url = "https://thredds.met.no/thredds/dodsC/reference_nc_with_altID_multiple.nc" with monkeypatch.context() as mp: - mp.setattr("py_mmd_tools.nc_to_mmd.Dataset", - lambda *args, **kwargs: patchedDataset(url, *args, **kwargs)) + mp.setattr( + "py_mmd_tools.nc_to_mmd.Dataset", + lambda *args, **kwargs: patchedDataset(url, *args, **kwargs), + ) md = Nc_to_mmd(fn, url, output_file=tested, checksum_calculation=True) md.to_mmd() - reference_xsd = os.path.join(os.environ['MMD_PATH'], 'xsd/mmd_strict.xsd') + reference_xsd = os.path.join(os.environ["MMD_PATH"], "xsd/mmd_strict.xsd") xsd_obj = etree.XMLSchema(etree.parse(reference_xsd)) xml_doc = etree.ElementTree(file=tested) valid = xsd_obj.validate(xml_doc) assert valid is True """ Check content of the xml_doc """ # alternate_identifier - assert xml_doc.getroot().find( - "{http://www.met.no/schema/mmd}alternate_identifier[@type='dummy_type']" - ).text == "dummy_id_no1" - assert xml_doc.getroot().find( - "{http://www.met.no/schema/mmd}alternate_identifier[@type='other_type']" - ).text == "dummy_id_no2" + assert ( + xml_doc.getroot() + .find("{http://www.met.no/schema/mmd}alternate_identifier[@type='dummy_type']") + .text + == "dummy_id_no1" + ) + assert ( + xml_doc.getroot() + .find("{http://www.met.no/schema/mmd}alternate_identifier[@type='other_type']") + .text + == "dummy_id_no2" + ) # platform platform = xml_doc.getroot().find("{http://www.met.no/schema/mmd}platform") assert platform.getchildren()[0].text == "SNPP" @@ -303,18 +351,20 @@ def test_create_mmd_1(monkeypatch): def test_create_and_validate_mmd_platform(monkeypatch): - """ Test that an MMD file for a netcdf with missing platform + """Test that an MMD file for a netcdf with missing platform short_name and long_name validates as long as the vocabulary is given.""" tested = tempfile.mkstemp()[1] - fn = os.path.abspath('tests/data/reference_nc_platform_names_missing.nc') - url = 'https://thredds.met.no/thredds/dodsC/reference_nc_platform_names_missing.nc' + fn = os.path.abspath("tests/data/reference_nc_platform_names_missing.nc") + url = "https://thredds.met.no/thredds/dodsC/reference_nc_platform_names_missing.nc" with monkeypatch.context() as mp: - mp.setattr("py_mmd_tools.nc_to_mmd.Dataset", - lambda *args, **kwargs: patchedDataset(url, *args, **kwargs)) + mp.setattr( + "py_mmd_tools.nc_to_mmd.Dataset", + lambda *args, **kwargs: patchedDataset(url, *args, **kwargs), + ) md = Nc_to_mmd(fn, url, output_file=tested, checksum_calculation=True) md.to_mmd() - reference_xsd = os.path.join(os.environ['MMD_PATH'], 'xsd/mmd_strict.xsd') + reference_xsd = os.path.join(os.environ["MMD_PATH"], "xsd/mmd_strict.xsd") xsd_obj = etree.XMLSchema(etree.parse(reference_xsd)) xml_doc = etree.ElementTree(file=tested) valid = xsd_obj.validate(xml_doc) @@ -324,92 +374,103 @@ def test_create_and_validate_mmd_platform(monkeypatch): @pytest.mark.py_mmd_tools def test_get_data_access_dict_with_wms(monkeypatch): """ToDo: Add docstring""" - netcdf_file = os.path.abspath('tests/data/reference_nc.nc') + netcdf_file = os.path.abspath("tests/data/reference_nc.nc") opendap_url = ( - 'https://thredds.met.no/thredds/dodsC/arcticdata/' - 'S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/' + "https://thredds.met.no/thredds/dodsC/arcticdata/" + "S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/" ) + netcdf_file kwargs = { "dataset_citation": { "author": "Some name to ensure that kwargs not required " - "by get_data_access_dict are allowed" + "by get_data_access_dict are allowed" }, "add_wms_data_access": True, } with monkeypatch.context() as mp: - mp.setattr("py_mmd_tools.nc_to_mmd.Dataset", - lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs)) + mp.setattr( + "py_mmd_tools.nc_to_mmd.Dataset", + lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs), + ) md = Nc_to_mmd(netcdf_file, opendap_url, check_only=True) ncin = Dataset(md.netcdf_file) data = md.get_data_access_dict(ncin, **kwargs) - assert data[0]['type'] == 'OPeNDAP' - assert data[1]['type'] == 'OGC WMS' - assert data[2]['type'] == 'HTTP' + assert data[0]["type"] == "OPeNDAP" + assert data[1]["type"] == "OGC WMS" + assert data[2]["type"] == "HTTP" @pytest.mark.py_mmd_tools def test_get_data_access_dict_with_custom_wms(monkeypatch): """WMS link is set.""" - netcdf_file = os.path.abspath('tests/data/reference_nc.nc') + netcdf_file = os.path.abspath("tests/data/reference_nc.nc") opendap_url = ( - 'https://thredds.met.no/thredds/dodsC/arcticdata/' - 'S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/' + "https://thredds.met.no/thredds/dodsC/arcticdata/" + "S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/" ) + netcdf_file with monkeypatch.context() as mp: - mp.setattr("py_mmd_tools.nc_to_mmd.Dataset", - lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs)) + mp.setattr( + "py_mmd_tools.nc_to_mmd.Dataset", + lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs), + ) md = Nc_to_mmd(netcdf_file, opendap_url, check_only=True) ncin = Dataset(md.netcdf_file) - data = md.get_data_access_dict(ncin, add_wms_data_access=True, - wms_link='http://test-link') - assert data[1]['type'] == 'OGC WMS' - assert 'http://test-link' in str(data[1]['resource']) - assert data[1]['wms_layers'] == ['M01', - 'M01_copy'] + data = md.get_data_access_dict( + ncin, add_wms_data_access=True, wms_link="http://test-link" + ) + assert data[1]["type"] == "OGC WMS" + assert "http://test-link" in str(data[1]["resource"]) + assert data[1]["wms_layers"] == ["M01", "M01_copy"] @pytest.mark.py_mmd_tools def test_get_data_access_dict_with_custom_wms_and_layer_names(monkeypatch): """WMS link and layer names are set.""" - netcdf_file = os.path.abspath('tests/data/reference_nc.nc') + netcdf_file = os.path.abspath("tests/data/reference_nc.nc") opendap_url = ( - 'https://thredds.met.no/thredds/dodsC/arcticdata/' - 'S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/' + "https://thredds.met.no/thredds/dodsC/arcticdata/" + "S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/" ) + netcdf_file with monkeypatch.context() as mp: - mp.setattr("py_mmd_tools.nc_to_mmd.Dataset", - lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs)) + mp.setattr( + "py_mmd_tools.nc_to_mmd.Dataset", + lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs), + ) md = Nc_to_mmd(netcdf_file, opendap_url, check_only=True) ncin = Dataset(md.netcdf_file) - data = md.get_data_access_dict(ncin, add_wms_data_access=True, - wms_link='http://test-link', - wms_layer_names=['layer_name_1', 'layer_name_2']) - assert data[1]['type'] == 'OGC WMS' - assert 'http://test-link' in str(data[1]['resource']) - assert data[1]['wms_layers'] == ['layer_name_1', 'layer_name_2'] + data = md.get_data_access_dict( + ncin, + add_wms_data_access=True, + wms_link="http://test-link", + wms_layer_names=["layer_name_1", "layer_name_2"], + ) + assert data[1]["type"] == "OGC WMS" + assert "http://test-link" in str(data[1]["resource"]) + assert data[1]["wms_layers"] == ["layer_name_1", "layer_name_2"] @pytest.mark.py_mmd_tools def test_get_data_access_dict(monkeypatch): """ToDo: Add docstring""" - netcdf_file = os.path.abspath('tests/data/reference_nc.nc') + netcdf_file = os.path.abspath("tests/data/reference_nc.nc") opendap_url = ( - 'https://thredds.met.no/thredds/dodsC/arcticdata/' - 'S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/' + "https://thredds.met.no/thredds/dodsC/arcticdata/" + "S2S_drift_TCD/SIDRIFT_S2S_SH/2019/07/31/" ) + netcdf_file with monkeypatch.context() as mp: - mp.setattr("py_mmd_tools.nc_to_mmd.Dataset", - lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs)) + mp.setattr( + "py_mmd_tools.nc_to_mmd.Dataset", + lambda *args, **kwargs: patchedDataset(opendap_url, *args, **kwargs), + ) md = Nc_to_mmd(netcdf_file, opendap_url, check_only=True) ncin = Dataset(md.netcdf_file) data = md.get_data_access_dict(ncin) - assert data[0]['type'] == 'OPeNDAP' - assert data[1]['type'] == 'HTTP' + assert data[0]["type"] == "OPeNDAP" + assert data[1]["type"] == "HTTP" @pytest.mark.py_mmd_tools def test_not_absolute_path(): - fn = 'tests/data/reference_nc.nc' + fn = "tests/data/reference_nc.nc" md = Nc_to_mmd(fn, check_only=True) assert os.path.isabs(md.netcdf_file) is True @@ -432,8 +493,10 @@ def test_get_operational_status(dataDir, monkeypatch): ncin.processing_level = "kjhhas" mmd_element["maxOccurs"] = "1" value = md.get_operational_status(mmd_element, ncin) - assert ("The ACDD attribute 'processing_level' in MMD attribute 'operational_status'" - in md.missing_attributes['errors'][0]) + assert ( + "The ACDD attribute 'processing_level' in MMD attribute 'operational_status'" + in md.missing_attributes["errors"][0] + ) # repetition of processing_level is allowed mmd_element["maxOccurs"] = "unbounded" @@ -461,7 +524,7 @@ def test_dataset_production_status(dataDir): mmd_element["maxOccurs"] = "1" value = md.get_dataset_production_status(mmd_element, ncin) assert "The ACDD attribute 'dataset_production_status'" - "must " in md.missing_attributes['errors'][0] + "must " in md.missing_attributes["errors"][0] # repetition of processing_level is allowed mmd_element["maxOccurs"] = "unbounded" @@ -484,7 +547,10 @@ def test_get_quality_control(dataDir): ncin.quality_control = "kjhhas" mmd_element["maxOccurs"] = "1" md.get_quality_control(mmd_element, ncin) - assert "The ACDD attribute 'quality_control' must " in md.missing_attributes['errors'][0] + assert ( + "The ACDD attribute 'quality_control' must " + in md.missing_attributes["errors"][0] + ) # repetition of processing_level is allowed mmd_element["maxOccurs"] = "unbounded" @@ -495,7 +561,6 @@ def test_get_quality_control(dataDir): @pytest.mark.py_mmd_tools def test_nc_wrapper_global_attrs(dataDir): - test_ncin = os.path.join(dataDir, "reference_nc.nc") test_ncin = Dataset(test_ncin) @@ -505,18 +570,19 @@ def test_nc_wrapper_global_attrs(dataDir): test_json_header = nc_wrapper(json.load(file)) for attr in test_ncin.ncattrs(): - assert test_ncin.getncattr(attr) == test_json_header.getncattr(attr), \ - (f"Divergence in global attribute between header and nc header at {attr}," - f" nc: {test_ncin.getncattr(attr)}, json: {test_json_header.getncattr(attr)}") + assert test_ncin.getncattr(attr) == test_json_header.getncattr(attr), ( + f"Divergence in global attribute between header and nc header at {attr}," + f" nc: {test_ncin.getncattr(attr)}, json: {test_json_header.getncattr(attr)}" + ) for attr in test_ncin.ncattrs(): - assert test_ncin.getncattr(attr) == test_json_header[attr], \ - (f"Divergence in global attribute between header and nc header at {attr}," - f" nc: {test_ncin.getncattr(attr)}, json: {test_json_header[attr]}") + assert test_ncin.getncattr(attr) == test_json_header[attr], ( + f"Divergence in global attribute between header and nc header at {attr}," + f" nc: {test_ncin.getncattr(attr)}, json: {test_json_header[attr]}" + ) @pytest.mark.py_mmd_tools def test_nc_wrapper_variable_attrs(dataDir): - test_ncin = os.path.join(dataDir, "reference_nc.nc") test_ncin = Dataset(test_ncin) @@ -530,7 +596,7 @@ def handle_type_comparison(a, b): if np.isnan(a) and np.isnan(b): return True else: - return np.abs(a-b) < 1e-8 + return np.abs(a - b) < 1e-8 if isinstance(a, (np.ndarray, list)) and isinstance(b, (np.ndarray, list)): return (a == b).all() @@ -539,33 +605,39 @@ def handle_type_comparison(a, b): for var, var_attrs in test_ncin.variables.items(): for attr in var_attrs.ncattrs(): - assert handle_type_comparison(var_attrs.getncattr(attr), - test_json_header.variables[var].getncattr(attr)), \ - (f"Divergence in variable attribute between json and nc header at {var}:{attr}," - f" nc: {var_attrs.getncattr(attr)} of type {type(var_attrs.getncattr(attr))}," - f"json: {test_json_header.variables[var].getncattr(attr)}" - f" of type {type(test_json_header.variables[var].getncattr(attr))}") + assert handle_type_comparison( + var_attrs.getncattr(attr), + test_json_header.variables[var].getncattr(attr), + ), ( + f"Divergence in variable attribute between json and nc header at {var}:{attr}," + f" nc: {var_attrs.getncattr(attr)} of type {type(var_attrs.getncattr(attr))}," + f"json: {test_json_header.variables[var].getncattr(attr)}" + f" of type {type(test_json_header.variables[var].getncattr(attr))}" + ) for var, var_attrs in test_ncin.variables.items(): for attr in var_attrs.ncattrs(): - assert handle_type_comparison(var_attrs.getncattr(attr), - test_json_header.variables[var][attr]), \ - (f"Divergence in variable attribute between json and nc header at {var}:{attr}," - f" nc: {var_attrs.getncattr(attr)} of type {type(var_attrs.getncattr(attr))}," - f"json: {test_json_header.variables[var][attr]}" - f" of type {type(test_json_header.variables[var][attr])}") + assert handle_type_comparison( + var_attrs.getncattr(attr), test_json_header.variables[var][attr] + ), ( + f"Divergence in variable attribute between json and nc header at {var}:{attr}," + f" nc: {var_attrs.getncattr(attr)} of type {type(var_attrs.getncattr(attr))}," + f"json: {test_json_header.variables[var][attr]}" + f" of type {type(test_json_header.variables[var][attr])}" + ) @pytest.mark.py_mmd_tools def test_json(dataDir, monkeypatch): - """TODO: add docstring - """ + """TODO: add docstring""" test_json_header = os.path.join(dataDir, "reference_nc_header.json") with open(test_json_header, "r") as file: test_json_header = json.load(file) - tmp = Nc_to_mmd(test_json_header, json_input=True, check_only=True, checksum_calculation=True) + tmp = Nc_to_mmd( + test_json_header, json_input=True, check_only=True, checksum_calculation=True + ) tmp.to_mmd() with monkeypatch.context() as mp: @@ -574,14 +646,22 @@ def test_json(dataDir, monkeypatch): mp.setattr(nc_wrapper, "getncattr", lambda *a, **k: ["what", "ever"]) test_json_header.pop("file_size") with pytest.raises(KeyError) as ee: - tmp = Nc_to_mmd(test_json_header, json_input=True, - opendap_url="https://thredds.met.no/etc", output_file="somefn.xml") + tmp = Nc_to_mmd( + test_json_header, + json_input=True, + opendap_url="https://thredds.met.no/etc", + output_file="somefn.xml", + ) assert "'file_size'" == str(ee.value) test_json_header.pop("archive_location") with pytest.raises(KeyError) as ee: - tmp = Nc_to_mmd(test_json_header, json_input=True, - opendap_url="https://thredds.met.no/etc", output_file="somefn.xml") + tmp = Nc_to_mmd( + test_json_header, + json_input=True, + opendap_url="https://thredds.met.no/etc", + output_file="somefn.xml", + ) assert "'archive_location'" in str(ee.value) @@ -616,12 +696,15 @@ def test_nc_wrapper_ncatters(dataDir): with open(test_json_header, "r") as file: test_json_header = nc_wrapper(json.load(file)) - assert test_json_header.ncattrs() == test_ncin.ncattrs(), \ + assert test_json_header.ncattrs() == test_ncin.ncattrs(), ( "Keys in global ncattrs does not match between json and nc." + ) for var in test_ncin.variables: - assert test_ncin.variables[var].ncattrs() == test_json_header.variables[var].ncattrs(), \ - f"Mismatch in variable attributes, for variable {var}" + assert ( + test_ncin.variables[var].ncattrs() + == test_json_header.variables[var].ncattrs() + ), f"Mismatch in variable attributes, for variable {var}" @pytest.mark.py_mmd_tools @@ -638,244 +721,257 @@ def test_attribute_error_title_json(dataDir): @pytest.mark.py_mmd_tools def test_get_CFSTDN_keywords(dataDir): - """ Test that get_CFSTDN_keywords returns a list with one CF + """Test that get_CFSTDN_keywords returns a list with one CF standard name from the test file. """ - md = Nc_to_mmd(os.path.join(dataDir, 'reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.join(dataDir, "reference_nc.nc"), check_only=True) vars = md.get_CFSTDN_keywords(md.ncin) assert len(vars) == 1 assert vars[0] == "toa_bidirectional_reflectance" class TestNCAttrsFromYaml(unittest.TestCase): - def setUp(self): - """ Sets up class with an attribute `mmd_yaml` containing + """Sets up class with an attribute `mmd_yaml` containing the MMD to ACDD translations. """ self.maxDiff = None self.mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) self.attributes = {} - self.attributes['acdd'] = {} - self.attributes['acdd']['required'] = [] - self.attributes['acdd']['not_required'] = [] - self.attributes['acdd_ext'] = {} - self.attributes['acdd_ext']['required'] = [] - self.attributes['acdd_ext']['not_required'] = [] + self.attributes["acdd"] = {} + self.attributes["acdd"]["required"] = [] + self.attributes["acdd"]["not_required"] = [] + self.attributes["acdd_ext"] = {} + self.attributes["acdd_ext"]["required"] = [] + self.attributes["acdd_ext"]["not_required"] = [] def test_set_attribute__wrong_input(self): - """ Test that errors are raised in case of wrong input to the + """Test that errors are raised in case of wrong input to the set_attribute method. """ - mmd_field = 'keywords' - key = 'maxOccurs' + mmd_field = "keywords" + key = "maxOccurs" val = self.mmd_yaml[mmd_field][key] - convention = 'acdd' + convention = "acdd" with self.assertRaises(ValueError): - set_attribute(mmd_field, val, convention, self.attributes, req='not_required') + set_attribute( + mmd_field, val, convention, self.attributes, req="not_required" + ) def test_set_attribute__no_convention(self): - """ Test the set_attribute method when no convention is + """Test the set_attribute method when no convention is defined in mmd_yaml. """ - mmd_field = 'alternate_identifier' + mmd_field = "alternate_identifier" val = self.mmd_yaml[mmd_field] - convention = 'acdd' - set_attribute(mmd_field, val, convention, self.attributes, req='not_required') - self.assertFalse(bool(self.attributes['acdd']['not_required'])) + convention = "acdd" + set_attribute(mmd_field, val, convention, self.attributes, req="not_required") + self.assertFalse(bool(self.attributes["acdd"]["not_required"])) def test_set_attribute__one_conv_field(self): - """ Test the set_attribute method when one convention field + """Test the set_attribute method when one convention field is provided in mmd_yaml. Tests both acdd and acdd_ext. """ # Test acdd_ext - mmd_field = 'dataset_production_status' + mmd_field = "dataset_production_status" val = self.mmd_yaml[mmd_field] - convention = 'acdd_ext' - set_attribute(mmd_field, val, convention, self.attributes, req='not_required') - self.assertEqual(self.attributes['acdd_ext']['not_required'][0], { - 'attribute': 'dataset_production_status', - 'comment': 'No repetition allowed.', - 'default': 'Complete', - 'description': - 'Production status for the dataset, using a controlled ' - 'vocabulary. The valid keywords are listed in ' - 'https://htmlpreview.github.io/?https://github.com/metno/mmd/blob/' - 'master/doc/mmd-specification.html#dataset-production-status-types[section ' + convention = "acdd_ext" + set_attribute(mmd_field, val, convention, self.attributes, req="not_required") + self.assertEqual( + self.attributes["acdd_ext"]["not_required"][0], + { + "attribute": "dataset_production_status", + "comment": "No repetition allowed.", + "default": "Complete", + "description": "Production status for the dataset, using a controlled " + "vocabulary. The valid keywords are listed in " + "https://htmlpreview.github.io/?https://github.com/metno/mmd/blob/" + "master/doc/mmd-specification.html#dataset-production-status-types[section " '4.2 of the MMD specification]. If set as "In Work", remember ' - 'that end_date in ' - 'https://htmlpreview.github.io/?https://github.com/metno/mmd/blob/' - 'master/doc/mmd-specification.html#temporal_extent[section ' - '2.8 of the MMD specification] can (should) be empty.', - 'mmd_field': 'dataset_production_status', - 'recommended': True, - 'repetition_allowed': False, - 'separator': '', - }) + "that end_date in " + "https://htmlpreview.github.io/?https://github.com/metno/mmd/blob/" + "master/doc/mmd-specification.html#temporal_extent[section " + "2.8 of the MMD specification] can (should) be empty.", + "mmd_field": "dataset_production_status", + "recommended": True, + "repetition_allowed": False, + "separator": "", + }, + ) # Test acdd - mmd_field = 'operational_status' + mmd_field = "operational_status" val = self.mmd_yaml[mmd_field] - convention = 'acdd' - set_attribute(mmd_field, val, convention, self.attributes, req='not_required') - self.assertEqual(self.attributes['acdd']['not_required'][0], { - 'attribute': 'processing_level', - 'comment': 'Optional', - 'default': '', - 'description': - 'A textual description of the processing ' - 'level of the data. Valid keywords are listed in ' - 'https://htmlpreview.github.io/?https://github.com/metno/mmd/blob/' - 'master/doc/mmd-specification.html#operational-status[Section ' - '4.5 of the MMD specification].', - 'mmd_field': 'operational_status', - 'recommended': True, - 'repetition_allowed': False, - 'separator': '', - }) + convention = "acdd" + set_attribute(mmd_field, val, convention, self.attributes, req="not_required") + self.assertEqual( + self.attributes["acdd"]["not_required"][0], + { + "attribute": "processing_level", + "comment": "Optional", + "default": "", + "description": "A textual description of the processing " + "level of the data. Valid keywords are listed in " + "https://htmlpreview.github.io/?https://github.com/metno/mmd/blob/" + "master/doc/mmd-specification.html#operational-status[Section " + "4.5 of the MMD specification].", + "mmd_field": "operational_status", + "recommended": True, + "repetition_allowed": False, + "separator": "", + }, + ) def test_set_attribute__two_conv_fields(self): - """ Test the set_attribute method when two convention fields + """Test the set_attribute method when two convention fields are provided in mmd_yaml. """ - mmd_field = 'metadata_identifier' + mmd_field = "metadata_identifier" val = self.mmd_yaml[mmd_field] - convention = 'acdd' - set_attribute(mmd_field, val, convention, self.attributes, req='required') - self.assertEqual(self.attributes['acdd']['required'][0], { - 'attribute': 'id', - 'comment': 'Required, and should be UUID. No repetition allowed.', - 'default': '', - 'description': - 'An identifier for the dataset, provided by and unique within ' + convention = "acdd" + set_attribute(mmd_field, val, convention, self.attributes, req="required") + self.assertEqual( + self.attributes["acdd"]["required"][0], + { + "attribute": "id", + "comment": "Required, and should be UUID. No repetition allowed.", + "default": "", + "description": "An identifier for the dataset, provided by and unique within " 'its naming authority. The combination of the "naming ' 'authority" and the "id" should be globally unique, but the id ' - 'can be globally unique by itself also. A uuid is recommended.', - 'mmd_field': 'metadata_identifier', - 'recommended': True, - 'repetition_allowed': False, - 'separator': '', - }) - self.assertEqual(self.attributes['acdd']['required'][1], { - 'attribute': 'naming_authority', - 'comment': 'Required. We recommend using reverse-DNS naming. ' - 'No repetition allowed.', - 'default': '', - 'description': - 'The organisation that provides the initial id (see above) for ' - 'the dataset. The naming authority should be uniquely ' - 'specified by this attribute. We recommend using reverse-DNS ' - 'naming for the naming authority.', - 'mmd_field': 'metadata_identifier', - 'recommended': True, - 'repetition_allowed': False, - 'separator': '', - }) + "can be globally unique by itself also. A uuid is recommended.", + "mmd_field": "metadata_identifier", + "recommended": True, + "repetition_allowed": False, + "separator": "", + }, + ) + self.assertEqual( + self.attributes["acdd"]["required"][1], + { + "attribute": "naming_authority", + "comment": "Required. We recommend using reverse-DNS naming. " + "No repetition allowed.", + "default": "", + "description": "The organisation that provides the initial id (see above) for " + "the dataset. The naming authority should be uniquely " + "specified by this attribute. We recommend using reverse-DNS " + "naming for the naming authority.", + "mmd_field": "metadata_identifier", + "recommended": True, + "repetition_allowed": False, + "separator": "", + }, + ) def test_set_attribute__required_not_req(self): - """ Test the set_attribute method when the + """Test the set_attribute method when the attributes[convention]['required'] field should be populated but the convention is not required. """ - mmd_field = 'operational_status' + mmd_field = "operational_status" val = self.mmd_yaml[mmd_field] - convention = 'acdd' - set_attribute(mmd_field, val, convention, self.attributes, req='required') - self.assertEqual(self.attributes['acdd']['required'], []) - self.assertEqual(self.attributes['acdd']['not_required'], []) + convention = "acdd" + set_attribute(mmd_field, val, convention, self.attributes, req="required") + self.assertEqual(self.attributes["acdd"]["required"], []) + self.assertEqual(self.attributes["acdd"]["not_required"], []) def test_set_attribute__not_required_req(self): - """ Test the set_attribute method when the + """Test the set_attribute method when the attributes[convention]['not_required'] field should be populated but the convention is required. """ - mmd_field = 'iso_topic_category' + mmd_field = "iso_topic_category" val = self.mmd_yaml[mmd_field] - convention = 'acdd_ext' - set_attribute(mmd_field, val, convention, self.attributes, req='not_required') - self.assertEqual(self.attributes['acdd']['required'], []) - self.assertEqual(self.attributes['acdd']['not_required'], []) + convention = "acdd_ext" + set_attribute(mmd_field, val, convention, self.attributes, req="not_required") + self.assertEqual(self.attributes["acdd"]["required"], []) + self.assertEqual(self.attributes["acdd"]["not_required"], []) def test_set_attributes__single(self): - mmd_field = 'metadata_identifier' + mmd_field = "metadata_identifier" val = self.mmd_yaml[mmd_field] set_attributes(mmd_field, val, self.attributes) - self.assertEqual(self.attributes['acdd']['required'][0], { - 'attribute': 'id', - 'comment': 'Required, and should be UUID. No repetition allowed.', - 'default': '', - 'description': - 'An identifier for the dataset, provided by and unique within ' + self.assertEqual( + self.attributes["acdd"]["required"][0], + { + "attribute": "id", + "comment": "Required, and should be UUID. No repetition allowed.", + "default": "", + "description": "An identifier for the dataset, provided by and unique within " 'its naming authority. The combination of the "naming ' 'authority" and the "id" should be globally unique, but the id ' - 'can be globally unique by itself also. A uuid is recommended.', - 'mmd_field': 'metadata_identifier', - 'recommended': True, - 'repetition_allowed': False, - 'separator': '', - }) + "can be globally unique by itself also. A uuid is recommended.", + "mmd_field": "metadata_identifier", + "recommended": True, + "repetition_allowed": False, + "separator": "", + }, + ) def test_set_attributes__nested(self): - mmd_field = 'keywords' + mmd_field = "keywords" val = self.mmd_yaml[mmd_field] set_attributes(mmd_field, val, self.attributes) - self.assertEqual(self.attributes['acdd']['required'][0], { - 'attribute': 'keywords', - 'comment': 'Comma separated list.', - 'default': '', - 'description': - 'A comma-separated list of keywords and/or ' - 'phrases. Keywords may be common words or phrases, ' - 'terms from a controlled vocabulary (GCMD is ' - 'required), or URIs for terms from a controlled ' + self.assertEqual( + self.attributes["acdd"]["required"][0], + { + "attribute": "keywords", + "comment": "Comma separated list.", + "default": "", + "description": "A comma-separated list of keywords and/or " + "phrases. Keywords may be common words or phrases, " + "terms from a controlled vocabulary (GCMD is " + "required), or URIs for terms from a controlled " 'vocabulary (see also "keywords_vocabulary" ' - 'attribute). If keywords are extracted from, e.g., ' - 'GCMD Science Keywords, add ' + "attribute). If keywords are extracted from, e.g., " + "GCMD Science Keywords, add " 'keywords_vocabulary="GCMDSK" and prefix in any case ' - 'each keyword with the appropriate prefix.', - 'mmd_field': 'keywords>keyword', - 'recommended': True, - 'repetition_allowed': True, - 'separator': ',', - }) + "each keyword with the appropriate prefix.", + "mmd_field": "keywords>keyword", + "recommended": True, + "repetition_allowed": True, + "separator": ",", + }, + ) def test_required__is_required(self): - """ Test method required on a required MMD field. - """ - self.assertTrue(required(self.mmd_yaml['temporal_extent']['start_date'])) + """Test method required on a required MMD field.""" + self.assertTrue(required(self.mmd_yaml["temporal_extent"]["start_date"])) def test_required__not_required(self): - """ Test method required on a not required MMD field. - """ - self.assertFalse(required(self.mmd_yaml['temporal_extent']['end_date'])) + """Test method required on a not required MMD field.""" + self.assertFalse(required(self.mmd_yaml["temporal_extent"]["end_date"])) def test_required__minOccurs_missing(self): - """ Test method required on an MMD field where the minOccurs + """Test method required on an MMD field where the minOccurs key is missing. """ - sd = self.mmd_yaml['temporal_extent']['start_date'] - sd.pop('minOccurs') + sd = self.mmd_yaml["temporal_extent"]["start_date"] + sd.pop("minOccurs") self.assertFalse(required(sd)) def test_repetition_allowed(self): - """ Test method repetition_allowed with an MMD field which + """Test method repetition_allowed with an MMD field which can allows repetition. """ - self.assertTrue(repetition_allowed(self.mmd_yaml['temporal_extent']['start_date'])) + self.assertTrue( + repetition_allowed(self.mmd_yaml["temporal_extent"]["start_date"]) + ) def test_repetition_allowed__not_allowed(self): - """ Test method repetition_allowed with an MMD field which + """Test method repetition_allowed with an MMD field which can does not allow repetition. """ - self.assertFalse(repetition_allowed(self.mmd_yaml['geographic_extent'])) + self.assertFalse(repetition_allowed(self.mmd_yaml["geographic_extent"])) def test_repetition_allowed__maxOccurs_missing(self): - """ Test method repetition_allowed with an MMD field where + """Test method repetition_allowed with an MMD field where the maxOccurs key is missing. """ - sd = self.mmd_yaml['temporal_extent']['start_date'] - sd.pop('maxOccurs') + sd = self.mmd_yaml["temporal_extent"]["start_date"] + sd.pop("maxOccurs") self.assertTrue(repetition_allowed(sd)) def test_nc_attrs_from_yaml(self): @@ -884,15 +980,15 @@ def test_nc_attrs_from_yaml(self): self.assertEqual(type(adoc), str) def test_creator_role_acdd_extension(self): - mmd_field = 'personnel' + mmd_field = "personnel" val = self.mmd_yaml[mmd_field] set_attributes(mmd_field, val, self.attributes) - self.assertEqual(self.attributes['acdd_ext']['not_required'][0]['attribute'], - 'creator_role') + self.assertEqual( + self.attributes["acdd_ext"]["not_required"][0]["attribute"], "creator_role" + ) class TestNC2MMD(unittest.TestCase): - def setUp(self): """ToDo: Add docstring""" # @@ -903,10 +999,10 @@ def setUp(self): # unset the output limit when printing the xml diff # current_dir = pathlib.Path.cwd() - self.reference_nc = str(current_dir / 'tests' / 'data' / 'reference_nc.nc') - self.fail_nc = str(current_dir / 'tests' / 'data' / 'reference_nc_fail.nc') - self.reference_xml = str(current_dir / 'tests' / 'data' / 'reference_nc.xml') - self.reference_xsd = os.path.join(os.environ['MMD_PATH'], 'xsd/mmd_strict.xsd') + self.reference_nc = str(current_dir / "tests" / "data" / "reference_nc.nc") + self.fail_nc = str(current_dir / "tests" / "data" / "reference_nc_fail.nc") + self.reference_xml = str(current_dir / "tests" / "data" / "reference_nc.xml") + self.reference_xsd = os.path.join(os.environ["MMD_PATH"], "xsd/mmd_strict.xsd") # @patch('py_mmd_utils.nc_to_mmd.Nc_to_mmd.__init__') # @patch('mmd_utils.nc_to_mmd.Nc_to_mmd.to_mmd') @@ -916,38 +1012,39 @@ def setUp(self): # self.assertTrue(mock_to_mmd.called) def test_valid_url(self): - self.assertTrue(valid_url('http://www.google.com')) - self.assertTrue(valid_url('http://spdx.org/licenses/CC-BY-4.0')) - self.assertFalse(valid_url('www.google.com')) + self.assertTrue(valid_url("http://www.google.com")) + self.assertTrue(valid_url("http://spdx.org/licenses/CC-BY-4.0")) + self.assertFalse(valid_url("www.google.com")) self.assertFalse(valid_url(None)) def test_license__deprecated_attrs(self): mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.license = "CC-BY-4.0" ncin.license_resource = "http://spdx.org/licenses/CC-BY-4.0" - value = md.get_license(mmd_yaml['use_constraint'], ncin) - self.assertEqual(value['resource'], 'http://spdx.org/licenses/CC-BY-4.0') - self.assertEqual(value['identifier'], 'CC-BY-4.0') + value = md.get_license(mmd_yaml["use_constraint"], ncin) + self.assertEqual(value["resource"], "http://spdx.org/licenses/CC-BY-4.0") + self.assertEqual(value["identifier"], "CC-BY-4.0") self.assertEqual( md.missing_attributes["warnings"][0], - '"license_resource" is a deprecated attribute') + '"license_resource" is a deprecated attribute', + ) def test_license__invalid_url(self): mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.license = "spdx.org/licenses/CC-BY-4.0" - value = md.get_license(mmd_yaml['use_constraint'], ncin) + value = md.get_license(mmd_yaml["use_constraint"], ncin) self.assertIsNone(value) self.assertEqual( md.missing_attributes["errors"][0], - '"spdx.org/licenses/CC-BY-4.0" is not a valid url' + '"spdx.org/licenses/CC-BY-4.0" is not a valid url', ) def test_license__basic(self): @@ -955,29 +1052,28 @@ def test_license__basic(self): identifier is accepted and parsed correctly. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.license = "http://spdx.org/licenses/CC-BY-4.0 (CC-BY-4.0)" - value = md.get_license(mmd_yaml['use_constraint'], ncin) - self.assertEqual(value['resource'], 'http://spdx.org/licenses/CC-BY-4.0') - self.assertEqual(value['identifier'], 'CC-BY-4.0') + value = md.get_license(mmd_yaml["use_constraint"], ncin) + self.assertEqual(value["resource"], "http://spdx.org/licenses/CC-BY-4.0") + self.assertEqual(value["identifier"], "CC-BY-4.0") def test_license__simple(self): - """Test that a license with valid url only is accepted and parsed correctly. - """ + """Test that a license with valid url only is accepted and parsed correctly.""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.license = "http://spdx.org/licenses/CC-BY-4.0" - value = md.get_license(mmd_yaml['use_constraint'], ncin) + value = md.get_license(mmd_yaml["use_constraint"], ncin) # met-vocab-tools should be able to find the license by # searching a url (see issue https://github.com/metno/met-vocab-tools/issues/25): # self.assertEqual(value['identifier'], 'CC-BY-4.0') - self.assertEqual(value['resource'], 'http://spdx.org/licenses/CC-BY-4.0') + self.assertEqual(value["resource"], "http://spdx.org/licenses/CC-BY-4.0") self.assertEqual(len(list(value.keys())), 2) def test_license__only_url_but_not_standard(self): @@ -985,301 +1081,333 @@ def test_license__only_url_but_not_standard(self): license causes an error. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.license = "http://spdx.org/licenses/CC-BY-4.1" - value = md.get_license(mmd_yaml['use_constraint'], ncin) + value = md.get_license(mmd_yaml["use_constraint"], ncin) # met-vocab-tools should be able to find the license by # searching a url (see issue https://github.com/metno/met-vocab-tools/issues/25): # self.assertEqual(value['identifier'], 'CC-BY-4.0') - self.assertEqual(value['resource'], 'http://spdx.org/licenses/CC-BY-4.1') + self.assertEqual(value["resource"], "http://spdx.org/licenses/CC-BY-4.1") self.assertEqual(len(list(value.keys())), 1) - self.assertEqual(md.missing_attributes['errors'][0], - 'license should be provided as ()') + self.assertEqual( + md.missing_attributes["errors"][0], + "license should be provided as ()", + ) def test_license__according_to_adc1(self): - """Test that a license passed as url(identifier) is accepted and parsed correctly. - """ + """Test that a license passed as url(identifier) is accepted and parsed correctly.""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.license = "http://spdx.org/licenses/CC-BY-4.0(CC-BY-4.0)" - value = md.get_license(mmd_yaml['use_constraint'], ncin) - self.assertEqual(value['resource'], 'http://spdx.org/licenses/CC-BY-4.0') - self.assertEqual(value['identifier'], 'CC-BY-4.0') + value = md.get_license(mmd_yaml["use_constraint"], ncin) + self.assertEqual(value["resource"], "http://spdx.org/licenses/CC-BY-4.0") + self.assertEqual(value["identifier"], "CC-BY-4.0") def test_license__according_to_adc2(self): - """Test that a license passed as url (identifier) is accepted and parsed correctly. - """ + """Test that a license passed as url (identifier) is accepted and parsed correctly.""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.license = "http://spdx.org/licenses/CC-BY-4.0 (CC-BY-4.0)" - value = md.get_license(mmd_yaml['use_constraint'], ncin) - self.assertEqual(value['resource'], 'http://spdx.org/licenses/CC-BY-4.0') - self.assertEqual(value['identifier'], 'CC-BY-4.0') + value = md.get_license(mmd_yaml["use_constraint"], ncin) + self.assertEqual(value["resource"], "http://spdx.org/licenses/CC-BY-4.0") + self.assertEqual(value["identifier"], "CC-BY-4.0") def test_license__not_standard(self): - """ Test that a license string that is not standard is added + """Test that a license string that is not standard is added as license_text. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.reference_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.license = "https://earth.esa.int/eogateway/documents/20142/1564626/" \ - "ESA-Data-Policy-ESA-PB-EO-2010-54.pdf (ESA earth observation data policy)" - value = md.get_license(mmd_yaml['use_constraint'], ncin) + ncin.license = ( + "https://earth.esa.int/eogateway/documents/20142/1564626/" + "ESA-Data-Policy-ESA-PB-EO-2010-54.pdf (ESA earth observation data policy)" + ) + value = md.get_license(mmd_yaml["use_constraint"], ncin) self.assertEqual( - value['license_text'], + value["license_text"], "https://earth.esa.int/eogateway/documents/20142/1564626/" - "ESA-Data-Policy-ESA-PB-EO-2010-54.pdf (ESA earth observation data policy)") + "ESA-Data-Policy-ESA-PB-EO-2010-54.pdf (ESA earth observation data policy)", + ) def test_init_raises_error(self): """Nc_to_mmd.__init__ should raise error if check_only=False, but output_file is None. Test that this is the case. """ with self.assertRaises(ValueError): - Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), output_file=None, - check_only=False) + Nc_to_mmd( + os.path.abspath("tests/data/reference_nc.nc"), + output_file=None, + check_only=False, + ) - @patch('metvocab.mmdgroup.MMDGroup.init_vocab') + @patch("metvocab.mmdgroup.MMDGroup.init_vocab") def test_init_raises_error_on_mmd_group(self, mock_init_vocab): """Nc_to_mmd.__init__ should raise error if mmdgroups are not initialised. """ with self.assertRaises(ValueError): - Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), output_file=None, - check_only=True) + Nc_to_mmd( + os.path.abspath("tests/data/reference_nc.nc"), + output_file=None, + check_only=True, + ) def test_default_when_no_acdd_or_acdd_ext(self): - """ Test that a default value can be used even if no acdd + """Test that a default value can be used even if no acdd or acdd_ext fields are present. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) value = md.get_acdd_metadata( - mmd_yaml['metadata_status'], - ncin, 'metadata_status' + mmd_yaml["metadata_status"], ncin, "metadata_status" ) - self.assertEqual(value, 'Active') + self.assertEqual(value, "Active") def test__get_acdd_metadata__dont_accept_alternatives(self): - """ Test that the function get_acdd_metadata raises an + """Test that the function get_acdd_metadata raises an error if there are several alternative acdd or acdd_ext fields. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) with self.assertRaises(ValueError) as e: md.get_acdd_metadata( - mmd_yaml['metadata_identifier'], - ncin, 'metadata_identifier' + mmd_yaml["metadata_identifier"], ncin, "metadata_identifier" ) - self.assertEqual(str(e.exception), - 'Multiple ACDD or ACCD extension fields provided.' - ' Please use another translation function.') + self.assertEqual( + str(e.exception), + "Multiple ACDD or ACCD extension fields provided." + " Please use another translation function.", + ) def test_get_acdd_metadata_uses_default_date_created_type(self): """Test that the get_acdd_metadata function uses default date_created_type.""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_metadata_updates(mmd_yaml['last_metadata_update'], ncin) - self.assertEqual(value['update'][0]['type'], 'Created') + value = md.get_metadata_updates(mmd_yaml["last_metadata_update"], ncin) + self.assertEqual(value["update"][0]["type"], "Created") def test_polygon_is_not_wkt(self): """The geospatial_bounds nc attribute may not be a proper wkt string. Test that this case is properly handled. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd(self.fail_nc, check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.geospatial_bounds = "" - md.get_geographic_extent_polygon( - mmd_yaml['geographic_extent']['polygon'], ncin - ) + md.get_geographic_extent_polygon(mmd_yaml["geographic_extent"]["polygon"], ncin) self.assertEqual( md.missing_attributes["errors"][0], - "geospatial_bounds must be formatted as a WKT string" + "geospatial_bounds must be formatted as a WKT string", ) def test_geographic_extent_polygon(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) value = md.get_geographic_extent_polygon( - mmd_yaml['geographic_extent']['polygon'], ncin + mmd_yaml["geographic_extent"]["polygon"], ncin ) - self.assertEqual(value['srsName'], 'EPSG:4326') - self.assertEqual(value['pos'][0], '69.0000 3.7900') + self.assertEqual(value["srsName"], "EPSG:4326") + self.assertEqual(value["pos"][0], "69.0000 3.7900") def test_missing_nc_attrs(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_attrs.nc"), check_only=True ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_attrs.nc'), - check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_acdd_metadata(mmd_yaml['geographic_extent'], ncin, 'geographic_extent') - self.assertEqual(value['rectangle']['north'], None) - value = md.get_data_centers(mmd_yaml['data_center'], ncin) + value = md.get_acdd_metadata( + mmd_yaml["geographic_extent"], ncin, "geographic_extent" + ) + self.assertEqual(value["rectangle"]["north"], None) + value = md.get_data_centers(mmd_yaml["data_center"], ncin) self.assertEqual( - md.missing_attributes['errors'][0], 'geospatial_lat_max is a required attribute' + md.missing_attributes["errors"][0], + "geospatial_lat_max is a required attribute", ) self.assertEqual( - md.missing_attributes['errors'][4], 'institution is a required attribute' + md.missing_attributes["errors"][4], "institution is a required attribute" ) def test_geographic_extent_rectangle_is_floatable(self): - """ Test that the provided geospatial coordinates can be + """Test that the provided geospatial coordinates can be converted to float. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file, "w", diskless=True) ncin.geospatial_lat_max = "60.158733f; // float" ncin.geospatial_lat_min = "59.78492f; // float" ncin.geospatial_lon_max = "10.944986f; // float" ncin.geospatial_lon_min = "10.508897f; // float" md.get_geographic_extent_rectangle( - mmd_yaml['geographic_extent']['rectangle'], ncin) + mmd_yaml["geographic_extent"]["rectangle"], ncin + ) self.assertEqual( - md.missing_attributes['errors'][0], - 'geospatial_lat_max must be convertible to float type.') + md.missing_attributes["errors"][0], + "geospatial_lat_max must be convertible to float type.", + ) def test_missing_geographic_extent_but_provided_as_kwarg(self): """Test that the geographic extent rectangle can be added as a kwarg. """ yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_rectangle.nc"), + check_only=True, + ) + md.to_mmd( + overrides={ + "geographic_extent_rectangle": { + "geospatial_lat_max": 90, + "geospatial_lat_min": -90, + "geospatial_lon_min": -180, + "geospatial_lon_max": 180, + } + } ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_rectangle.nc'), - check_only=True) - md.to_mmd(overrides={ - "geographic_extent_rectangle": {"geospatial_lat_max": 90, - "geospatial_lat_min": -90, - "geospatial_lon_min": -180, - "geospatial_lon_max": 180}}) - self.assertEqual(md.metadata['geographic_extent']['rectangle']['north'], 90) + self.assertEqual(md.metadata["geographic_extent"]["rectangle"]["north"], 90) def test_collection_is_not_list(self): """Test that an error is raised if the collection input parameter is wrong type. """ - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_attrs.nc'), - check_only=True) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_attrs.nc"), check_only=True + ) with self.assertRaises(ValueError) as e: md.to_mmd(collection=2) - self.assertEqual(str(e.exception), 'collection must be of type str') + self.assertEqual(str(e.exception), "collection must be of type str") def test_collection_not_set(self): """ToDo: Add docstring""" - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_collection.nc'), - check_only=True) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_collection.nc"), + check_only=True, + ) req_ok, msg = md.to_mmd() self.assertTrue(req_ok) - self.assertEqual(md.metadata['collection'], ['METNCS']) + self.assertEqual(md.metadata["collection"], ["METNCS"]) def test_collection_set(self): """ToDo: Add docstring""" - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) - status, msg = md.to_mmd(collection='ADC') + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) + status, msg = md.to_mmd(collection="ADC") # nc files should normally not have a collection element, as this is # set during harvesting self.assertTrue(status) - self.assertEqual(md.metadata['collection'], ['ADC']) + self.assertEqual(md.metadata["collection"], ["ADC"]) def test_abstract(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_abstracts(mmd_yaml['abstract'], ncin) + value = md.get_abstracts(mmd_yaml["abstract"], ncin) self.assertEqual(type(value), list) - self.assertTrue('lang' in value[0].keys()) - self.assertTrue('abstract' in value[0].keys()) + self.assertTrue("lang" in value[0].keys()) + self.assertTrue("abstract" in value[0].keys()) def test_title(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_titles(mmd_yaml['title'], ncin) + value = md.get_titles(mmd_yaml["title"], ncin) self.assertEqual(type(value), list) - self.assertTrue('lang' in value[0].keys()) - self.assertTrue('title' in value[0].keys()) + self.assertTrue("lang" in value[0].keys()) + self.assertTrue("title" in value[0].keys()) self.assertEqual( - value[0]['title'], - 'Direct Broadcast data processed in satellite swath to L1C.' + value[0]["title"], + "Direct Broadcast data processed in satellite swath to L1C.", ) - self.assertEqual(value[1]['title'], 'Norsk tittel') + self.assertEqual(value[1]["title"], "Norsk tittel") def test_title_one_language_only(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_id_missing.nc"), check_only=True ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_id_missing.nc'), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_titles(mmd_yaml['title'], ncin) + value = md.get_titles(mmd_yaml["title"], ncin) self.assertEqual(type(value), list) - self.assertTrue('lang' in value[0].keys()) - self.assertTrue('title' in value[0].keys()) + self.assertTrue("lang" in value[0].keys()) + self.assertTrue("title" in value[0].keys()) self.assertEqual( - md.missing_attributes['errors'][0], - "title_no is a required attribute") + md.missing_attributes["errors"][0], "title_no is a required attribute" + ) def test_data_center(self): """Test get_data_centers function""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_data_centers(mmd_yaml['data_center'], ncin) + value = md.get_data_centers(mmd_yaml["data_center"], ncin) self.assertEqual(type(value), list) - self.assertEqual(value, [{ - 'data_center_name': { - 'long_name': 'Norwegian Meteorological Institute', - 'short_name': 'MET Norway' - }, - 'data_center_url': 'met.no', - }]) + self.assertEqual( + value, + [ + { + "data_center_name": { + "long_name": "Norwegian Meteorological Institute", + "short_name": "MET Norway", + }, + "data_center_url": "met.no", + } + ], + ) def test_data_access(self): """ToDo: Add docstring""" - yaml.load(resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + yaml.load( + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) Dataset(md.netcdf_file) value = None self.assertEqual(value, None) @@ -1287,14 +1415,14 @@ def test_data_access(self): def test_dataset_production_status(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) value = md.get_acdd_metadata( - mmd_yaml['dataset_production_status'], ncin, 'dataset_production_status' + mmd_yaml["dataset_production_status"], ncin, "dataset_production_status" ) - self.assertEqual(value, 'In Work') + self.assertEqual(value, "In Work") def test_alternate_identifier_missing(self): """Test that get_alternate_identifier returns an empty list @@ -1302,11 +1430,11 @@ def test_alternate_identifier_missing(self): attributes of the nc-file. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_alternate_identifier(mmd_yaml['alternate_identifier'], ncin) + value = md.get_alternate_identifier(mmd_yaml["alternate_identifier"], ncin) self.assertEqual(len(value), 0) def test_alternate_identifier_wrong_format(self): @@ -1314,18 +1442,18 @@ def test_alternate_identifier_wrong_format(self): is missing the type between parentheses. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_with_altID.nc'), check_only=True) - ncin = Dataset(md.netcdf_file, "w", diskless=True) - ncin.alternate_identifier = 'wrong format, missing type' - value = md.get_alternate_identifier( - mmd_yaml['alternate_identifier'], ncin + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_with_altID.nc"), check_only=True ) + ncin = Dataset(md.netcdf_file, "w", diskless=True) + ncin.alternate_identifier = "wrong format, missing type" + value = md.get_alternate_identifier(mmd_yaml["alternate_identifier"], ncin) print(value) self.assertEqual( - md.missing_attributes['errors'][0], - 'alternate_identifier must be formed as ().' + md.missing_attributes["errors"][0], + "alternate_identifier must be formed as ().", ) def test_alternate_identifier(self): @@ -1333,166 +1461,199 @@ def test_alternate_identifier(self): provided in the nc-file. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_with_altID.nc'), check_only=True) - ncin = Dataset(md.netcdf_file) - value = md.get_alternate_identifier( - mmd_yaml['alternate_identifier'], ncin + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_with_altID.nc"), check_only=True ) - self.assertEqual(value[0]['alternate_identifier'], 'dummy_id_no1') - self.assertEqual(value[0]['alternate_identifier_type'], 'dummy_type') + ncin = Dataset(md.netcdf_file) + value = md.get_alternate_identifier(mmd_yaml["alternate_identifier"], ncin) + self.assertEqual(value[0]["alternate_identifier"], "dummy_id_no1") + self.assertEqual(value[0]["alternate_identifier_type"], "dummy_type") def test_alternate_identifier_multiple(self): """Test that MMD alternate_identifier is equal to the ones provided in the nc-file. """ mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_with_altID_multiple.nc'), - check_only=True) - ncin = Dataset(md.netcdf_file) - value = md.get_alternate_identifier( - mmd_yaml['alternate_identifier'], ncin + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_with_altID_multiple.nc"), + check_only=True, ) - self.assertEqual(value[0]['alternate_identifier'], 'dummy_id_no1') - self.assertEqual(value[0]['alternate_identifier_type'], 'dummy_type') - self.assertEqual(value[1]['alternate_identifier_type'], 'other_type') + ncin = Dataset(md.netcdf_file) + value = md.get_alternate_identifier(mmd_yaml["alternate_identifier"], ncin) + self.assertEqual(value[0]["alternate_identifier"], "dummy_id_no1") + self.assertEqual(value[0]["alternate_identifier_type"], "dummy_type") + self.assertEqual(value[1]["alternate_identifier_type"], "other_type") def test_metadata_status_is_active(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_acdd_metadata(mmd_yaml['metadata_status'], ncin, 'metadata_status') - self.assertEqual(value, 'Active') + value = md.get_acdd_metadata( + mmd_yaml["metadata_status"], ncin, "metadata_status" + ) + self.assertEqual(value, "Active") def test_last_metadata_update(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc.nc'), check_only=True) + md = Nc_to_mmd(os.path.abspath("tests/data/reference_nc.nc"), check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_metadata_updates(mmd_yaml['last_metadata_update'], ncin) - self.assertEqual(value['update'][0]['datetime'], '2020-11-27T14:05:56Z') + value = md.get_metadata_updates(mmd_yaml["last_metadata_update"], ncin) + self.assertEqual(value["update"][0]["datetime"], "2020-11-27T14:05:56Z") def test_use_defaults_for_personnel(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader ) md = Nc_to_mmd( - os.path.abspath(os.path.abspath('tests/data/reference_nc_missing_attrs.nc')), - check_only=True) + os.path.abspath( + os.path.abspath("tests/data/reference_nc_missing_attrs.nc") + ), + check_only=True, + ) ncin = Dataset(md.netcdf_file) - value = md.get_personnel(mmd_yaml['personnel'], ncin) - self.assertEqual(value[0]['role'], 'Investigator') - self.assertEqual(value[0]['name'], 'Not available') - self.assertEqual(value[0]['email'], 'Not available') - self.assertEqual(value[0]['organisation'], 'Not available') + value = md.get_personnel(mmd_yaml["personnel"], ncin) + self.assertEqual(value[0]["role"], "Investigator") + self.assertEqual(value[0]["name"], "Not available") + self.assertEqual(value[0]["email"], "Not available") + self.assertEqual(value[0]["organisation"], "Not available") def test_missing_temporal_extent(self): """ToDo: Add docstring""" mmd_yaml = yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_attrs.nc"), check_only=True ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_attrs.nc'), - check_only=True) ncin = Dataset(md.netcdf_file) - value = md.get_temporal_extents(mmd_yaml['temporal_extent'], ncin) + value = md.get_temporal_extents(mmd_yaml["temporal_extent"], ncin) self.assertEqual(value, []) self.assertEqual( - md.missing_attributes['errors'][0], - 'time_coverage_start is a required ACDD attribute' + md.missing_attributes["errors"][0], + "time_coverage_start is a required ACDD attribute", ) def test_missing_temporal_extent_but_start_provided_in_dict(self): """ToDo: Add docstring""" yaml.load( - resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_attrs.nc"), check_only=True ) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_attrs.nc'), - check_only=True) with self.assertRaises(AttributeError): md.to_mmd(overrides={"time_coverage_start": "1850-01-01T00:00:00Z"}) - self.assertEqual(md.metadata['temporal_extent']['start_date'], - '1850-01-01T00:00:00Z') + self.assertEqual( + md.metadata["temporal_extent"]["start_date"], "1850-01-01T00:00:00Z" + ) def test_missing_temporal_extent_but_start_and_end_provided_in_dict(self): """ToDo: Add docstring""" - yaml.load(resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_attrs.nc'), - check_only=True) + yaml.load( + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_attrs.nc"), check_only=True + ) with self.assertRaises(AttributeError): - md.to_mmd(overrides={"time_coverage_start": "1850-01-01T00:00:00Z", - "time_coverage_end": "1950-01-01T00:00:00Z"}) - self.assertEqual(md.metadata['temporal_extent']['start_date'], - '1850-01-01T00:00:00Z') - self.assertEqual(md.metadata['temporal_extent']['end_date'], '1950-01-01T00:00:00Z') + md.to_mmd( + overrides={ + "time_coverage_start": "1850-01-01T00:00:00Z", + "time_coverage_end": "1950-01-01T00:00:00Z", + } + ) + self.assertEqual( + md.metadata["temporal_extent"]["start_date"], "1850-01-01T00:00:00Z" + ) + self.assertEqual( + md.metadata["temporal_extent"]["end_date"], "1950-01-01T00:00:00Z" + ) def test_missing_temporal_extent_but_start_and_end_provided_in_dict_and_wrong(self): """Test that errors are raised when input times are not iso""" - yaml.load(resource_string('py_mmd_tools', 'mmd_elements.yaml'), Loader=yaml.FullLoader) - md = Nc_to_mmd(os.path.abspath('tests/data/reference_nc_missing_attrs.nc'), - check_only=True) + yaml.load( + resource_string("py_mmd_tools", "mmd_elements.yaml"), Loader=yaml.FullLoader + ) + md = Nc_to_mmd( + os.path.abspath("tests/data/reference_nc_missing_attrs.nc"), check_only=True + ) with self.assertRaises(AttributeError): - md.to_mmd(overrides={"time_coverage_start": "1850/01/01 00:00:00", - "time_coverage_end": "1950/01/01 00:00:00"}) + md.to_mmd( + overrides={ + "time_coverage_start": "1850/01/01 00:00:00", + "time_coverage_end": "1950/01/01 00:00:00", + } + ) self.assertEqual( - md.missing_attributes['errors'][5], + md.missing_attributes["errors"][5], "time_coverage_start must be in ISO8601 format: " - "YYYY-mm-ddTHH:MM:SS