Coverage for call_graph / core.py: 93%

1"""Core call graph logic.""" 

2 

3from __future__ import annotations 

4 

5from collections import defaultdict 

6from pathlib import Path 

7 

8from call_graph.parsers.python import PythonCallParser 

9from models import CallGraph, CallEdge, FunctionDefinition 

10 

11 

12def build_call_graph(files, lang="python"): 

13 parser = PythonCallParser() 

14 from parsers.python import parse 

15 

16 functions = {} 

17 all_functions = [] 

18 

19 for filepath in files: 

20 source = filepath.read_text() 

21 root = parse(source) 

22 defs = parser.extract_function_definitions(root, source, str(filepath)) 

23 all_functions.extend(defs) 

24 for d in defs: 

25 functions[(d.name, d.filepath)] = d 

26 

27 calls = [] 

28 for filepath in files: 

29 source = filepath.read_text() 

30 root = parse(source) 

31 file_calls = parser.extract_calls(root, source, str(filepath), all_functions) 

32 calls.extend(file_calls) 

33 

34 resolved_calls = [] 

35 for call in calls: 

36 if call.call_type == "external": 

37 resolved_calls.append(call) 

38 continue 

39 

40 # Find all matching functions by name 

41 callee_defs = [f for f in all_functions if f.name == call.callee_function] 

42 if not callee_defs: 

43 call.call_type = "external" 

44 resolved_calls.append(call) 

45 continue 

46 

47 # Prioritize functions in the same file as the caller 

48 same_file_defs = [f for f in callee_defs if f.filepath == call.caller_file] 

49 if same_file_defs: 

50 callee = same_file_defs[0] 

51 else: 

52 callee = callee_defs[0] 

53 

54 call.callee_file = callee.filepath 

55 call.callee_line = callee.line 

56 resolved_calls.append(call) 

57 

58 reverse_calls = defaultdict(list) 

59 for call in resolved_calls: 

60 reverse_calls[call.callee_function].append(call) 

61 

62 return CallGraph( 

63 functions=functions, calls=resolved_calls, reverse_calls=dict(reverse_calls) 

64 ) 

65 

66 

67def find_call_cycles(graph): 

68 colors = {k: "WHITE" for k in graph.functions.keys()} 

69 cycles = [] 

70 

71 def dfs(key, path, edges): 

72 colors[key] = "GRAY" 

73 path.append(key) 

74 

75 func_name = key[0] 

76 calls = [ 

77 c 

78 for c in graph.calls 

79 if c.caller_function == func_name and c.call_type == "local" 

80 ] 

81 for call in calls: 

82 callee_key = ( 

83 (call.callee_function, call.callee_file) if call.callee_file else None 

84 ) 

85 if not callee_key: 

86 continue 

87 if callee_key in colors: 

88 if colors[callee_key] == "GRAY": 

89 cycles.append(path.copy() + [callee_key]) 

90 elif colors[callee_key] == "WHITE": 

91 dfs(callee_key, list(path), list(edges)) 

92 

93 colors[key] = "BLACK" 

94 

95 for key in graph.functions.keys(): 

96 if colors[key] == "WHITE": 

97 dfs(key, [], []) 

98 

99 return cycles 

100 

101 

102def find_unused_functions(graph): 

103 used = {c.callee_function for c in graph.calls} 

104 

105 unused = [] 

106 for (name, _), func_def in graph.functions.items(): 

107 if name not in used and not name.startswith("_"): 

108 unused.append(func_def) 

109 

110 return unused