Make checks optional for yaml_resolver

Author: Derek Hower

lib/test/test_yaml_loader.rb

@@ -21,7 +21,7 @@ def resolve_yaml(yaml)
 
       stdout, stderr, status =
         Dir.chdir($root) do
-          Open3.capture3("/bin/bash -c \"source #{$root}/.home/.venv/bin/activate && #{$root}/.home/.venv/bin/python3 #{$root}/lib/yaml_resolver.py resolve --no-progress #{arch_dir} #{resolved_dir}\"")
+          Open3.capture3("/bin/bash -c \"source #{$root}/.home/.venv/bin/activate && #{$root}/.home/.venv/bin/python3 #{$root}/lib/yaml_resolver.py resolve --no-progress --no-checks #{arch_dir} #{resolved_dir}\"")
         end
       # puts stdout
       # puts stderr
@@ -48,7 +48,7 @@ def resolve_multi_yaml(*yamls)
         File.write(test_dir / "test#{i + 1}.yaml", yaml)
       end
 
-      system "/bin/bash -c \"source #{$root}/.home/.venv/bin/activate && #{$root}/.home/.venv/bin/python3 #{$root}/lib/yaml_resolver.py resolve #{arch_dir} #{resolved_dir}\""
+      system "/bin/bash -c \"source #{$root}/.home/.venv/bin/activate && #{$root}/.home/.venv/bin/python3 #{$root}/lib/yaml_resolver.py resolve --no-checks #{arch_dir} #{resolved_dir}\""
       # `source #{$root}/.home/.venv/bin/activate && python3 #{$root}/lib/yaml_resolver.py resolve #{arch_dir} #{resolved_dir}`
 
       if $CHILD_STATUS == 0

lib/yaml_resolver.py

@@ -190,7 +190,7 @@ def dig(obj : dict, *keys):
     return None
 
 resolved_objs = {}
-def resolve(rel_path : str | Path, arch_root : str | Path) -> dict:
+def resolve(rel_path : str | Path, arch_root : str | Path, do_checks: bool) -> dict:
   """Resolve the file at arch_root/rel_path by expanding operators and applying defaults
 
   Parameters
@@ -209,22 +209,22 @@ def resolve(rel_path : str | Path, arch_root : str | Path) -> dict:
     return resolved_objs[str(rel_path)]
   else:
     unresolved_arch_data = read_yaml(os.path.join(arch_root, rel_path))
-    if not "name" in unresolved_arch_data:
+    if do_checks and (not "name" in unresolved_arch_data):
       print(f"ERROR: Missing 'name' key in {arch_root}/{rel_path}", file=sys.stderr)
       exit(1)
     fn_name = Path(rel_path).stem
-    if fn_name != unresolved_arch_data["name"]:
+    if do_checks and (fn_name != unresolved_arch_data["name"]):
       print(f"ERROR: 'name' key ({unresolved_arch_data["name"]}) must match filename ({fn_name} in {arch_root}/{rel_path}", file=sys.stderr)
       exit(1)
-    resolved_objs[str(rel_path)] = _resolve(unresolved_arch_data, [], rel_path, unresolved_arch_data, arch_root)
+    resolved_objs[str(rel_path)] = _resolve(unresolved_arch_data, [], rel_path, unresolved_arch_data, arch_root, do_checks)
     return resolved_objs[str(rel_path)]
 
-def _resolve(obj, obj_path, obj_file_path, doc_obj, arch_root):
+def _resolve(obj, obj_path, obj_file_path, doc_obj, arch_root, do_checks):
   if not (isinstance(obj, list) or isinstance(obj, dict)):
     return obj
 
   if isinstance(obj, list):
-    obj = list(map(lambda o: _resolve(o, obj_path, obj_file_path, doc_obj, arch_root), obj))
+    obj = list(map(lambda o: _resolve(o, obj_path, obj_file_path, doc_obj, arch_root, do_checks), obj))
     return obj
 
   if "$inherits" in obj:
@@ -245,16 +245,16 @@ def _resolve(obj, obj_path, obj_file_path, doc_obj, arch_root):
         ref_obj = dig(doc_obj, *ref_obj_path)
         if ref_obj == None:
           raise ValueError(f"{ref_obj_path} cannot be found in #{doc_obj}")
-        ref_obj = _resolve(ref_obj, ref_obj_path, ref_file_path, doc_obj, arch_root)
+        ref_obj = _resolve(ref_obj, ref_obj_path, ref_file_path, doc_obj, arch_root, do_checks)
       else:
         # this is a reference to another doc
         if not os.path.exists(os.path.join(arch_root, ref_file_path)):
           raise ValueError(f"{ref_file_path} does not exist in {arch_root}/")
 
-        ref_doc_obj = resolve(ref_file_path, arch_root)
+        ref_doc_obj = resolve(ref_file_path, arch_root, do_checks)
         ref_obj = dig(ref_doc_obj, *ref_obj_path)
 
-        ref_obj = _resolve(ref_obj, ref_obj_path, ref_file_path, ref_doc_obj, arch_root)
+        ref_obj = _resolve(ref_obj, ref_obj_path, ref_file_path, ref_doc_obj, arch_root, do_checks)
 
       for key in ref_obj:
         if key == "$parent_of" or key == "$child_of":
@@ -288,12 +288,12 @@ def _resolve(obj, obj_path, obj_file_path, doc_obj, arch_root):
       if not key in obj:
         final_obj[key] = parent_obj[key]
       elif not key in parent_obj:
-        final_obj[key] = _resolve(obj[key], obj_path + [key], obj_file_path, doc_obj, arch_root)
+        final_obj[key] = _resolve(obj[key], obj_path + [key], obj_file_path, doc_obj, arch_root, do_checks)
       else:
         if isinstance(parent_obj[key], dict):
           final_obj[key] = merge(yaml.load('{}'), parent_obj[key], obj[key], strategy=Strategy.REPLACE)
         else:
-          final_obj[key] = _resolve(obj[key], obj_path + [key], obj_file_path, doc_obj, arch_root)
+          final_obj[key] = _resolve(obj[key], obj_path + [key], obj_file_path, doc_obj, arch_root, do_checks)
 
     if "$remove" in final_obj:
       if isinstance(final_obj["$remove"], list):
@@ -308,7 +308,7 @@ def _resolve(obj, obj_path, obj_file_path, doc_obj, arch_root):
     return final_obj
   else:
     for key in obj:
-      obj[key] = _resolve(obj[key], obj_path + [key], obj_file_path, doc_obj, arch_root)
+      obj[key] = _resolve(obj[key], obj_path + [key], obj_file_path, doc_obj, arch_root, do_checks)
 
     if "$remove" in obj:
       if isinstance(obj["$remove"], list):
@@ -394,7 +394,7 @@ def _get_schema(uri):
   return schemas[rel_path]
 
 
-def resolve_file(rel_path : str | Path, arch_dir: str | Path, resolved_dir: str | Path):
+def resolve_file(rel_path : str | Path, arch_dir: str | Path, resolved_dir: str | Path, do_checks: bool):
   """Read object at arch_dir/rel_path, resolve it, and write it as YAML to resolved_dir/rel_path
 
   Parameters
