Add import smoke tests to PR CI (#19091)

Summary:

Add a PR QNN import job that validates backend module imports and statically checks internal imports for runnable Qualcomm example entrypoints. Also fix the stale `ExecutorchBackendConfig` import in the QAIHub stable diffusion example so the new check passes.

Differential Revision: D102218906

Author: abhinaykukkadapu

.ci/scripts/test_qnn_example_imports.py

@@ -0,0 +1,374 @@
+#!/usr/bin/env python3
+# Copyright (c) Qualcomm Innovation Center, Inc.
+# All rights reserved
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Validate internal imports for QNN example entrypoints.
+
+Entrypoints are discovered dynamically from `examples/qualcomm` by looking for
+Python files that define a standard `if __name__ == "__main__"` block. This
+keeps the check focused on runnable scripts while avoiding a hardcoded list
+that drifts as examples are added, moved, or removed.
+"""
+
+import ast
+import importlib.util
+import sys
+from pathlib import Path
+
+
+EXAMPLE_MODULE_PREFIX = "executorch.examples.qualcomm."
+
+
+def resolve_examples_root():
+    for parent in Path(__file__).resolve().parents:
+        candidate = parent / "examples" / "qualcomm"
+        if candidate.is_dir():
+            return candidate
+    return None
+
+
+def is_main_guard(test: ast.AST) -> bool:
+    if not isinstance(test, ast.Compare):
+        return False
+    if len(test.ops) != 1 or len(test.comparators) != 1:
+        return False
+    if not isinstance(test.ops[0], ast.Eq):
+        return False
+    if not isinstance(test.left, ast.Name) or test.left.id != "__name__":
+        return False
+    comparator = test.comparators[0]
+    return isinstance(comparator, ast.Constant) and comparator.value == "__main__"
+
+
+def is_entrypoint(tree: ast.AST) -> bool:
+    for node in ast.walk(tree):
+        if isinstance(node, ast.If) and is_main_guard(node.test):
+            return True
+    return False
+
+
+def discover_entrypoints(examples_root: Path) -> list[str]:
+    entrypoints = []
+    for path in sorted(examples_root.rglob("*.py")):
+        if path.name == "__init__.py":
+            continue
+        tree = ast.parse(path.read_text(), filename=str(path))
+        if is_entrypoint(tree):
+            entrypoints.append(path.relative_to(examples_root).as_posix())
+    return entrypoints
+
+
+def module_base_path(repo_root: Path, module_name: str) -> Path:
+    return repo_root.joinpath(*module_name.split(".")[1:])
+
+
+def module_exists(repo_root: Path, module_name: str) -> bool:
+    base_path = module_base_path(repo_root, module_name)
+    if base_path.is_dir() or base_path.with_suffix(".py").is_file():
+        return True
+
+    try:
+        return importlib.util.find_spec(module_name) is not None
+    except (AttributeError, ImportError, ValueError):
+        return False
+
+
+def module_source_file(repo_root: Path, module_name: str):
+    base_path = module_base_path(repo_root, module_name)
+    file_path = base_path.with_suffix(".py")
+    if file_path.is_file():
+        return file_path
+    init_path = base_path / "__init__.py"
+    if init_path.is_file():
+        return init_path
+    return None
+
+
+def source_module_name(repo_root: Path, source_file: Path) -> str:
+    relative_path = source_file.relative_to(repo_root)
+    if relative_path.name == "__init__.py":
+        relative_path = relative_path.parent
+    else:
+        relative_path = relative_path.with_suffix("")
+    return "executorch." + ".".join(relative_path.parts)
+
+
+def target_names(node):
+    names = set()
+    if isinstance(node, ast.Name):
+        names.add(node.id)
+    elif isinstance(node, (ast.Tuple, ast.List)):
+        for element in node.elts:
+            names.update(target_names(element))
+    return names
+
+
+def collect_names_from_import_from(
+    repo_root: Path,
+    module_name: str,
+    node: ast.ImportFrom,
+    export_cache: dict[Path, set[str]],
+    is_package: bool,
+) -> set[str]:
+    names = set()
+    try:
+        imported_module = resolve_from_module(module_name, node, is_package=is_package)
+    except ImportError:
+        imported_module = ""
+
+    for alias in node.names:
+        if alias.name == "*":
+            if imported_module.startswith("executorch.") and module_exists(
+                repo_root, imported_module
+            ):
+                source_file = module_source_file(repo_root, imported_module)
+                if source_file is not None:
+                    names.update(exported_names(repo_root, source_file, export_cache))
+            continue
+
+        names.add(alias.asname or alias.name)
+
+    return names
+
+
+def nested_statement_bodies(node: ast.stmt) -> list[list[ast.stmt]] | None:
+    if isinstance(node, ast.If):
+        return [node.body, node.orelse]
+    if isinstance(node, ast.Try):
+        bodies = [node.body, node.orelse, node.finalbody]
+        bodies.extend(handler.body for handler in node.handlers)
+        return bodies
+    if isinstance(node, (ast.For, ast.AsyncFor, ast.While, ast.With, ast.AsyncWith)):
+        return [node.body, getattr(node, "orelse", [])]
+    if isinstance(node, ast.Match):
+        return [case.body for case in node.cases]
+    return None
+
+
+def collect_names_from_node(
+    repo_root: Path,
+    module_name: str,
+    node: ast.stmt,
+    export_cache: dict[Path, set[str]],
+    is_package: bool,
+) -> set[str]:
+    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+        return {node.name}
+    if isinstance(node, ast.Import):
+        return {alias.asname or alias.name.split(".")[0] for alias in node.names}
+    if isinstance(node, ast.ImportFrom):
+        return collect_names_from_import_from(
+            repo_root, module_name, node, export_cache, is_package
+        )
+    if isinstance(node, ast.Assign):
+        names = set()
+        for target in node.targets:
+            names.update(target_names(target))
+        return names
+    if isinstance(node, ast.AnnAssign):
+        return target_names(node.target)
+
+    bodies = nested_statement_bodies(node)
+    if bodies is None:
+        return set()
+
+    names = set()
+    for body in bodies:
+        names.update(
+            collect_exported_names(
+                repo_root, module_name, body, export_cache, is_package
+            )
+        )
+    return names
+
+
+def collect_exported_names(
+    repo_root: Path,
+    module_name: str,
+    body: list[ast.stmt],
+    export_cache: dict[Path, set[str]],
+    is_package: bool,
+) -> set[str]:
+    names = set()
+    for node in body:
+        names.update(
+            collect_names_from_node(
+                repo_root, module_name, node, export_cache, is_package
+            )
+        )
+    return names
+
+
+def exported_names(
+    repo_root: Path, source_file: Path, export_cache: dict[Path, set[str]]
+) -> set[str]:
+    cached_names = export_cache.get(source_file)
+    if cached_names is not None:
+        return cached_names
+
+    names = set()
+    export_cache[source_file] = names
+
+    module_name = source_module_name(repo_root, source_file)
+    tree = ast.parse(source_file.read_text(), filename=str(source_file))
+    names.update(
+        collect_exported_names(
+            repo_root,
+            module_name,
+            tree.body,
+            export_cache,
+            source_file.name == "__init__.py",
+        )
+    )
+    return names
+
+
+def resolve_from_module(
+    module_name: str, node: ast.ImportFrom, is_package: bool = False
+) -> str:
+    if node.level == 0:
+        return node.module or ""
+    package_name = module_name if is_package else module_name.rpartition(".")[0]
+    relative_name = "." * node.level + (node.module or "")
+    return importlib.util.resolve_name(relative_name, package_name)
+
+
+def validate_import_from(
+    repo_root: Path,
+    module_name: str,
+    entrypoint: str,
+    node: ast.ImportFrom,
+    export_cache: dict[Path, set[str]],
+) -> tuple[list[str], int]:
+    failures = []
+    try:
+        imported_module = resolve_from_module(module_name, node)
+    except ImportError as error:
+        failures.append(
+            f"{entrypoint}:{node.lineno} relative import could not be resolved: {error}"
+        )
+        return failures, 0
+
+    if not imported_module.startswith("executorch."):
+        return failures, 0
+
+    checks = 1
+    if not module_exists(repo_root, imported_module):
+        failures.append(
+            f"{entrypoint}:{node.lineno} missing internal module `{imported_module}`"
+        )
+        return failures, checks
+
+    source_file = module_source_file(repo_root, imported_module)
+    exported = (
+        exported_names(repo_root, source_file, export_cache) if source_file else set()
+    )
+
+    for alias in node.names:
+        if alias.name == "*":
+            continue
+        submodule_name = f"{imported_module}.{alias.name}"
+        if module_exists(repo_root, submodule_name):
+            checks += 1
+            continue
+        if source_file is None or alias.name not in exported:
+            failures.append(
+                f"{entrypoint}:{node.lineno} unresolved internal import "
+                f"`{alias.name}` from `{imported_module}`"
+            )
+        checks += 1
+
+    return failures, checks
+
+
+def validate_entrypoint(
+    repo_root: Path,
+    examples_root: Path,
+    relative_path: str,
+    export_cache: dict[Path, set[str]],
+) -> tuple[list[str], int]:
+    entrypoint_path = examples_root / relative_path
+    if not entrypoint_path.is_file():
+        return [f"{relative_path}: allowlisted entrypoint not found"], 0
+
+    module_name = EXAMPLE_MODULE_PREFIX + str(
+        Path(relative_path).with_suffix("")
+    ).replace("/", ".")
+    tree = ast.parse(entrypoint_path.read_text(), filename=str(entrypoint_path))
+
+    failures = []
+    checks = 0
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name.startswith("executorch."):
+                    checks += 1
+                    if not module_exists(repo_root, alias.name):
+                        failures.append(
+                            f"{relative_path}:{node.lineno} missing internal module `{alias.name}`"
+                        )
+        elif isinstance(node, ast.ImportFrom):
+            import_failures, import_checks = validate_import_from(
+                repo_root,
+                module_name,
+                relative_path,
+                node,
+                export_cache,
+            )
+            failures.extend(import_failures)
+            checks += import_checks
+
+    return failures, checks
+
+
+def main():
+    if sys.version_info < (3, 10):
+        print("Python 3.10+ is required to parse QNN example sources")
+        sys.exit(1)
+
+    examples_root = resolve_examples_root()
+    if examples_root is None:
+        print(f"QNN examples root not found from {Path(__file__).resolve()}")
+        sys.exit(1)
+
+    repo_root = examples_root.parent.parent
+    entrypoints = discover_entrypoints(examples_root)
+    if not entrypoints:
+        print(f"No QNN example entrypoints found under {examples_root}")
+        sys.exit(1)
+
+    all_failures = []
+    total_checks = 0
+    export_cache = {}
+
+    for relative_path in entrypoints:
+        failures, checks = validate_entrypoint(
+            repo_root, examples_root, relative_path, export_cache
+        )
+        all_failures.extend(failures)
+        total_checks += checks
+
+    if total_checks == 0:
+        print("No QNN example imports were checked")
+        sys.exit(1)
+
+    if all_failures:
+        print(
+            f"{len(all_failures)} unresolved internal import(s) "
+            f"across {len(entrypoints)} QNN example entrypoint(s):"
+        )
+        for failure in all_failures:
+            print(f"  FAIL: {failure}")
+        sys.exit(1)
+
+    print(
+        f"Validated {total_checks} internal import(s) across "
+        f"{len(entrypoints)} QNN example entrypoint(s)"
+    )
+
+
+if __name__ == "__main__":
+    main()