diff --git a/geonode/metadata/handlers/abstract.py b/geonode/metadata/handlers/abstract.py
index 74d0dce3f97..292e860fdde 100644
--- a/geonode/metadata/handlers/abstract.py
+++ b/geonode/metadata/handlers/abstract.py
@@ -186,3 +186,18 @@ def _localize_subschema_labels(context, subschema: dict, lang: str, property_nam
@staticmethod
def _check_type(declared, checked):
return declared == checked or (type(declared) is list and checked in declared)
+
+ @staticmethod
+ def _is_nullable(subschema):
+ return MetadataHandler._check_type(subschema.get("type", None), "null")
+
+ @staticmethod
+ def _add_oneof_null_option(subschema):
+ # Without a null option a nullable oneOf rejects the null it gets when left empty.
+ # Never creates the oneOf: that would restrict a field free to take any value
+ oneof = subschema.get("oneOf")
+ if oneof is None or not MetadataHandler._is_nullable(subschema): # an empty one is to be filled in
+ return
+
+ if not any("const" in option and option["const"] is None for option in oneof):
+ oneof.insert(0, {"const": None, "title": "-"}) # not "": the client turns it into None
diff --git a/geonode/metadata/handlers/base.py b/geonode/metadata/handlers/base.py
index 589b9aad828..623566dc6ef 100644
--- a/geonode/metadata/handlers/base.py
+++ b/geonode/metadata/handlers/base.py
@@ -95,12 +95,14 @@ class FrequencySubHandler(SubHandler):
@classmethod
def update_subschema(cls, subschema, lang=None):
subschema["oneOf"] = [{"const": key, "title": val} for key, val in dict(UPDATE_FREQUENCIES).items()]
+ MetadataHandler._add_oneof_null_option(subschema)
class LanguageSubHandler(SubHandler):
@classmethod
def update_subschema(cls, subschema, lang=None):
subschema["oneOf"] = [{"const": key, "title": val} for key, val in dict(ALL_LANGUAGES).items()]
+ MetadataHandler._add_oneof_null_option(subschema)
class LicenseSubHandler(SubHandler):
@@ -132,6 +134,7 @@ def update_subschema(cls, subschema, lang=None):
{"const": tc.identifier, "title": tc.identifier, "description": tc.description}
for tc in RestrictionCodeType.objects.order_by("identifier")
]
+ MetadataHandler._add_oneof_null_option(subschema)
@classmethod
def serialize(cls, db_value):
@@ -151,6 +154,7 @@ def update_subschema(cls, subschema, lang=None):
{"const": tc.identifier, "title": tc.identifier, "description": tc.description}
for tc in SpatialRepresentationType.objects.order_by("identifier")
]
+ MetadataHandler._add_oneof_null_option(subschema)
@classmethod
def serialize(cls, db_value):
diff --git a/geonode/metadata/handlers/sparse.py b/geonode/metadata/handlers/sparse.py
index 48db6c5c1a2..3e1c3ac1bd9 100644
--- a/geonode/metadata/handlers/sparse.py
+++ b/geonode/metadata/handlers/sparse.py
@@ -99,10 +99,11 @@ def _recurse_thesauri_autocomplete(self, d, lang):
self._recurse_thesauri_autocomplete(item, lang)
def _add_thesaurus_options(self, d, thesaurus_id, lang):
- oneof = []
- for keyword in get_localized_tkeywords(lang, thesaurus_id):
- oneof.append({"const": keyword["about"], "title": keyword["label"]})
- d["oneOf"] = oneof
+ d["oneOf"] = [
+ {"const": keyword["about"], "title": keyword["label"] or keyword["default"]}
+ for keyword in get_localized_tkeywords(lang, thesaurus_id)
+ ]
+ self._add_oneof_null_option(d)
def update_schema(self, jsonschema, context, lang=None):
# add all registered fields
@@ -139,10 +140,10 @@ def set_sparse_field(context, fieldname, value):
context[CONTEXT_ID]["fields"][fieldname] = value
def get_jsonschema_instance(self, resource, field_name, context, errors, lang=None):
- field_type = context[CONTEXT_ID]["schema"]["properties"][field_name]["type"]
field_value = context[CONTEXT_ID]["fields"].get(field_name, None)
-
- is_nullable = self._check_type(field_type, "null")
+ subschema = context[CONTEXT_ID]["schema"]["properties"][field_name]
+ is_nullable = self._is_nullable(subschema)
+ field_type = subschema["type"]
if field_name not in context[CONTEXT_ID]["fields"] and not is_nullable:
raise UnsetFieldException()
@@ -192,11 +193,10 @@ def load_deserialization_context(self, resource, jsonschema: dict, context: dict
context[CONTEXT_ID] = {"schema": jsonschema}
def update_resource(self, resource, field_name, json_instance, context, errors, **kwargs):
-
bare_value = json_instance.get(field_name, None)
- field_type = context[CONTEXT_ID]["schema"]["properties"][field_name]["type"]
-
- is_nullable = self._check_type(field_type, "null")
+ subschema = context[CONTEXT_ID]["schema"]["properties"][field_name]
+ field_type = subschema["type"]
+ is_nullable = self._is_nullable(subschema)
if self._check_type(field_type, "string"):
field_value = bare_value
diff --git a/geonode/metadata/handlers/thesaurus.py b/geonode/metadata/handlers/thesaurus.py
index f9ff5bc8b35..e867955a13b 100644
--- a/geonode/metadata/handlers/thesaurus.py
+++ b/geonode/metadata/handlers/thesaurus.py
@@ -126,6 +126,12 @@ def update_schema(self, jsonschema, context, lang=None):
"properties": thesauri,
}
+ # minItems is not enough: it only constrains a thesaurus that is in the instance already.
+ # Same for tkeywords itself, hence the requiredness bubbling up
+ if mandatory_thesauri := [id for id, ct in collected_thesauri.items() if ct["card"]["minItems"] > 0]:
+ subschema["required"] = mandatory_thesauri # the thesauri to be found inside tkeywords
+ subschema["geonode:required"] = True # tkeywords itself, to be found in the resource
+
# We are going to hide the tkeywords property if there's no thesaurus configured
# We can't remove the property altogether, since hkeywords relies on tkeywords for positioning
if not thesauri:
diff --git a/geonode/metadata/management/__init__.py b/geonode/metadata/management/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/geonode/metadata/management/commands/__init__.py b/geonode/metadata/management/commands/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/geonode/metadata/management/commands/validate_metadata.py b/geonode/metadata/management/commands/validate_metadata.py
new file mode 100644
index 00000000000..f368698b891
--- /dev/null
+++ b/geonode/metadata/management/commands/validate_metadata.py
@@ -0,0 +1,535 @@
+#########################################################################
+#
+# Copyright (C) 2026 OSGeo
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+#
+#########################################################################
+
+import csv
+import json
+import logging
+import sys
+
+from django.core.management.base import BaseCommand, CommandError
+from rest_framework.utils.encoders import JSONEncoder
+
+from geonode.base.management import command_utils
+from geonode.base.models import ResourceBase
+from geonode.metadata.manager import metadata_manager
+
+logger = logging.getLogger(__name__)
+
+# Max length of the offending instance value reported along with an error
+MAX_VALUE_REPR = 200
+
+# Max number of problems reported for the schema itself
+MAX_SCHEMA_ERRORS = 20
+
+# Exit codes, combined: a run with both problems exits with 3
+EXIT_INVALID_RESOURCES = 1
+EXIT_INVALID_SCHEMA = 2
+
+
+BASE_FIELDS = ["id", "uuid", "resource_type", "title", "valid", "error_count"]
+ERROR_FIELDS = ["error_path", "error_title", "error_validator", "error_message", "error_value"]
+
+
+class Command(BaseCommand):
+ help = (
+ "Validate the metadata of the GeoNode resources: the jsonschema instance of each resource "
+ "is built and validated against the metadata jsonschema. Empty fields are only reported "
+ "when mandatory, and only at top level: within a nested object every unset field is reported. "
+ "Exits with 1 if any resource is invalid, 2 if the schema itself is, 3 if both"
+ )
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.summary = [] # the lines to be repeated in the report
+ self.schema_errors = 0
+
+ def add_arguments(self, parser):
+ parser.add_argument(
+ "--id",
+ dest="ids",
+ nargs="+",
+ default=None,
+ help="Only validate these resource ids (space and/or comma separated)",
+ )
+
+ parser.add_argument(
+ "--resource-type",
+ dest="resource_types",
+ nargs="+",
+ default=None,
+ help="Only validate these resource types, e.g. dataset map document " "(space and/or comma separated)",
+ )
+
+ parser.add_argument(
+ "--lang",
+ default=None,
+ help="Language used to build the schema and the schema instances (default: schema default)",
+ )
+
+ parser.add_argument(
+ "--include-valid",
+ action="store_true",
+ help="Also report the resources that validate successfully (default: only the invalid ones)",
+ )
+
+ parser.add_argument(
+ "--no-error-details",
+ dest="error_details",
+ action="store_false",
+ default=True,
+ help="Only report the number of errors, without the details of each single error",
+ )
+
+ parser.add_argument(
+ "--max-errors",
+ type=int,
+ default=0,
+ help="Max number of errors reported per resource; 0 (default) means no limit. "
+ "The error count is not affected by this limit",
+ )
+
+ parser.add_argument(
+ "--check-formats",
+ action="store_true",
+ help='Also enforce the jsonschema "format" keyword (needs the jsonschema format extra installed)',
+ )
+
+ parser.add_argument(
+ "--format",
+ dest="outformat",
+ default="json",
+ choices=["json", "csv"],
+ help="Output format (default: json)",
+ )
+
+ parser.add_argument(
+ "--output",
+ default=None,
+ help="Write the report to this file (default: stdout)",
+ )
+
+ parser.add_argument(
+ "--schema",
+ dest="schema_file",
+ default=None,
+ help="Read the jsonschema from this file instead of building it through the metadata manager",
+ )
+
+ parser.add_argument(
+ "--instance",
+ dest="instance_files",
+ nargs="+",
+ default=None,
+ help="Validate the jsonschema instances read from these files, one instance per file, "
+ "instead of the stored resources. Not compatible with --id and --resource-type",
+ )
+
+ parser.add_argument("--debug", dest="debug", action="store_true", help="Set log level to debug")
+
+ def handle(self, *args, **options):
+ log_level = logging.DEBUG if options.get("debug") else logging.INFO
+ command_utils.setup_logger(logger_name=logger.name, level=log_level)
+
+ if options["instance_files"] and (options["ids"] or options["resource_types"]):
+ raise CommandError("--instance selects the instances by itself: --id and --resource-type do not apply")
+
+ validator = self.build_validator(options["lang"], options["check_formats"], options["schema_file"])
+
+ records = []
+ total = 0
+ invalid_count = 0
+
+ for record in self.validate_all(validator, options):
+ total += 1
+ if not record["valid"]:
+ invalid_count += 1
+ if not record["valid"] or options["include_valid"]:
+ records.append(record)
+
+ self.report_schema_defects()
+ self.notify(
+ f"Validated {total} resource(s): {invalid_count} invalid, {total - invalid_count} valid", logging.INFO
+ )
+
+ self.write_report(records, options)
+
+ exit_code = (EXIT_INVALID_RESOURCES if invalid_count else 0) | (
+ EXIT_INVALID_SCHEMA if self.schema_errors else 0
+ )
+ if exit_code:
+ sys.exit(exit_code)
+
+ # -------------------------------------------------------------------
+ # Setup
+ # -------------------------------------------------------------------
+ def build_validator(self, lang, check_formats, schema_file=None):
+ try:
+ from jsonschema import validators as js_validators
+ except ImportError:
+ raise CommandError("The 'jsonschema' package is needed by this command. Please install it")
+
+ # A schema read from a file is plain JSON already, no django objects to render down
+ schema = self.read_json(schema_file) if schema_file else self.as_json_types(metadata_manager.get_schema(lang))
+ validator_class = js_validators.validator_for(schema)
+ logger.debug(f"Using validator {validator_class.__name__}")
+
+ # A broken schema is worth reporting, but the resources may still be validated against it.
+ # check_schema() would raise on the first problem only: we want to see them all at once
+ metaschema_validator = js_validators.validator_for(validator_class.META_SCHEMA, default=validator_class)
+ for count, error in enumerate(metaschema_validator(validator_class.META_SCHEMA).iter_errors(schema), 1):
+ self.schema_errors += 1
+ if count > MAX_SCHEMA_ERRORS:
+ self.notify(f"More than {MAX_SCHEMA_ERRORS} problems in the metadata schema, skipping the rest")
+ break
+ path = "/" + "/".join(str(item) for item in error.absolute_path)
+ self.notify(f"The metadata schema itself is not valid: {path}: {error.message}")
+
+ # A schema reported as invalid above may hold anything in place of a keyword: nothing
+ # derived from it can be taken on trust, starting with the shape of these two
+ self.schema = schema
+ required = schema.get("required", [])
+ self.required_fields = set(required) if isinstance(required, list) else set()
+ properties = schema.get("properties", {})
+
+ # Optional fields not accepting the null they get when left empty: a defect of the schema,
+ # reported once at the end of the run rather than against every resource.
+ # Probed with an actual null: a "oneOf" or an "enum" may reject it as well as "type" does
+ self.broken_optionals = {
+ name
+ for name, subschema in (properties.items() if isinstance(properties, dict) else ())
+ if name not in self.required_fields and self.rejects_null(validator_class, subschema)
+ }
+ self.observed_defects = {}
+
+ format_checker = getattr(validator_class, "FORMAT_CHECKER", None) if check_formats else None
+ return validator_class(schema, format_checker=format_checker)
+
+ def rejects_null(self, validator_class, subschema):
+ try:
+ return not validator_class(subschema).is_valid(None)
+ except Exception: # a malformed subschema: the metaschema check has reported it already
+ return False
+
+ def read_json(self, path):
+ try:
+ with open(path, encoding="utf-8") as infile:
+ return json.load(infile)
+ except (OSError, ValueError) as e:
+ raise CommandError(f"Can not read the JSON file '{path}': {e}")
+
+ def as_json_types(self, data):
+ """
+ Render the data down to plain JSON types, as the API does before sending it to the client:
+ a lazy translation is not a str, and jsonschema would reject it as "not of type 'string'"
+ """
+ return json.loads(json.dumps(data, cls=JSONEncoder))
+
+ def select_resources(self, ids, resource_types):
+ queryset = ResourceBase.objects.all()
+
+ if ids:
+ requested_ids = self.parse_list(ids, as_int=True)
+ queryset = queryset.filter(pk__in=requested_ids)
+ found_ids = set(queryset.values_list("pk", flat=True))
+ if missing := set(requested_ids) - found_ids:
+ logger.warning(f"Requested resources not found: {sorted(missing)}")
+
+ if resource_types:
+ queryset = queryset.filter(resource_type__in=self.parse_list(resource_types))
+
+ return queryset.order_by("pk")
+
+ def parse_list(self, values, as_int=False):
+ parsed = []
+ for value in values:
+ for item in str(value).split(","):
+ item = item.strip()
+ if not item:
+ continue
+ if as_int:
+ try:
+ item = int(item)
+ except ValueError:
+ raise CommandError(f"Not a valid resource id: '{item}'")
+ parsed.append(item)
+ return parsed
+
+ # -------------------------------------------------------------------
+ # Validation
+ # -------------------------------------------------------------------
+ def validate_all(self, validator, options):
+ """Yield a record per entry, reading them from the given files or from the stored resources"""
+ if options["instance_files"]:
+ logger.info(f"Validating {len(options['instance_files'])} instance file(s)")
+ for path in options["instance_files"]:
+ yield self.validate_file(path, validator, options["max_errors"])
+ return
+
+ queryset = self.select_resources(options["ids"], options["resource_types"])
+ logger.info(f"Validating {queryset.count()} resource(s)")
+
+ # Non polymorphic, exactly like the API read path: we validate what the client is served.
+ # iterator() keeps the resources out of memory; the records being reported are not
+ for resource in queryset.iterator():
+ yield self.validate_resource(resource, validator, options["lang"], options["max_errors"])
+
+ def validate_file(self, path, validator, max_errors):
+ instance = self.read_json(path)
+ record = {
+ "id": path,
+ "uuid": instance.get("uuid"),
+ "resource_type": None, # not part of the schema
+ "title": instance.get("title"),
+ }
+ return self.validate_instance(record, instance, validator, max_errors)
+
+ def validate_resource(self, resource, validator, lang, max_errors):
+ record = {
+ "id": resource.pk,
+ "uuid": resource.uuid,
+ "resource_type": resource.resource_type,
+ "title": resource.title,
+ }
+ logger.debug(f"Validating resource {resource.pk}: '{resource.title}'")
+
+ try:
+ instance = metadata_manager.build_schema_instance(resource, lang)
+ instance = self.as_json_types(instance)
+ except Exception as e:
+ logger.error(f"Can not build the schema instance for resource {resource.pk}", exc_info=e)
+ return dict(
+ record,
+ valid=False,
+ error_count=1,
+ errors=[
+ {
+ "path": "/",
+ "title": "",
+ "validator": "geonode:build_schema_instance",
+ "message": f"Can not build the schema instance: {e}",
+ "value": None,
+ }
+ ],
+ )
+
+ return self.validate_instance(record, instance, validator, max_errors)
+
+ def validate_instance(self, record, instance, validator, max_errors):
+ # GeoNode injects this key in the instance for testing purposes only (resources titled "*error*")
+ instance.pop("extraErrors", None)
+
+ try:
+ errors = self.drop_redundant(validator.iter_errors(instance))
+ except Exception as e:
+ # A keyword the metaschema has already refused can blow up on a value of the right type
+ logger.error(f"Can not validate {record['id']} against the schema", exc_info=e)
+ return dict(
+ record,
+ valid=False,
+ error_count=1,
+ errors=[
+ {
+ "path": "/",
+ "title": "",
+ "validator": "geonode:invalid_schema",
+ "message": f"Can not validate against the schema: {e}",
+ "value": None,
+ }
+ ],
+ )
+
+ errors = [error for error in errors if not self.is_schema_defect(error)]
+ errors = sorted(errors, key=lambda err: ([str(p) for p in err.absolute_path], str(err.validator)))
+ reported = errors[:max_errors] if max_errors > 0 else errors
+
+ return dict(
+ record,
+ valid=not errors,
+ error_count=len(errors),
+ errors=[self.format_error(err) for err in reported],
+ )
+
+ def is_unset(self, error):
+ # Keyed on the value: depending on the subschema a null may trip "type", "oneOf", or both
+ return error.instance is None
+
+ def is_schema_defect(self, error):
+ # An optional field left empty: blame the schema, not the resource
+ if not (self.is_unset(error) and len(error.absolute_path) == 1):
+ return False
+
+ field = error.absolute_path[0]
+ if field not in self.broken_optionals:
+ return False
+
+ self.observed_defects[field] = self.observed_defects.get(field, 0) + 1
+ return True
+
+ def notify(self, message, level=logging.WARNING):
+ # Logged, and repeated in the report: the report file is meant to travel on its own
+ logger.log(level, message)
+ self.summary.append(message)
+
+ def report_schema_defects(self):
+ # Only what some resource really ran into: a null never handed over is nobody's problem
+ for field, count in sorted(self.observed_defects.items()):
+ self.notify(
+ f"The metadata schema does not allow a null value on the optional field "
+ f"'{field}', left empty by {count} resource(s)"
+ )
+
+ def drop_redundant(self, errors):
+ """
+ A null fails every keyword declared for the field, all reporting the same problem: keep
+ one error per unset field, preferring "type" since it names the expected types
+ """
+ errors = list(errors)
+ unset_paths = {tuple(error.absolute_path) for error in errors if self.is_unset(error)}
+
+ kept = [error for error in errors if tuple(error.absolute_path) not in unset_paths]
+
+ for path in unset_paths:
+ at_path = [error for error in errors if tuple(error.absolute_path) == path]
+ kept.append(next((e for e in at_path if e.validator == "type"), at_path[0]))
+
+ return kept
+
+ def path_title(self, path):
+ # The property names are ids: walk the schema alongside the path to pick up their titles
+ titles = []
+ subschema = self.schema
+ for item in path:
+ # A malformed schema may hold anything in place of a subschema: fall back to the names
+ if isinstance(item, int):
+ subschema = self.as_dict(subschema).get("items")
+ titles.append(f"[{item}]")
+ else:
+ subschema = self.as_dict(self.as_dict(subschema).get("properties")).get(item)
+ titles.append(self.as_dict(subschema).get("title", item))
+ return "/".join(titles)
+
+ @staticmethod
+ def as_dict(value):
+ return value if isinstance(value, dict) else {}
+
+ def missing_property(self, error):
+ # jsonschema names the missing property in the message only, and does not add it to the
+ # path: match the message back to the required list, so a reworded one just yields nothing
+ if error.validator != "required":
+ return None
+ return next((p for p in error.validator_value if f"{p!r} is a required property" == error.message), None)
+
+ def format_error(self, error):
+ path = tuple(error.absolute_path)
+ message = error.message
+
+ # A missing property is reported against the object holding it: point at the property
+ missing = self.missing_property(error)
+ if missing is not None:
+ path += (missing,)
+
+ # The instance behind a container error is the container itself: useless noise
+ value = None
+ if path and missing is None:
+ value = json.dumps(error.instance, ensure_ascii=False, default=str)
+ if len(value) > MAX_VALUE_REPR:
+ value = f"{value[:MAX_VALUE_REPR]}…"
+
+ # A top level field only gets reported when mandatory, no point in repeating it. Deeper
+ # down nothing is filtered, so there it is worth telling a mandatory field from the rest
+ if missing is not None:
+ message = "not set" if len(path) == 1 else "not set (required field)"
+
+ # jsonschema puts it in python terms ("None is not of type 'object'"), reading as a type
+ # mismatch where the field is simply empty. Spell it out, and drop the pointless null
+ elif self.is_unset(error):
+ if len(path) > 1 and error.validator == "type":
+ expected = error.validator_value
+ expected = expected if isinstance(expected, list) else [expected]
+ message = f"not set (expected {' or '.join(expected)})"
+ else:
+ message = "not set"
+ value = None
+
+ # jsonschema quotes the whole offending string in the message ("'xxxxx…' is too long"),
+ # which is exactly the value we are careful to truncate in the value column
+ elif error.validator == "maxLength":
+ message = f"too long: {len(error.instance)} chars (max {error.validator_value})"
+
+ return {
+ "path": "/" + "/".join(str(item) for item in path),
+ "title": self.path_title(path),
+ "validator": str(error.validator),
+ "message": message,
+ "value": value,
+ }
+
+ # -------------------------------------------------------------------
+ # Reporting
+ # -------------------------------------------------------------------
+ def write_report(self, records, options):
+ outpath = options["output"]
+
+ if outpath:
+ try:
+ with open(outpath, "w", newline="", encoding="utf-8") as out:
+ self.dump(records, out, options)
+ except OSError as e:
+ raise CommandError(f"Can not write the report to '{outpath}': {e}")
+ logger.info(f"Report written to {outpath}")
+ else:
+ # OutputWrapper appends a newline to every write() call, which would break the report
+ self.stdout.ending = ""
+ self.dump(records, self.stdout, options)
+
+ def dump(self, records, out, options):
+ if not options["error_details"]:
+ records = [{k: v for k, v in record.items() if k != "errors"} for record in records]
+
+ if options["outformat"] == "json":
+ json.dump({"summary": self.summary, "resources": records}, out, indent=2, ensure_ascii=False, default=str)
+ out.write("\n")
+ else:
+ # CSV has no comment syntax: "#" is only a widespread convention
+ for message in self.summary:
+ out.write(f"# {message}\n")
+ self.dump_csv(records, out, options["error_details"])
+
+ def dump_csv(self, records, out, error_details):
+ fields = BASE_FIELDS + (ERROR_FIELDS if error_details else [])
+ writer = csv.DictWriter(out, fieldnames=fields, extrasaction="ignore")
+ writer.writeheader()
+
+ for record in records:
+ if not error_details or not record["errors"]:
+ writer.writerow(record)
+ continue
+
+ # one row per error, the resource columns are repeated
+ for error in record["errors"]:
+ writer.writerow(
+ dict(
+ record,
+ error_path=error["path"],
+ error_title=error["title"],
+ error_validator=error["validator"],
+ error_message=error["message"],
+ error_value=error["value"],
+ )
+ )
diff --git a/geonode/metadata/schemas/base.json b/geonode/metadata/schemas/base.json
index f022662ecfb..4d1f0397fda 100644
--- a/geonode/metadata/schemas/base.json
+++ b/geonode/metadata/schemas/base.json
@@ -93,7 +93,7 @@
}
},
"restriction_code_type": {
- "type": "string",
+ "type": ["string", "null"],
"title": "restrictions",
"description": "limitation(s) placed upon the access or use of the data.",
"maxLength": 255
@@ -148,13 +148,13 @@
"description": "time period covered by the content of the dataset (end)"
},
"maintenance_frequency": {
- "type": "string",
+ "type": ["string", "null"],
"title": "maintenance frequency",
"description": "frequency with which modifications and deletions are made to the data after it is first produced",
"maxLength": 255
},
"spatial_representation_type": {
- "type": "string",
+ "type": ["string", "null"],
"title": "spatial representation type",
"description": "method used to represent geographic information in the dataset.",
"maxLength": 255
diff --git a/geonode/metadata/tests/test_handlers.py b/geonode/metadata/tests/test_handlers.py
index 7ce59cd8289..f6ffd0390c8 100644
--- a/geonode/metadata/tests/test_handlers.py
+++ b/geonode/metadata/tests/test_handlers.py
@@ -25,7 +25,7 @@
from datetime import datetime
from django.contrib.auth import get_user_model
-from django.test import RequestFactory
+from django.test import RequestFactory, SimpleTestCase
from django.utils.translation import gettext as _
from geonode.people import Roles
@@ -55,6 +55,7 @@
RestrictionsSubHandler,
SpatialRepresentationTypeSubHandler,
)
+from geonode.metadata.handlers.abstract import MetadataHandler
from geonode.metadata.handlers.region import RegionHandler
from geonode.metadata.handlers.doi import DOIHandler
from geonode.metadata.handlers.linkedresource import LinkedResourceHandler
@@ -537,6 +538,27 @@ def test_restrictions_subhandler_update_subschema(self):
self.assertEqual(subschema["oneOf"][0]["title"], "fake_restrictions")
self.assertEqual(subschema["oneOf"][0]["description"], "a detailed description")
+ def test_restrictions_subhandler_update_subschema_nullable(self):
+ """
+ A nullable field needs an explicit empty choice in its oneOf, or the null it is given
+ when left unset matches no branch at all
+ """
+
+ subschema = {
+ "type": ["string", "null"],
+ "title": "restrictions",
+ "maxLength": 255,
+ }
+
+ fake_restrictions = RestrictionCodeType.objects.get(identifier="fake_restrictions")
+ RestrictionCodeType.objects.exclude(identifier=fake_restrictions.identifier).delete()
+
+ RestrictionsSubHandler.update_subschema(subschema, lang="en")
+
+ self.assertEqual(2, len(subschema["oneOf"]))
+ self.assertEqual({"const": None, "title": "-"}, subschema["oneOf"][0]) # first: it is the empty one
+ self.assertEqual("fake_restrictions", subschema["oneOf"][1]["const"])
+
def test_restrictions_subhandler_serialize_with_existed_db_value(self):
"""
Test the serialize method with existed db value.
@@ -1385,6 +1407,10 @@ def test_tkeywords_handler_update_schema_with_thesauri(self, mock_collect_thesau
self.assertEqual(tkeywords["type"], "object")
self.assertEqual(tkeywords["title"], "Keywords from Thesaurus")
+ # Only the thesaurus with card_min > 0 is required, and it makes tkeywords required in turn
+ self.assertEqual(tkeywords["required"], ["3-2-4-1-gemet-inspire-themes"])
+ self.assertTrue(tkeywords["geonode:required"])
+
# Assert thesaurus structure for "3-2-4-3-spatialscope"
thesaurus = tkeywords["properties"]["3-2-4-3-spatialscope"]
self.assertEqual(thesaurus["type"], "array")
@@ -1436,6 +1462,10 @@ def test_tkeywords_handler_update_schema_no_thesauri(self, mock_collect_thesauri
self.assertIsNotNone(tkeywords)
self.assertEqual(tkeywords["ui:widget"], "hidden")
+ # With no thesaurus at all there is nothing to require
+ self.assertNotIn("required", tkeywords)
+ self.assertNotIn("geonode:required", tkeywords)
+
def test_tkeywords_handler_get_jsonschema_instance_translated_keywords(self):
"""
Test for the get_jsonschema_instance of the tkeywords handler.
@@ -1731,3 +1761,55 @@ def test_sparse_handler_update_resource(self):
self.resource, "integer_field", json_instance_invalid_int_number, self.context, self.errors
)
self.assertIn("metadata_sparse_error_parse", str(self.errors))
+
+
+class NullOptionTests(SimpleTestCase):
+ """
+ MetadataHandler._add_oneof_null_option works on plain dicts: no need for the db fixtures
+ """
+
+ NULL_OPTION = {"const": None, "title": "-"}
+
+ def test_null_option_comes_first(self):
+ subschema = {"type": ["string", "null"], "oneOf": [{"const": "a"}]}
+
+ MetadataHandler._add_oneof_null_option(subschema)
+
+ self.assertEqual([self.NULL_OPTION, {"const": "a"}], subschema["oneOf"])
+
+ def test_an_existing_null_option_is_not_duplicated(self):
+ subschema = {"type": ["string", "null"], "oneOf": [self.NULL_OPTION, {"const": "a"}]}
+
+ MetadataHandler._add_oneof_null_option(subschema)
+
+ self.assertEqual([self.NULL_OPTION, {"const": "a"}], subschema["oneOf"])
+
+ def test_an_option_without_const_is_not_taken_for_the_null_one(self):
+ subschema = {"type": ["string", "null"], "oneOf": [{"title": "no const here"}]}
+
+ MetadataHandler._add_oneof_null_option(subschema)
+
+ self.assertEqual([self.NULL_OPTION, {"title": "no const here"}], subschema["oneOf"])
+
+ def test_a_non_nullable_field_is_left_alone(self):
+ subschema = {"type": "string", "oneOf": [{"const": "a"}]}
+
+ MetadataHandler._add_oneof_null_option(subschema)
+
+ self.assertEqual([{"const": "a"}], subschema["oneOf"])
+
+ def test_the_oneof_is_never_created(self):
+ # creating it would restrict a field which was free to take any value
+ subschema = {"type": ["string", "null"]}
+
+ MetadataHandler._add_oneof_null_option(subschema)
+
+ self.assertNotIn("oneOf", subschema)
+
+ def test_an_empty_oneof_is_filled_in(self):
+ # an empty oneOf matches no instance at all, null included
+ subschema = {"type": ["string", "null"], "oneOf": []}
+
+ MetadataHandler._add_oneof_null_option(subschema)
+
+ self.assertEqual([self.NULL_OPTION], subschema["oneOf"])
diff --git a/geonode/metadata/tests/test_validate_metadata.py b/geonode/metadata/tests/test_validate_metadata.py
new file mode 100644
index 00000000000..615a4c2ba31
--- /dev/null
+++ b/geonode/metadata/tests/test_validate_metadata.py
@@ -0,0 +1,288 @@
+#########################################################################
+#
+# Copyright (C) 2026 OSGeo
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+#
+#########################################################################
+import csv
+import json
+import os
+import tempfile
+from io import StringIO
+from unittest.mock import patch
+from uuid import uuid4
+
+from django.contrib.auth import get_user_model
+from django.core import management
+from django.core.management.base import CommandError
+from django.test import SimpleTestCase, TestCase
+
+from geonode.base.models import ResourceBase
+from geonode.metadata.management.commands import validate_metadata
+
+COMMAND = "validate_metadata"
+
+# A schema small enough to make every assertion below exact
+SCHEMA = {
+ "$schema": "https://json-schema.org/draft/2020-12/schema",
+ "type": "object",
+ "required": ["title", "abstract"],
+ "properties": {
+ "uuid": {"type": "string", "title": "UUID"},
+ "title": {"type": "string", "title": "Title", "maxLength": 10},
+ "abstract": {"type": ["string", "null"], "title": "Abstract"},
+ # optional and not nullable: the null it gets when empty is a defect of the schema
+ "edition": {"type": "string", "title": "Edition"},
+ },
+}
+
+
+class ValidateMetadataCommandMixin:
+ def call(self, *args, **options):
+ """Run the command, returning its output and its exit code"""
+ out = StringIO()
+ code = 0
+ try:
+ management.call_command(COMMAND, *args, stdout=out, **options)
+ except SystemExit as e:
+ code = e.code
+ return out.getvalue(), code
+
+ def write_json(self, name, content):
+ path = os.path.join(self.tmpdir.name, name)
+ with open(path, "w") as outfile:
+ json.dump(content, outfile)
+ return path
+
+
+class ValidateMetadataOfflineTests(ValidateMetadataCommandMixin, SimpleTestCase):
+ """
+ The reporting contract, exercised through --schema and --instance: no db, no metadata manager
+ """
+
+ def setUp(self):
+ self.tmpdir = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmpdir.cleanup)
+ self.schema = self.write_json("schema.json", SCHEMA)
+
+ self.valid = self.write_json("valid.json", {"uuid": "u-1", "title": "fine", "abstract": "a"})
+ self.invalid = self.write_json("invalid.json", {"uuid": "u-2", "title": 42, "abstract": "a"})
+
+ def report(self, *instances, **options):
+ out, code = self.call(schema_file=self.schema, instance_files=list(instances), **options)
+ return json.loads(out), code
+
+ def test_valid_instance_is_not_reported_and_exits_zero(self):
+ report, code = self.report(self.valid)
+
+ self.assertEqual([], report["resources"])
+ self.assertEqual(0, code)
+
+ def test_valid_instance_is_reported_on_demand(self):
+ report, _ = self.report(self.valid, include_valid=True)
+
+ self.assertEqual(1, len(report["resources"]))
+ self.assertTrue(report["resources"][0]["valid"])
+ self.assertEqual(self.valid, report["resources"][0]["id"])
+ self.assertEqual("u-1", report["resources"][0]["uuid"])
+
+ def test_invalid_instance_exits_one(self):
+ report, code = self.report(self.invalid)
+
+ self.assertEqual(1, code)
+ self.assertEqual(1, len(report["resources"]))
+
+ error = report["resources"][0]["errors"][0]
+ self.assertEqual("/title", error["path"])
+ self.assertEqual("Title", error["title"]) # the schema title, not the property name
+ self.assertEqual("type", error["validator"])
+
+ def test_missing_mandatory_field_is_reported_against_the_field(self):
+ # jsonschema reports "required" against the object: the field itself is the useful subject
+ path = self.write_json("no_title.json", {"uuid": "u-3", "abstract": "a"})
+ report, _ = self.report(path)
+
+ errors = report["resources"][0]["errors"]
+ self.assertEqual(["/title"], [e["path"] for e in errors])
+ self.assertEqual(["not set"], [e["message"] for e in errors])
+
+ def test_empty_optional_field_is_a_schema_defect_not_a_resource_error(self):
+ path = self.write_json("no_edition.json", {"uuid": "u-4", "title": "fine", "abstract": "a", "edition": None})
+ report, code = self.report(path, include_valid=True)
+
+ # the resource is not to blame: the schema is the one not allowing the null
+ self.assertTrue(report["resources"][0]["valid"])
+ self.assertEqual(0, code)
+ self.assertIn(
+ "The metadata schema does not allow a null value on the optional field 'edition', "
+ "left empty by 1 resource(s)",
+ report["summary"],
+ )
+
+ def test_schema_defects_are_not_reported_when_no_instance_runs_into_them(self):
+ report, _ = self.report(self.valid, include_valid=True)
+
+ self.assertFalse([message for message in report["summary"] if "optional field" in message])
+
+ def test_max_errors_caps_the_details_but_not_the_count(self):
+ path = self.write_json("three.json", {"uuid": 1, "title": "way too long to fit", "edition": 2})
+ full, _ = self.report(path)
+ capped, _ = self.report(path, max_errors=1)
+
+ self.assertEqual(4, full["resources"][0]["error_count"]) # uuid, title, edition, missing abstract
+ self.assertEqual(4, len(full["resources"][0]["errors"]))
+
+ self.assertEqual(4, capped["resources"][0]["error_count"])
+ self.assertEqual(1, len(capped["resources"][0]["errors"]))
+
+ def test_long_value_is_not_quoted_back_in_the_message(self):
+ path = self.write_json("long.json", {"uuid": "u-5", "title": "x" * 300, "abstract": "a"})
+ report, _ = self.report(path)
+
+ error = report["resources"][0]["errors"][0]
+ self.assertEqual("too long: 300 chars (max 10)", error["message"])
+
+ def test_json_report_carries_the_summary(self):
+ report, _ = self.report(self.invalid)
+
+ self.assertIn("Validated 1 resource(s): 1 invalid, 0 valid", report["summary"])
+ self.assertEqual(["summary", "resources"], list(report))
+
+ def test_csv_report_has_one_row_per_error_and_comments_the_summary(self):
+ out, _ = self.call(schema_file=self.schema, instance_files=[self.valid, self.invalid], outformat="csv")
+
+ comments = [line for line in out.splitlines() if line.startswith("#")]
+ rows = list(csv.DictReader(line for line in out.splitlines() if not line.startswith("#")))
+
+ self.assertIn("# Validated 2 resource(s): 1 invalid, 1 valid", comments)
+ self.assertEqual(1, len(rows)) # only the invalid one
+ self.assertEqual("/title", rows[0]["error_path"])
+ self.assertEqual("Title", rows[0]["error_title"])
+ self.assertEqual("u-2", rows[0]["uuid"])
+
+ def test_csv_report_without_details_has_one_row_per_instance(self):
+ out, _ = self.call(
+ schema_file=self.schema,
+ instance_files=[self.valid, self.invalid],
+ outformat="csv",
+ include_valid=True,
+ error_details=False,
+ )
+
+ rows = list(csv.DictReader(line for line in out.splitlines() if not line.startswith("#")))
+ self.assertEqual(2, len(rows))
+ self.assertNotIn("error_path", rows[0])
+
+ def test_report_can_be_written_to_a_file(self):
+ outpath = os.path.join(self.tmpdir.name, "report.json")
+ self.call(schema_file=self.schema, instance_files=[self.valid], output=outpath, include_valid=True)
+
+ with open(outpath) as infile:
+ self.assertEqual(1, len(json.load(infile)["resources"]))
+
+ def test_broken_schema_is_reported_and_flagged_in_the_exit_code(self):
+ broken = dict(SCHEMA, properties=dict(SCHEMA["properties"], edition={"maxLength": "not a number"}))
+ schema = self.write_json("broken.json", broken)
+
+ out, code = self.call(schema_file=schema, instance_files=[self.valid], include_valid=True)
+
+ self.assertEqual(2, code) # the instance is valid: only the schema flag is raised
+ self.assertTrue([m for m in json.loads(out)["summary"] if "schema itself is not valid" in m])
+
+ def test_broken_schema_and_invalid_instance_combine_in_the_exit_code(self):
+ broken = dict(SCHEMA, properties=dict(SCHEMA["properties"], edition={"maxLength": "not a number"}))
+ schema = self.write_json("broken.json", broken)
+
+ _, code = self.call(schema_file=schema, instance_files=[self.invalid])
+
+ self.assertEqual(3, code)
+
+ def test_unreadable_file_is_refused(self):
+ with self.assertRaises(CommandError):
+ self.call(schema_file=self.schema, instance_files=[os.path.join(self.tmpdir.name, "nope.json")])
+
+ def test_instances_and_resource_filters_are_mutually_exclusive(self):
+ with self.assertRaises(CommandError):
+ self.call(schema_file=self.schema, instance_files=[self.valid], ids=["1"])
+
+ with self.assertRaises(CommandError):
+ self.call(schema_file=self.schema, instance_files=[self.valid], resource_types=["dataset"])
+
+
+class ValidateMetadataResourceTests(ValidateMetadataCommandMixin, TestCase):
+ """
+ Selection of the stored resources. The schema comes from a file and the instances from a mock,
+ so that the assertions are about which resources are picked, not about the metadata of a fixture
+ """
+
+ def setUp(self):
+ self.tmpdir = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmpdir.cleanup)
+ self.schema = self.write_json("schema.json", SCHEMA)
+
+ owner = get_user_model().objects.create_user("validate_user", "v@fakemail.com", "pwd")
+ self.dataset = ResourceBase.objects.create(
+ title="A dataset", uuid=str(uuid4()), resource_type="dataset", owner=owner
+ )
+ self.document = ResourceBase.objects.create(
+ title="A document", uuid=str(uuid4()), resource_type="document", owner=owner
+ )
+
+ patcher = patch("geonode.metadata.management.commands.validate_metadata.metadata_manager")
+ self.manager = patcher.start()
+ self.addCleanup(patcher.stop)
+ # every resource yields the same valid instance: what is under test here is the selection
+ self.manager.build_schema_instance.side_effect = lambda resource, lang: {
+ "uuid": resource.uuid,
+ "title": "fine",
+ "abstract": "an abstract",
+ }
+
+ def reported_ids(self, **options):
+ out, _ = self.call(schema_file=self.schema, include_valid=True, **options)
+ return [record["id"] for record in json.loads(out)["resources"]]
+
+ def test_all_the_resources_are_validated_by_default(self):
+ self.assertEqual([self.dataset.pk, self.document.pk], self.reported_ids())
+
+ def test_filter_by_id(self):
+ self.assertEqual([self.document.pk], self.reported_ids(ids=[str(self.document.pk)]))
+
+ def test_filter_by_resource_type(self):
+ self.assertEqual([self.dataset.pk], self.reported_ids(resource_types=["dataset"]))
+
+ def test_filters_are_combined(self):
+ self.assertEqual([], self.reported_ids(ids=[str(self.dataset.pk)], resource_types=["document"]))
+
+ def test_unknown_id_is_warned_about_and_skipped(self):
+ # not assertLogs: setup_logger() drops every handler on this logger, capture included
+ with patch.object(validate_metadata.logger, "warning") as warning:
+ self.assertEqual([self.dataset.pk], self.reported_ids(ids=[f"{self.dataset.pk},999999"]))
+
+ self.assertIn("Requested resources not found: [999999]", [call.args[0] for call in warning.call_args_list])
+
+ def test_not_an_id_is_refused(self):
+ with self.assertRaises(CommandError):
+ self.call(schema_file=self.schema, ids=["not-a-number"])
+
+ def test_a_failing_handler_is_reported_against_the_resource(self):
+ self.manager.build_schema_instance.side_effect = RuntimeError("handler exploded")
+
+ out, code = self.call(schema_file=self.schema, ids=[str(self.dataset.pk)])
+ record = json.loads(out)["resources"][0]
+
+ self.assertEqual(1, code)
+ self.assertEqual("geonode:build_schema_instance", record["errors"][0]["validator"])
+ self.assertIn("handler exploded", record["errors"][0]["message"])