Write validator plugins to register.json as validatorPlugins

Author: avillar

ogc/bblocks/postprocess.py

@@ -81,7 +81,7 @@ def postprocess(registered_items_path: str | Path = 'registereditems',
         )
 
     transform_plugins = load_transform_plugins(sandbox_dir, allowed_modules=allowed_plugin_modules)
-    plugin_validators = load_validation_plugins(sandbox_dir, allowed_modules=allowed_validator_modules)
+    plugin_validators, validator_plugins = load_validation_plugins(sandbox_dir, allowed_modules=allowed_validator_modules)
 
     if not isinstance(test_outputs_path, Path):
         test_outputs_path = Path(test_outputs_path)
@@ -585,6 +585,9 @@ def do_postprocess(bblock: BuildingBlock, light: bool = False) -> bool:
         if transform_plugins:
             output_register_json['transformPlugins'] = transform_plugins
 
+        if validator_plugins:
+            output_register_json['validatorPlugins'] = validator_plugins
+
         remote_cache_dir_url = f"{base_url}{os.path.relpath(Path(remote_cache_dir).resolve(), cwd)}"
         output_register_json['remoteCacheDir'] = remote_cache_dir_url
 

ogc/bblocks/validate.py

@@ -37,19 +37,24 @@
 
 
 def load_validation_plugins(sandbox_dir: Path,
-                            allowed_modules: set[str] | None = None) -> list[PluginValidator]:
+                            allowed_modules: set[str] | None = None,
+                            ) -> tuple[list[PluginValidator], list[dict]]:
     """Read validator plugin config, create per-plugin venvs, and return PluginValidator instances.
 
     Reads from ``plugins.validators`` in bblocks-config.yaml.
     allowed_modules: if provided, only install/register modules in this set. Pass None to allow all.
+
+    Returns a tuple of (plugin_validators, register_entries) where register_entries is the enriched
+    plugin list suitable for inclusion in register.json under 'validatorPlugins'.
     """
-    from ogc.bblocks.transform import read_plugin_entries
+    from ogc.bblocks.transform import read_plugin_entries, _pip_to_url
 
     plugin_entries = read_plugin_entries('validators')
     if not plugin_entries:
-        return []
+        return [], []
 
     result: list[PluginValidator] = []
+    output_plugins: list[dict] = []
 
     for plugin in plugin_entries:
         pip_deps = plugin.get('pip', [])
@@ -60,6 +65,8 @@ def load_validation_plugins(sandbox_dir: Path,
         if isinstance(modules, str):
             modules = [modules]
 
+        output_modules = []
+
         for module_path in modules:
             if allowed_modules is not None and module_path not in allowed_modules:
                 logger.info("Skipping validator plugin '%s': not permitted by user", module_path)
@@ -82,6 +89,7 @@ def load_validation_plugins(sandbox_dir: Path,
                 logger.warning("No validator classes found in plugin '%s'", module_path)
                 continue
 
+            output_validators = []
             for entry in discovered:
                 mime_types = entry.get('mime_types', [])
                 file_extensions = entry.get('file_extensions', [])
@@ -98,8 +106,29 @@ def load_validation_plugins(sandbox_dir: Path,
                 logger.info("Registered validator plugin '%s' (%s) for mime_types=%s extensions=%s",
                             module_path, entry['class'], mime_types, file_extensions)
                 result.append(pv)
-
-    return result
+                output_validators.append({
+                    'class': entry['class'],
+                    'mimeTypes': mime_types,
+                    'fileExtensions': file_extensions,
+                })
+
+            if output_validators:
+                output_modules.append({'module': module_path, 'validators': output_validators})
+
+        if output_modules:
+            output_entry: dict = {'modules': output_modules}
+            original_pip = plugin.get('pip')
+            if original_pip:
+                output_entry['pip'] = original_pip
+                if explicit_url := plugin.get('url'):
+                    output_entry['urls'] = [explicit_url]
+                else:
+                    urls = [u for s in pip_deps for u in [_pip_to_url(s)] if u]
+                    if urls:
+                        output_entry['urls'] = urls
+            output_plugins.append(output_entry)
+
+    return result, output_plugins
 
 
 def report_to_dict(bblock: BuildingBlock,