Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / emit / validation.py: 93%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Structural validation performed before a generated bundle is published.""" 

16 

17from __future__ import annotations 

18 

19import ast 

20from collections import Counter 

21from typing import Mapping 

22 

23 

24def verify_generated_source_structure(files: Mapping[str, str]) -> None: 

25 """Reject ambiguous forward composition in generated Python sources.""" 

26 errors: list[str] = [] 

27 for filename, source_text in sorted(files.items()): 

28 if not filename.endswith(".py") or filename == "__init__.py": 

29 continue 

30 try: 

31 tree = ast.parse(source_text, filename=filename) 

32 except SyntaxError as exc: 

33 errors.append(f"{filename}: invalid Python source ({exc.msg})") 

34 continue 

35 for class_node in ( 

36 node for node in ast.walk(tree) if isinstance(node, ast.ClassDef) 

37 ): 

38 _verify_class(filename, class_node, errors) 

39 if errors: 

40 raise RuntimeError( 

41 "codegen: generated forward composition is invalid: " 

42 + "; ".join(errors) 

43 ) 

44 

45 

46def _verify_class( 

47 filename: str, class_node: ast.ClassDef, errors: list[str] 

48) -> None: 

49 """Append structural errors found in one generated class.""" 

50 methods = [ 

51 node 

52 for node in class_node.body 

53 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) 

54 ] 

55 counts = Counter(method.name for method in methods) 

56 duplicates = sorted(name for name, count in counts.items() if count > 1) 

57 location = f"{filename}:{class_node.name}" 

58 if duplicates: 

59 errors.append(f"{location} defines methods more than once: {duplicates}") 

60 for implementation in ( 

61 method for method in methods if method.name == "_forward_impl" 

62 ): 

63 if _calls_self_method(implementation, "_forward_impl"): 

64 errors.append(f"{location}._forward_impl calls itself") 

65 dispatch_methods = [ 

66 method for method in methods if method.name in {"forward", "_forward_impl"} 

67 ] 

68 for state_flag in _unused_enable_flags(class_node, dispatch_methods): 

69 errors.append( 

70 f"{location} declares {state_flag!r} but its forward pipeline " 

71 "does not read it" 

72 ) 

73 

74 

75def _unused_enable_flags( 

76 class_node: ast.ClassDef, dispatch_methods: list[ast.AST] 

77) -> list[str]: 

78 """Return declared enable flags not consumed by the forward pipeline.""" 

79 return [ 

80 flag 

81 for flag in sorted(_class_enable_flags(class_node)) 

82 if not any(_reads_self_attribute(method, flag) for method in dispatch_methods) 

83 ] 

84 

85 

86def _class_enable_flags(class_node: ast.ClassDef) -> set[str]: 

87 """Return conditional parallel enable flags declared by a class.""" 

88 flags: set[str] = set() 

89 for statement in class_node.body: 

90 targets = ( 

91 statement.targets 

92 if isinstance(statement, ast.Assign) 

93 else (statement.target,) if isinstance(statement, ast.AnnAssign) else () 

94 ) 

95 for target in targets: 

96 if isinstance(target, ast.Name) and target.id.endswith("_enable"): 

97 flags.add(target.id) 

98 return flags 

99 

100 

101def _calls_self_method(method: ast.AST, method_name: str) -> bool: 

102 return any( 

103 isinstance(node.func, ast.Attribute) 

104 and isinstance(node.func.value, ast.Name) 

105 and node.func.value.id == "self" 

106 and node.func.attr == method_name 

107 for node in ast.walk(method) 

108 if isinstance(node, ast.Call) 

109 ) 

110 

111 

112def _reads_self_attribute(method: ast.AST, attribute_name: str) -> bool: 

113 return any( 

114 isinstance(node, ast.Attribute) 

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

116 and node.value.id == "self" 

117 and node.attr == attribute_name 

118 for node in ast.walk(method) 

119 ) 

120 

121 

122__all__ = ["verify_generated_source_structure"]