Coverage for src/debputy/lsp/lsp_features.py: 57%

1import collections 

2import inspect 

3from typing import Callable, TypeVar, Sequence, Union, Dict, List, Optional 

4 

5from lsprotocol.types import ( 

6 TEXT_DOCUMENT_WILL_SAVE_WAIT_UNTIL, 

7 TEXT_DOCUMENT_CODE_ACTION, 

8 DidChangeTextDocumentParams, 

9 Diagnostic, 

10 DidOpenTextDocumentParams, 

11 SemanticTokensLegend, 

12) 

13 

14from debputy.plugin.api.feature_set import PluginProvidedFeatureSet 

15 

16try: 

17 from pygls.server import LanguageServer 

18 from debputy.lsp.debputy_ls import DebputyLanguageServer 

19except ImportError: 

20 pass 

21 

22from debputy.linting.lint_util import LinterImpl 

23from debputy.lsp.quickfixes import provide_standard_quickfixes_from_diagnostics 

24from debputy.lsp.text_util import on_save_trim_end_of_line_whitespace 

25 

26C = TypeVar("C", bound=Callable) 

27 

28SEMANTIC_TOKENS_LEGEND = SemanticTokensLegend( 

29 token_types=["keyword", "enumMember"], 

30 token_modifiers=[], 

31) 

32SEMANTIC_TOKEN_TYPES_IDS = { 

33 t: idx for idx, t in enumerate(SEMANTIC_TOKENS_LEGEND.token_types) 

34} 

35 

36DIAGNOSTIC_HANDLERS = {} 

37COMPLETER_HANDLERS = {} 

38HOVER_HANDLERS = {} 

39CODE_ACTION_HANDLERS = {} 

40FOLDING_RANGE_HANDLERS = {} 

41SEMANTIC_TOKENS_FULL_HANDLERS = {} 

42WILL_SAVE_WAIT_UNTIL_HANDLERS = {} 

43_ALIAS_OF = {} 

44 

45_STANDARD_HANDLERS = { 45 ↛ exitline 45 didn't jump to the function exit

46 TEXT_DOCUMENT_CODE_ACTION: ( 

47 CODE_ACTION_HANDLERS, 

48 lambda ls, params: provide_standard_quickfixes_from_diagnostics(params), 

49 ), 

50 TEXT_DOCUMENT_WILL_SAVE_WAIT_UNTIL: ( 

51 WILL_SAVE_WAIT_UNTIL_HANDLERS, 

52 on_save_trim_end_of_line_whitespace, 

53 ), 

54} 

55 

56 

57def lint_diagnostics( 

58 file_formats: Union[str, Sequence[str]] 

59) -> Callable[[LinterImpl], LinterImpl]: 

60 

61 def _wrapper(func: C) -> C: 

62 if not inspect.iscoroutinefunction(func): 62 ↛ 76line 62 didn't jump to line 76, because the condition on line 62 was never false

63 

64 async def _lint_wrapper( 

65 ls: "DebputyLanguageServer", 

66 params: Union[ 

67 DidOpenTextDocumentParams, 

68 DidChangeTextDocumentParams, 

69 ], 

70 ) -> Optional[List[Diagnostic]]: 

71 doc = ls.workspace.get_text_document(params.text_document.uri) 

72 lint_state = ls.lint_state(doc) 

73 yield func(lint_state) 

74 

75 else: 

76 raise ValueError("Linters are all non-async at the moment") 

77 

78 for file_format in file_formats: 

79 if file_format in DIAGNOSTIC_HANDLERS: 

80 raise AssertionError( 

81 "There is already a diagnostics handler for " + file_format 

82 ) 

83 DIAGNOSTIC_HANDLERS[file_format] = _lint_wrapper 

84 

85 return func 

86 

87 return _wrapper 

88 

89 

90def lsp_diagnostics(file_formats: Union[str, Sequence[str]]) -> Callable[[C], C]: 

91 

92 def _wrapper(func: C) -> C: 

93 

94 if not inspect.iscoroutinefunction(func): 94 ↛ 106line 94 didn't jump to line 106, because the condition on line 94 was never false

95 

96 async def _linter(*args, **kwargs) -> None: 

97 res = func(*args, **kwargs) 

98 if inspect.isgenerator(res): 

99 for r in res: 

100 yield r 

101 else: 

102 yield res 

103 

104 else: 

105 

106 _linter = func 

107 

108 _register_handler(file_formats, DIAGNOSTIC_HANDLERS, _linter) 

109 

110 return func 

111 

