Azure DevOps Demo Models · main
validate_catalog.py

tools/validate_catalog.py · 7,686 bytes · read-only public source view
from __future__ import annotations

import argparse
import ast
import json
import re
from pathlib import Path

parser = argparse.ArgumentParser(description="Validate the ModelOps demonstration catalogue.")
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1], help="Repository root containing catalog.json")
args = parser.parse_args()
root = args.root.resolve()
errors: list[str] = []
ignored = {'.git', '.artifacts', '__pycache__', 'bin', 'obj', 'outputs'}


def fail(message: str) -> None:
    errors.append(message)


def ignored_path(path: Path) -> bool:
    return any(part in ignored for part in path.parts)


for path in root.rglob('*'):
    if not path.is_file() or ignored_path(path):
        continue
    data = path.read_bytes()
    bad = sorted({value for value in data if value < 9 or value in (11, 12) or 13 < value < 32})
    if bad:
        fail(f'Control characters {bad} in {path.relative_to(root)}')

try:
    catalog = json.loads((root / 'catalog.json').read_text(encoding='utf-8'))
except Exception as exc:
    raise SystemExit(f'catalog.json is invalid: {exc}')

models = catalog.get('models', [])
if len(models) != 10:
    fail(f'Catalogue must contain exactly ten models; found {len(models)}')

catalog_fields = {'modelKey', 'displayName', 'industry', 'technology', 'version', 'criticality', 'path', 'engineKey', 'description', 'tags'}
manifest_fields = {'schemaVersion', 'modelKey', 'displayName', 'description', 'industry', 'technology', 'version', 'engineKey', 'runtime', 'execution', 'parameters', 'inputs', 'outputs', 'resources', 'governance'}
keys: set[str] = set()
engines: set[str] = set()
industries: set[str] = set()
technologies: set[str] = set()

for entry in models:
    missing_catalog = catalog_fields - entry.keys()
    if missing_catalog:
        fail(f"Catalogue entry {entry.get('modelKey', '<missing>')} is missing {sorted(missing_catalog)}")
    key = entry.get('modelKey', '')
    if not key or key in keys:
        fail(f'Duplicate or missing modelKey: {key}')
    keys.add(key)
    engines.add(entry.get('engineKey', ''))
    industries.add(entry.get('industry', ''))
    technologies.add(entry.get('technology', ''))
    folder = root / entry.get('path', '')
    try:
        folder.resolve().relative_to(root.resolve())
    except ValueError:
        fail(f'{key}: catalogue path leaves the repository root')
        continue
    required_files = [
        folder / 'modelops.json',
        folder / 'sample-inputs' / 'config.json',
        folder / 'expected-output' / 'README.md',
        folder / 'run.ps1',
        folder / 'run.sh',
    ]
    for required in required_files:
        if not required.is_file():
            fail(f'Missing required model file: {required.relative_to(root)}')
    source_files = [p for p in (folder / 'src').rglob('*') if p.is_file()] if (folder / 'src').is_dir() else []
    if not source_files:
        fail(f'{key}: no representative source files')

    manifest_path = folder / 'modelops.json'
    if not manifest_path.is_file():
        continue
    try:
        manifest = json.loads(manifest_path.read_text(encoding='utf-8'))
    except Exception as exc:
        fail(f'Invalid manifest {manifest_path.relative_to(root)}: {exc}')
        continue
    missing_manifest = manifest_fields - manifest.keys()
    if missing_manifest:
        fail(f'{key}: manifest missing {sorted(missing_manifest)}')
    for field in ('modelKey', 'displayName', 'industry', 'technology', 'version', 'engineKey'):
        if manifest.get(field) != entry.get(field):
            fail(f'{key}: manifest/catalogue {field} mismatch')
    parameter_keys = [x.get('key') for x in manifest.get('parameters', [])]
    if any(not value for value in parameter_keys) or len(parameter_keys) != len(set(parameter_keys)):
        fail(f'{key}: duplicate or missing parameter key')
    input_keys = [x.get('key') for x in manifest.get('inputs', [])]
    if any(not value for value in input_keys) or len(input_keys) != len(set(input_keys)):
        fail(f'{key}: duplicate or missing input key')
    output_keys = [x.get('key') for x in manifest.get('outputs', [])]
    if any(not value for value in output_keys) or len(output_keys) != len(set(output_keys)):
        fail(f'{key}: duplicate or missing output key')
    content_types = {x.get('contentType') for x in manifest.get('outputs', [])}
    if 'text/csv' not in content_types or 'application/json' not in content_types:
        fail(f'{key}: outputs must include CSV and JSON')
    for output in manifest.get('outputs', []):
        value = output.get('pathTemplate', '')
        if not value or value.startswith(('/', '\\')) or '..' in Path(value).parts:
            fail(f'{key}: unsafe output path {value}')
    for input_item in manifest.get('inputs', []):
        sample_path = input_item.get('samplePath', '')
        if sample_path:
            target = folder / sample_path
            if not target.is_file():
                fail(f'{key}: sample input does not exist: {sample_path}')

    config_path = folder / 'sample-inputs' / 'config.json'
    if config_path.is_file():
        try:
            config = json.loads(config_path.read_text(encoding='utf-8'))
            if config.get('modelKey') not in (None, key):
                fail(f'{key}: sample configuration modelKey mismatch')
            parameters = config.get('parameters', config)
            unknown = set(parameters) - set(parameter_keys) - {'schemaVersion', 'modelKey', 'inputs'}
            if unknown:
                fail(f'{key}: sample configuration has unknown parameters {sorted(unknown)}')
        except Exception as exc:
            fail(f'{key}: invalid sample configuration: {exc}')

runner_path = root / 'tools' / 'modelops_demo_runner.py'
try:
    runner_tree = ast.parse(runner_path.read_text(encoding='utf-8'))
    runner_engines: set[str] = set()
    for node in ast.walk(runner_tree):
        if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == 'ENGINES' for target in node.targets) and isinstance(node.value, ast.Dict):
            for key_node in node.value.keys:
                if isinstance(key_node, ast.Constant) and isinstance(key_node.value, str):
                    runner_engines.add(key_node.value)
    if runner_engines != engines:
        fail(f'Universal runner engine allowlist mismatch. Catalogue={sorted(engines)}, runner={sorted(runner_engines)}')
except Exception as exc:
    fail(f'Unable to inspect universal runner allowlist: {exc}')

if len(industries) < 8:
    fail(f'Expected broad industry coverage; found {len(industries)} industries')
if len(technologies) < 6:
    fail(f'Expected broad technology coverage; found {len(technologies)} technologies')

for path in root.rglob('*'):
    if path.is_dir() and path.name in {'bin', 'obj', '__pycache__'} and not ignored_path(path.parent):
        fail(f'Generated directory present: {path.relative_to(root)}')
    if path.is_file() and not ignored_path(path) and path.resolve() != Path(__file__).resolve():
        text = path.read_text(encoding='utf-8', errors='ignore')
        if any(marker in text for marker in ('<' * 7, '>' * 7)):
            fail(f'Merge conflict marker in {path.relative_to(root)}')
        if re.search(r'(?i)(personalaccesstoken|apikey|password|runnerkey)[ \t]*[=:][ \t]*["\']?[A-Za-z0-9_\-]{20,}', text):
            fail(f'Credential-shaped value in {path.relative_to(root)}')

if errors:
    for error in errors:
        print(f'ERROR: {error}')
    raise SystemExit(1)
print(f'Validated {len(models)} models, {len(industries)} industries, {len(technologies)} technologies and {len(engines)} safe engines')