Coverage for scripts/test_handlers.py: 100%

1#!/usr/bin/env python3 

2 

3import ast 

4from pathlib import Path 

5 

6 

7EXAMPLES_IMPORT_PREFIX = "async_durable_execution_examples" 

8HANDLER_SUFFIX = ".handler" 

9 

10 

11def load_test_handlers(test_root: Path) -> set[str]: 

12 """Load handler identifiers referenced by the example pytest suite.""" 

13 handlers: set[str] = set() 

14 for path in sorted(test_root.rglob("test_*.py")): 

15 handlers.update(load_test_handlers_from_file(path)) 

16 return handlers 

17 

18 

19def load_test_handlers_from_file(path: Path) -> set[str]: 

20 """Load handler identifiers from a single pytest file.""" 

21 tree = ast.parse(path.read_text(), filename=str(path)) 

22 imported_modules: dict[str, str] = {} 

23 

24 for node in ast.walk(tree): 

25 if isinstance(node, ast.ImportFrom) and node.module: 

26 if not node.module.startswith(EXAMPLES_IMPORT_PREFIX): 

27 continue 

28 for alias in node.names: 

29 # Example: from async_durable_execution_examples.step import step 

30 imported_modules[alias.asname or alias.name] = ( 

31 f"{node.module}.{alias.name}{HANDLER_SUFFIX}" 

32 ) 

33 

34 handlers: set[str] = set() 

35 for node in ast.walk(tree): 

36 if not isinstance(node, ast.Call): 

37 continue 

38 if not _is_durable_execution_marker(node.func): 

39 continue 

40 

41 handler_name = _get_handler_name(node) 

42 if not handler_name: 

43 continue 

44 

45 handler = imported_modules.get(handler_name) 

46 if handler: 

47 handlers.add(handler) 

48 

49 return handlers 

50 

51 

52def _is_durable_execution_marker(node: ast.AST) -> bool: 

53 return ( 

54 isinstance(node, ast.Attribute) 

55 and node.attr == "durable_execution" 

56 and isinstance(node.value, ast.Attribute) 

57 and node.value.attr == "mark" 

58 and isinstance(node.value.value, ast.Name) 

59 and node.value.value.id == "pytest" 

60 ) 

61 

62 

63def _get_handler_name(node: ast.Call) -> str | None: 

64 for keyword in node.keywords: 

65 if keyword.arg != "handler": 

66 continue 

67 value = keyword.value 

68 if ( 

69 isinstance(value, ast.Attribute) 

70 and value.attr == "handler" 

71 and isinstance(value.value, ast.Name) 

72 ): 

73 return value.value.id 

74 return None