112 return _wrapper 

113 

114 

115def lsp_completer(file_formats: Union[str, Sequence[str]]) -> Callable[[C], C]: 

116 return _registering_wrapper(file_formats, COMPLETER_HANDLERS) 

117 

118 

119def lsp_hover(file_formats: Union[str, Sequence[str]]) -> Callable[[C], C]: 

120 return _registering_wrapper(file_formats, HOVER_HANDLERS) 

121 

122 

123def lsp_folding_ranges(file_formats: Union[str, Sequence[str]]) -> Callable[[C], C]: 

124 return _registering_wrapper(file_formats, FOLDING_RANGE_HANDLERS) 

125 

126 

127def lsp_semantic_tokens_full( 

128 file_formats: Union[str, Sequence[str]] 

129) -> Callable[[C], C]: 

130 return _registering_wrapper(file_formats, SEMANTIC_TOKENS_FULL_HANDLERS) 

131 

132 

133def lsp_standard_handler(file_formats: Union[str, Sequence[str]], topic: str) -> None: 

134 res = _STANDARD_HANDLERS.get(topic) 

135 if res is None: 135 ↛ 136line 135 didn't jump to line 136, because the condition on line 135 was never true

136 raise ValueError(f"No standard handler for {topic}") 

137 

138 table, handler = res 

139 

140 _register_handler(file_formats, table, handler) 

141 

142 

143def _registering_wrapper( 

144 file_formats: Union[str, Sequence[str]], handler_dict: Dict[str, C] 

145) -> Callable[[C], C]: 

146 def _wrapper(func: C) -> C: 

147 _register_handler(file_formats, handler_dict, func) 

148 return func 

149 

150 return _wrapper 

151 

152 

153def _register_handler( 

154 file_formats: Union[str, Sequence[str]], 

155 handler_dict: Dict[str, C], 

156 handler: C, 

157) -> None: 

158 if isinstance(file_formats, str): 158 ↛ 159line 158 didn't jump to line 159, because the condition on line 158 was never true

159 file_formats = [file_formats] 

160 else: 

161 if not file_formats: 161 ↛ 162line 161 didn't jump to line 162, because the condition on line 161 was never true

162 raise ValueError("At least one language ID (file format) must be provided") 

163 main = file_formats[0] 

164 for alias in file_formats[1:]: 

165 if alias not in _ALIAS_OF: 

166 _ALIAS_OF[alias] = main 

167 

168 for file_format in file_formats: 

169 if file_format in handler_dict: 

170 raise AssertionError(f"There is already a handler for {file_format}") 

171 

172 handler_dict[file_format] = handler 

173 

174 

175def ensure_lsp_features_are_loaded() -> None: 

176 # FIXME: This import is needed to force loading of the LSP files. But it only works 

177 # for files with a linter (which currently happens to be all of them, but this is 

178 # a bit fragile). 

179 from debputy.linting.lint_impl import LINTER_FORMATS 

180 

181 assert LINTER_FORMATS 

182 

183 

184def describe_lsp_features() -> None: 

185 

186 ensure_lsp_features_are_loaded() 

187 

188 feature_list = [ 

189 ("diagnostics (lint)", DIAGNOSTIC_HANDLERS), 

190 ("code actions/quickfixes", CODE_ACTION_HANDLERS), 

191 ("completion suggestions", COMPLETER_HANDLERS), 

192 ("hover docs", HOVER_HANDLERS), 

193 ("folding ranges", FOLDING_RANGE_HANDLERS), 

194 ("semantic tokens", SEMANTIC_TOKENS_FULL_HANDLERS), 

195 ("on-save handler", WILL_SAVE_WAIT_UNTIL_HANDLERS), 

196 ] 

197 print("LSP language IDs and their features:") 

198 all_ids = sorted(set(lid for _, t in feature_list for lid in t)) 

199 for lang_id in all_ids: 

200 if lang_id in _ALIAS_OF: 

201 continue 

202 features = [n for n, t in feature_list if lang_id in t] 

203 print(f" * {lang_id}:") 

204 for feature in features: 

205 print(f" - {feature}") 

206 

207 aliases = collections.defaultdict(list) 

208 for lang_id in all_ids: 

209 main_lang = _ALIAS_OF.get(lang_id) 

210 if main_lang is None: 

211 continue 

212 aliases[main_lang].append(lang_id) 

213 

214 print() 

215 print("Aliases:") 

216 for main_id, aliases in aliases.items(): 

217 print(f" * {main_id}: {', '.join(aliases)}")