Coverage for .github / scripts / verify_lambda_imports.py: 100.00%

1#!/usr/bin/env python3 

2"""Import every deployable Python Lambda handler in an isolated process.""" 

3 

4from __future__ import annotations 

5 

6import argparse 

7import os 

8import subprocess 

9import sys 

10from dataclasses import dataclass 

11from pathlib import Path 

12 

13ENTRYPOINT_OVERRIDES = { 

14 "analytics-cleanup": "handler", 

15 "helm-orchestrator": "on_event", 

16} 

17DEFAULT_ENTRYPOINT = "lambda_handler" 

18IMPORT_CHECK = ( 

19 "import importlib,sys;" 

20 "sys.path.insert(0,sys.argv[1]);" 

21 "module=importlib.import_module('handler');" 

22 "entrypoint=getattr(module,sys.argv[2]);" 

23 "raise SystemExit(0 if callable(entrypoint) else 1)" 

24) 

25 

26 

27@dataclass(frozen=True) 

28class HandlerTarget: 

29 """One handler module and the callable configured as its entrypoint.""" 

30 

31 directory: Path 

32 entrypoint: str 

33 

34 

35def discover_handlers(root: Path) -> list[HandlerTarget]: 

36 """Return every non-generated ``lambda/*/handler.py`` target.""" 

37 lambda_root = root.resolve() / "lambda" 

38 targets = [ 

39 HandlerTarget( 

40 directory=handler_path.parent, 

41 entrypoint=ENTRYPOINT_OVERRIDES.get( 

42 handler_path.parent.name, 

43 DEFAULT_ENTRYPOINT, 

44 ), 

45 ) 

46 for handler_path in sorted(lambda_root.glob("*/handler.py")) 

47 if not handler_path.parent.name.endswith("-build") 

48 ] 

49 if not targets: 

50 raise RuntimeError(f"no Python Lambda handlers found under {lambda_root}") 

51 

52 discovered_names = {target.directory.name for target in targets} 

53 stale_overrides = set(ENTRYPOINT_OVERRIDES) - discovered_names 

54 if stale_overrides: 

55 names = ", ".join(sorted(stale_overrides)) 

56 raise RuntimeError(f"entrypoint overrides reference missing handlers: {names}") 

57 return targets 

58 

59 

60def verify_handler(target: HandlerTarget, *, timeout: int = 30) -> None: 

61 """Import one handler in a fresh interpreter and require a callable entrypoint.""" 

62 environment = os.environ.copy() 

63 environment.setdefault("AWS_DEFAULT_REGION", "us-east-1") 

64 environment.setdefault("AWS_REGION", "us-east-1") 

65 environment.setdefault("AWS_EC2_METADATA_DISABLED", "true") 

66 subprocess.run( 

67 [sys.executable, "-c", IMPORT_CHECK, str(target.directory), target.entrypoint], 

68 check=True, 

69 env=environment, 

70 timeout=timeout, 

71 ) 

72 

73 

74def verify_handlers(targets: list[HandlerTarget]) -> None: 

75 """Verify every discovered target without sharing imported module state.""" 

76 for target in targets: 

77 relative_directory = target.directory.name 

78 print(f"=== lambda/{relative_directory} ({target.entrypoint}) ===", flush=True) 

79 verify_handler(target) 

80 print(f"Imported {len(targets)} Python Lambda handlers.") 

81 

82 

83def main(argv: list[str] | None = None) -> int: 

84 parser = argparse.ArgumentParser(description=__doc__) 

85 parser.add_argument( 

86 "--root", 

87 type=Path, 

88 default=Path(__file__).resolve().parents[2], 

89 help="Repository root (defaults to the checkout containing this script)", 

90 ) 

91 args = parser.parse_args(argv) 

92 verify_handlers(discover_handlers(args.root)) 

93 return 0 

94 

95 

96if __name__ == "__main__": 

97 raise SystemExit(main())