@@ -414,10 +414,10 @@ def resolve_file(rel_path : str | Path, arch_dir: str | Path, resolved_dir: str
   elif not os.path.exists(resolved_path) or (os.path.getmtime(arch_path) > os.path.getmtime(resolved_path)) or (os.path.getmtime(__file__) > os.path.getmtime(resolved_path)):
     if os.path.exists(resolved_path):
       os.remove(resolved_path)
-    resolved_obj = resolve(rel_path, args.arch_dir)
+    resolved_obj = resolve(rel_path, args.arch_dir, do_checks)
     resolved_obj["$source"] = os.path.join(args.arch_dir, rel_path)
 
-    if "$schema" in resolved_obj:
+    if do_checks and ("$schema" in resolved_obj):
       schema = _get_schema(resolved_obj["$schema"])
       try:
         schema.validate(instance=resolved_obj)
@@ -443,6 +443,7 @@ def resolve_file(rel_path : str | Path, arch_dir: str | Path, resolved_dir: str
   all_parser.add_argument("arch_dir", type=str, help="Unresolved architecture (input) directory")
   all_parser.add_argument("resolved_dir", type=str, help="Resolved architecture (output) directory")
   all_parser.add_argument("--no-progress", action="store_true", help="Don't display progress bar")
+  all_parser.add_argument("--no-checks", action="store_true", help="Don't verify schema")
 
   args =  cmdparser.parse_args()
 
@@ -473,6 +474,6 @@ def resolve_file(rel_path : str | Path, arch_dir: str | Path, resolved_dir: str
     for arch_path in iter:
       resolved_arch_path = f"{UDB_ROOT}/{args.resolved_dir}/{arch_path}" if not os.path.isabs(args.resolved_dir) else f"{args.resolved_dir}/{arch_path}"
       os.makedirs(os.path.dirname(resolved_arch_path), exist_ok=True)
-      resolve_file(arch_path, args.arch_dir, args.resolved_dir)
+      resolve_file(arch_path, args.arch_dir, args.resolved_dir, not args.no_checks)
 
     print(f"[INFO] Resolved architecture files written to {args.resolved_dir}")