maple/prplos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ci: scripts: add YAML manifest loader and package_apply subcommand

Browse Source 
Currently testbed-cdrouter.py only supports importing packages from
opaque .gz binary exports via the package_import subcommand. This
requires staging, committing, and name-patching through the CDRouter
import API, and provides no way to define packages as readable,
version-controlled YAML.

So lets add a resolve_manifest() method that loads YAML package
manifests with full include resolution and a package_apply
subcommand that upserts packages on the CDRouter instance via the
packages API.

The manifest resolver has strict validation:
  - Key allowlist (name, description, includes, tests, options,
    dedup_tests); unknown keys are rejected
  - Forbidden keys (config, device, config_id, device_id)
  - Root manifest name must match its filename
  - Group manifests (under groups/) must not define name,
    description, or options
  - Type checks on all fields

Include resolution uses depth-first traversal with cycle
detection, path traversal rejection, and root-wins option merge.
Test deduplication is configurable via the root-only dedup_tests
boolean (defaults to false, preserving duplicates). Non-root
manifests (both includes and groups) must not define dedup_tests.

The package_apply subcommand performs deterministic upserts:
it always writes the complete manifest-driven state (testlist,
description, options) and only creates a new package on genuine
not-found errors from the CDRouter API.

Runtime config/device binding remains in the existing package_run
flow, keeping package manifests environment-agnostic.

References: PCF-2532
Assisted-by: codex:gpt-5.4:xhigh
Assisted-by: claude:opus-4-6:1m-context
Signed-off-by: Petr Štetiar <petr.stetiar@prplfoundation.org>
This commit is contained in:
Petr Štetiar 2026-02-19 18:45:47 +01:00 committed by Hela Saadani
parent 6bc69fde54
commit fe9c778e77
1 changed files with 278 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

279
.gitlab/scripts/testbed-cdrouter.py
View File

@ -7,6 +7,7 @@ import os
import sys
import time
import logging
import yaml
import humanize
from dataclasses import dataclass
@ -257,7 +258,9 @@ class TestbedCDRouter:
if timeout and (time.time() - start_time) >= timeout:
elapsed = time.time() - start_time
logging.error(
"Test execution timeout exceeded after {}.".format(humanize.naturaldelta(elapsed))
"Test execution timeout exceeded after {}.".format(
humanize.naturaldelta(elapsed)
)
)
self.package_stop()
return False
@ -513,6 +516,272 @@ class TestbedCDRouter:
self.cdr.configs.create(config)
logging.info("Imported config '{}' from '{}'".format(name, self.args.filename))
def resolve_manifest(
self, manifest_path, visited=None, root_manifest_path=None, dedup_tests=None
):
manifest_path = os.path.realpath(manifest_path)
packages_dir = os.path.realpath(self.packages_path)
groups_dir = os.path.realpath(os.path.join(packages_dir, "groups"))
allowed_keys = {
"name",
"description",
"includes",
"tests",
"options",
"dedup_tests",
}
if not manifest_path.startswith(packages_dir + os.sep):
logging.error(
"Manifest path '{}' is outside packages directory '{}'".format(
manifest_path, packages_dir
)
)
exit(1)
if visited is None:
visited = []
if root_manifest_path is None:
root_manifest_path = manifest_path
if manifest_path in visited:
cycle_path = " -> ".join(visited + [manifest_path])
logging.error("Include cycle detected: {}".format(cycle_path))
exit(1)
if not os.path.isfile(manifest_path):
logging.error("Package manifest not found: {}".format(manifest_path))
exit(1)
with open(manifest_path, "r") as f:
manifest = yaml.safe_load(f)
if not isinstance(manifest, dict):
logging.error("Invalid manifest format in '{}'".format(manifest_path))
exit(1)
forbidden = {"config", "device", "config_id", "device_id"}
found = forbidden & set(manifest.keys())
if found:
logging.error(
"Forbidden keys in '{}': {}".format(
manifest_path, ", ".join(sorted(found))
)
)
exit(1)
unknown = set(manifest.keys()) - allowed_keys
if unknown:
logging.error(
"Unknown manifest keys in '{}': {}".format(
manifest_path, ", ".join(sorted(unknown))
)
)
exit(1)
is_group_manifest = manifest_path == groups_dir or manifest_path.startswith(
groups_dir + os.sep
)
if is_group_manifest:
group_forbidden = {"name", "description", "options"}
group_found = group_forbidden & set(manifest.keys())
if group_found:
logging.error(
"Group manifest '{}' must not define: {}".format(
manifest_path, ", ".join(sorted(group_found))
)
)
exit(1)
if "name" in manifest and not isinstance(manifest["name"], str):
logging.error(
"Manifest 'name' must be a string in '{}'".format(manifest_path)
)
exit(1)
if "description" in manifest and not isinstance(manifest["description"], str):
logging.error(
"Manifest 'description' must be a string in '{}'".format(manifest_path)
)
exit(1)
if "dedup_tests" in manifest and not isinstance(manifest["dedup_tests"], bool):
logging.error(
"Manifest 'dedup_tests' must be a boolean in '{}'".format(
manifest_path
)
)
exit(1)
includes = manifest.get("includes", [])
if not isinstance(includes, list) or not all(
isinstance(entry, str) for entry in includes
):
logging.error(
"Manifest 'includes' must be a list of strings in '{}'".format(
manifest_path
)
)
exit(1)
tests = manifest.get("tests")
if not isinstance(tests, list) or not all(
isinstance(entry, str) for entry in tests
):
logging.error(
"Manifest 'tests' must be a list of strings in '{}'".format(
manifest_path
)
)
exit(1)
options = manifest.get("options", {})
if options is None:
options = {}
if not isinstance(options, dict):
logging.error(
"Manifest 'options' must be a map in '{}'".format(manifest_path)
)
exit(1)
if manifest_path == root_manifest_path:
expected_name = os.path.splitext(os.path.basename(manifest_path))[0]
name = manifest.get("name")
if not name:
logging.error(
"Root manifest '{}' must define non-empty 'name'".format(
manifest_path
)
)
exit(1)
dedup_tests = manifest.get("dedup_tests", False)
if name != expected_name:
logging.error(
"Manifest name '{}' does not match filename '{}'".format(
name, expected_name
)
)
exit(1)
else:
if "dedup_tests" in manifest:
logging.error(
"Manifest '{}' must not define 'dedup_tests' "
"(root manifest only)".format(manifest_path)
)
exit(1)
if dedup_tests is None:
dedup_tests = False
# Process includes depth-first
all_tests = []
all_options = {}
manifest_dir = os.path.dirname(manifest_path)
for inc in includes:
if os.path.isabs(inc):
logging.error(
"Include path '{}' in '{}' must be relative".format(
inc, manifest_path
)
)
exit(1)
inc_path = os.path.realpath(os.path.join(manifest_dir, inc))
resolved = self.resolve_manifest(
inc_path,
visited=visited + [manifest_path],
root_manifest_path=root_manifest_path,
dedup_tests=dedup_tests,
)
all_tests.extend(resolved.get("tests", []))
all_options.update(resolved.get("options", {}))
# Append current manifest's tests
all_tests.extend(tests)
if dedup_tests:
# Dedup: keep first occurrence
seen = set()
resolved_tests = []
for t in all_tests:
if t not in seen:
seen.add(t)
resolved_tests.append(t)
else:
resolved_tests = all_tests
# Root options win on conflicts
all_options.update(options)
return {
"name": manifest.get("name", ""),
"description": manifest.get("description", ""),
"tests": resolved_tests,
"options": all_options,
}
def build_package_options(self, options):
from cdrouter.packages import Options as PackageOptions
package_options = PackageOptions()
for key, value in options.items():
if not hasattr(package_options, key):
logging.error("Unknown package option '{}'".format(key))
exit(1)
setattr(package_options, key, value)
return package_options
def package_apply(self):
self.connect()
manifest_path = os.path.join(
self.packages_path, self.args.package_name + ".yaml"
)
resolved = self.resolve_manifest(manifest_path)
name = resolved["name"]
tests = resolved["tests"]
description = resolved.get("description", "")
options = resolved.get("options", {})
if not name:
logging.error(
"Resolved manifest name is empty for '{}'".format(manifest_path)
)
exit(1)
if not tests:
logging.error("Resolved manifest has empty test list for '{}'".format(name))
exit(1)
package_options = self.build_package_options(options)
existing = None
try:
existing = self.cdr.packages.get_by_name(name)
except CDRouterError as e:
error = str(e).lower()
if "not found" not in error and "no such package" not in error:
raise
if existing:
p = self.cdr.packages.get(existing.id)
p.testlist = tests
p.description = description or ""
p.options = package_options
self.cdr.packages.edit(p)
logging.info("Updated package '{}' with {} tests".format(name, len(tests)))
return
from cdrouter.packages import Package
p = Package(name=name, description=description or "", testlist=tests)
p.options = package_options
p = self.cdr.packages.create(p)
logging.info("Created package '{}' with {} tests".format(name, len(tests)))
def main():
logging.basicConfig(
@ -575,6 +844,14 @@ def main():
subparser.add_argument("-n", "--name", help="package name, (default: filename)")
subparser.set_defaults(func=TestbedCDRouter.package_import)
subparser = subparsers.add_parser(
"package_apply", help="apply package from YAML manifest"
)
subparser.add_argument(
"package_name", help="package name (resolves to packages/<name>.yaml)"
)
subparser.set_defaults(func=TestbedCDRouter.package_apply)
subparser = subparsers.add_parser("config_export", help="export configuration")
subparser.add_argument("name", help="configuration name")
subparser.add_argument("filename", help="destination filename")