Coverage for src/debputy/lsp/lsp_debian_debputy_manifest.py: 77%

1from typing import ( 

2 Optional, 

3 List, 

4 Any, 

5 Tuple, 

6 Union, 

7 Iterable, 

8 Sequence, 

9 Literal, 

10 get_args, 

11 get_origin, 

12) 

13 

14from lsprotocol.types import ( 

15 Diagnostic, 

16 TEXT_DOCUMENT_WILL_SAVE_WAIT_UNTIL, 

17 Position, 

18 Range, 

19 DiagnosticSeverity, 

20 HoverParams, 

21 Hover, 

22 MarkupKind, 

23 MarkupContent, 

24 TEXT_DOCUMENT_CODE_ACTION, 

25 CompletionParams, 

26 CompletionList, 

27 CompletionItem, 

28 DiagnosticRelatedInformation, 

29 Location, 

30) 

31 

32from debputy.linting.lint_util import LintState 

33from debputy.lsp.quickfixes import propose_correct_text_quick_fix 

34from debputy.manifest_parser.base_types import DebputyDispatchableType 

35from debputy.plugin.api.feature_set import PluginProvidedFeatureSet 

36from debputy.yaml.compat import ( 

37 Node, 

38 CommentedMap, 

39 LineCol, 

40 CommentedSeq, 

41 CommentedBase, 

42 MarkedYAMLError, 

43 YAMLError, 

44) 

45 

46from debputy.highlevel_manifest import MANIFEST_YAML 

47from debputy.lsp.lsp_features import ( 

48 lint_diagnostics, 

49 lsp_standard_handler, 

50 lsp_hover, 

51 lsp_completer, 

52) 

53from debputy.lsp.text_util import ( 

54 LintCapablePositionCodec, 

55 detect_possible_typo, 

56) 

57from debputy.manifest_parser.declarative_parser import ( 

58 AttributeDescription, 

59 ParserGenerator, 

60 DeclarativeNonMappingInputParser, 

61) 

62from debputy.manifest_parser.declarative_parser import DeclarativeMappingInputParser 

63from debputy.manifest_parser.parser_doc import ( 

64 render_rule, 

65 render_attribute_doc, 

66 doc_args_for_parser_doc, 

67) 

68from debputy.manifest_parser.util import AttributePath 

69from debputy.plugin.api.impl import plugin_metadata_for_debputys_own_plugin 

70from debputy.plugin.api.impl_types import ( 

71 OPARSER_MANIFEST_ROOT, 

72 DeclarativeInputParser, 

73 DispatchingParserBase, 

74 DebputyPluginMetadata, 

75 ListWrappedDeclarativeInputParser, 

76 InPackageContextParser, 

77 DeclarativeValuelessKeywordInputParser, 

78) 

79from debputy.util import _info, _warn 

80 

81 

82try: 

83 from pygls.server import LanguageServer 

84 from debputy.lsp.debputy_ls import DebputyLanguageServer 

85except ImportError: 

86 pass 

87 

88 

89_LANGUAGE_IDS = [ 

90 "debian/debputy.manifest", 

91 "debputy.manifest", 

92 # LSP's official language ID for YAML files 

93 "yaml", 

94] 

95 

96 

97lsp_standard_handler(_LANGUAGE_IDS, TEXT_DOCUMENT_CODE_ACTION) 

98lsp_standard_handler(_LANGUAGE_IDS, TEXT_DOCUMENT_WILL_SAVE_WAIT_UNTIL) 

99 

100 

101def is_valid_file(path: str) -> bool: 

102 # For debian/debputy.manifest, the language ID is often set to makefile meaning we get random 

103 # "non-debian/debputy.manifest" YAML files here. Skip those. 

104 return path.endswith("debian/debputy.manifest") 

105 

106 

107def _word_range_at_position( 

108 lines: List[str], 

109 line_no: int, 

110 char_offset: int, 

111) -> Range: 

112 line = lines[line_no] 

113 line_len = len(line) 

114 start_idx = char_offset 

115 end_idx = char_offset 

116 while end_idx + 1 < line_len and not line[end_idx + 1].isspace(): 

117 end_idx += 1 

118 

119 while start_idx - 1 >= 0 and not line[start_idx - 1].isspace(): 

120 start_idx -= 1 

121 

122 return Range( 

123 Position(line_no, start_idx), 

124 Position(line_no, end_idx), 

125 ) 

126 

127 

128@lint_diagnostics(_LANGUAGE_IDS) 

129def _lint_debian_debputy_manifest( 

130 lint_state: LintState, 

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

132 lines = lint_state.lines 

133 position_codec = lint_state.position_codec 

134 doc_reference = lint_state.doc_uri 

135 path = lint_state.path 

136 if not is_valid_file(path): 136 ↛ 137line 136 didn't jump to line 137, because the condition on line 136 was never true

137 return None 

138 diagnostics = [] 

139 try: 

140 content = MANIFEST_YAML.load("".join(lines)) 

141 except MarkedYAMLError as e: 

142 if e.context_mark: 

143 line = e.context_mark.line 

144 column = e.context_mark.column + 1 

145 else: 

146 line = e.problem_mark.line 

147 column = e.problem_mark.column + 1 

148 error_range = position_codec.range_to_client_units( 

149 lines, 

150 _word_range_at_position( 

151 lines, 

152 line, 

153 column, 

154 ), 

155 ) 

156 diagnostics.append( 

157 Diagnostic( 

158 error_range, 

159 f"YAML parse error: {e}", 

160 DiagnosticSeverity.Error, 

161 ), 

162 ) 

163 except YAMLError as e: 

164 error_range = position_codec.range_to_client_units( 

165 lines, 

166 Range( 

167 Position(0, 0), 

168 Position(0, len(lines[0])), 

169 ), 

170 ) 

171 diagnostics.append( 

172 Diagnostic( 

173 error_range, 

174 f"Unknown YAML parse error: {e} [{e!r}]", 

175 DiagnosticSeverity.Error, 

176 ), 

177 ) 

178 else: 

179 feature_set = lint_state.plugin_feature_set 

180 pg = feature_set.manifest_parser_generator 

181 root_parser = pg.dispatchable_object_parsers[OPARSER_MANIFEST_ROOT] 

182 diagnostics.extend( 

183 _lint_content( 

184 doc_reference, 

185 pg, 

186 root_parser, 

187 content, 

188 lines, 

189 position_codec, 

190 ) 

191 ) 

192 return diagnostics 

193 

194 

195def _unknown_key( 

196 key: str, 

197 expected_keys: Iterable[str], 

198 line: int, 

199 col: int, 

200 lines: List[str], 

201 position_codec: LintCapablePositionCodec, 

202) -> Tuple["Diagnostic", Optional[str]]: 

203 key_range = position_codec.range_to_client_units( 

204 lines, 

205 Range( 

206 Position( 

207 line, 

208 col, 

209 ), 

210 Position( 

211 line, 

212 col + len(key), 

213 ), 

214 ), 

215 ) 

216 

217 candidates = detect_possible_typo(key, expected_keys) 

218 extra = "" 

219 corrected_key = None 

220 if candidates: 

221 extra = f' It looks like a typo of "{candidates[0]}".' 

222 # TODO: We should be able to tell that `install-doc` and `install-docs` are the same. 

223 # That would enable this to work in more cases. 

224 corrected_key = candidates[0] if len(candidates) == 1 else None 

225 

226 diagnostic = Diagnostic( 

227 key_range, 

228 f'Unknown or unsupported key "{key}".{extra}', 

229 DiagnosticSeverity.Error, 

230 source="debputy", 

231 data=[propose_correct_text_quick_fix(n) for n in candidates], 

232 ) 

233 return diagnostic, corrected_key 

234 

235 

236def _conflicting_key( 

237 uri: str, 

238 key_a: str, 

239 key_b: str, 

240 key_a_line: int, 

241 key_a_col: int, 

242 key_b_line: int, 

243 key_b_col: int, 

244 lines: List[str], 

245 position_codec: LintCapablePositionCodec, 

246) -> Iterable["Diagnostic"]: 

247 key_a_range = position_codec.range_to_client_units( 

248 lines, 

249 Range( 

250 Position( 

251 key_a_line, 

252 key_a_col, 

253 ), 

254 Position( 

255 key_a_line, 

256 key_a_col + len(key_a), 

257 ), 

258 ), 

259 ) 

260 key_b_range = position_codec.range_to_client_units( 

261 lines, 

262 Range( 

263 Position( 

264 key_b_line, 

265 key_b_col, 

266 ), 

267 Position( 

268 key_b_line, 

269 key_b_col + len(key_b), 

270 ), 

271 ), 

272 ) 

273 yield Diagnostic( 

274 key_a_range, 

275 f'The "{key_a}" cannot be used with "{key_b}".', 

276 DiagnosticSeverity.Error, 

277 source="debputy", 

278 related_information=[ 

279 DiagnosticRelatedInformation( 

280 location=Location( 

281 uri, 

282 key_b_range, 

283 ), 

284 message=f'The attribute "{key_b}" is used here.', 

285 ) 

286 ], 

287 ) 

288 

289 yield Diagnostic( 

290 key_b_range, 

291 f'The "{key_b}" cannot be used with "{key_a}".', 

292 DiagnosticSeverity.Error, 

293 source="debputy", 

294 related_information=[ 

295 DiagnosticRelatedInformation( 

296 location=Location( 

297 uri, 

298 key_a_range, 

299 ), 

300 message=f'The attribute "{key_a}" is used here.', 

301 ) 

302 ], 

303 ) 

304 

305 

306def _lint_attr_value( 

307 uri: str, 

308 attr: AttributeDescription, 

309 pg: ParserGenerator, 

310 value: Any, 

311 lines: List[str], 

312 position_codec: LintCapablePositionCodec, 

313) -> Iterable["Diagnostic"]: 

314 attr_type = attr.attribute_type 

315 orig = get_origin(attr_type) 

316 valid_values: Sequence[Any] = tuple() 

317 if orig == Literal: 317 ↛ 318line 317 didn't jump to line 318, because the condition on line 317 was never true

318 valid_values = get_args(attr.attribute_type) 

319 elif orig == bool or attr.attribute_type == bool: 319 ↛ 320line 319 didn't jump to line 320, because the condition on line 319 was never true

320 valid_values = ("true", "false") 

321 elif isinstance(attr_type, type) and issubclass(attr_type, DebputyDispatchableType): 

322 parser = pg.dispatch_parser_table_for(attr_type) 

323 yield from _lint_content( 

324 uri, 

325 pg, 

326 parser, 

327 value, 

328 lines, 

329 position_codec, 

330 ) 

331 return 

332 

333 if value in valid_values: 333 ↛ 334line 333 didn't jump to line 334, because the condition on line 333 was never true

334 return 

335 # TODO: Emit diagnostic for broken values 

336 return 

337 

338 

339def _lint_declarative_mapping_input_parser( 

340 uri: str, 

341 pg: ParserGenerator, 

342 parser: DeclarativeMappingInputParser, 

343 content: Any, 

344 lines: List[str], 

345 position_codec: LintCapablePositionCodec, 

346) -> Iterable["Diagnostic"]: 

347 if not isinstance(content, CommentedMap): 

348 return 

349 lc = content.lc 

350 for key, value in content.items(): 

351 attr = parser.manifest_attributes.get(key) 

352 line, col = lc.key(key) 

353 if attr is None: 

354 diag, corrected_key = _unknown_key( 

355 key, 

356 parser.manifest_attributes, 

357 line, 

358 col, 

359 lines, 

360 position_codec, 

361 ) 

362 yield diag 

363 if corrected_key: 363 ↛ 364line 363 didn't jump to line 364, because the condition on line 363 was never true

364 key = corrected_key 

365 attr = parser.manifest_attributes.get(corrected_key) 

366 if attr is None: 

367 continue 

368 

369 yield from _lint_attr_value( 

370 uri, 

371 attr, 

372 pg, 

373 value, 

374 lines, 

375 position_codec, 

376 ) 

377 

378 for forbidden_key in attr.conflicting_attributes: 

379 if forbidden_key in content: 

380 con_line, con_col = lc.key(forbidden_key) 

381 yield from _conflicting_key( 

382 uri, 

383 key, 

384 forbidden_key, 

385 line, 

386 col, 

387 con_line, 

388 con_col, 

389 lines, 

390 position_codec, 

391 ) 

392 for mx in parser.mutually_exclusive_attributes: 

393 matches = content.keys() & mx 

394 if len(matches) < 2: 

395 continue 

396 key, *others = list(matches) 

397 line, col = lc.key(key) 

398 for other in others: 

399 con_line, con_col = lc.key(other) 

400 yield from _conflicting_key( 

401 uri, 

402 key, 

403 other, 

404 line, 

405 col, 

406 con_line, 

407 con_col, 

408 lines, 

409 position_codec, 

410 ) 

411 

412 

413def _lint_content( 

414 uri: str, 

415 pg: ParserGenerator, 

416 parser: DeclarativeInputParser[Any], 

417 content: Any, 

418 lines: List[str], 

419 position_codec: LintCapablePositionCodec, 

420) -> Iterable["Diagnostic"]: 

421 if isinstance(parser, DispatchingParserBase): 

422 if not isinstance(content, CommentedMap): 

423 return 

424 lc = content.lc 

425 for key, value in content.items(): 

426 is_known = parser.is_known_keyword(key) 

427 if not is_known: 

428 line, col = lc.key(key) 

429 diag, corrected_key = _unknown_key( 

430 key, 

431 parser.registered_keywords(), 

432 line, 

433 col, 

434 lines, 

435 position_codec, 

436 ) 

437 yield diag 

438 if corrected_key is not None: 

439 key = corrected_key 

440 is_known = True 

441 

442 if is_known: 

443 subparser = parser.parser_for(key) 

444 assert subparser is not None 

445 yield from _lint_content( 

446 uri, 

447 pg, 

448 subparser.parser, 

449 value, 

450 lines, 

451 position_codec, 

452 ) 

453 elif isinstance(parser, ListWrappedDeclarativeInputParser): 

454 if not isinstance(content, CommentedSeq): 454 ↛ 455line 454 didn't jump to line 455, because the condition on line 454 was never true

455 return 

456 subparser = parser.delegate 

457 for value in content: 

458 yield from _lint_content(uri, pg, subparser, value, lines, position_codec) 

459 elif isinstance(parser, InPackageContextParser): 

460 if not isinstance(content, CommentedMap): 460 ↛ 461line 460 didn't jump to line 461, because the condition on line 460 was never true

461 return 

462 for v in content.values(): 

463 yield from _lint_content(uri, pg, parser.delegate, v, lines, position_codec) 

464 elif isinstance(parser, DeclarativeMappingInputParser): 

465 yield from _lint_declarative_mapping_input_parser( 

466 uri, 

467 pg, 

468 parser, 

469 content, 

470 lines, 

471 position_codec, 

472 ) 

473 

474 

475def is_at(position: Position, lc_pos: Tuple[int, int]) -> bool: 

476 return position.line == lc_pos[0] and position.character == lc_pos[1] 

477 

478 

479def is_before(position: Position, lc_pos: Tuple[int, int]) -> bool: 

480 line, column = lc_pos 

481 if position.line < line: 

482 return True 

483 if position.line == line and position.character < column: 

484 return True 

485 return False 

486 

487 

488def is_after(position: Position, lc_pos: Tuple[int, int]) -> bool: 

489 line, column = lc_pos 

490 if position.line > line: 

491 return True 

492 if position.line == line and position.character > column: 

493 return True 

494 return False 

495 

496 

497def _trace_cursor( 

498 content: Any, 

499 attribute_path: AttributePath, 

500 server_position: Position, 

501) -> Optional[Tuple[bool, AttributePath, Any, Any]]: 

502 matched_key: Optional[Union[str, int]] = None 

503 matched: Optional[Node] = None 

504 matched_was_key: bool = False 

505 

506 if isinstance(content, CommentedMap): 

507 dict_lc: LineCol = content.lc 

508 for k, v in content.items(): 

509 k_lc = dict_lc.key(k) 

510 if is_before(server_position, k_lc): 510 ↛ 511line 510 didn't jump to line 511, because the condition on line 510 was never true

511 break 

512 v_lc = dict_lc.value(k) 

513 if is_before(server_position, v_lc): 

514 # TODO: Handle ":" and "whitespace" 

515 matched = k 

516 matched_key = k 

517 matched_was_key = True 

518 break 

519 matched = v 

520 matched_key = k 

521 elif isinstance(content, CommentedSeq): 521 ↛ 530line 521 didn't jump to line 530, because the condition on line 521 was never false

522 list_lc: LineCol = content.lc 

523 for idx, value in enumerate(content): 

524 i_lc = list_lc.item(idx) 

525 if is_before(server_position, i_lc): 525 ↛ 526line 525 didn't jump to line 526, because the condition on line 525 was never true

526 break 

527 matched_key = idx 

528 matched = value 

529 

530 if matched is not None: 530 ↛ 536line 530 didn't jump to line 536, because the condition on line 530 was never false

531 assert matched_key is not None 

532 sub_path = attribute_path[matched_key] 

533 if not matched_was_key and isinstance(matched, CommentedBase): 

534 return _trace_cursor(matched, sub_path, server_position) 

535 return matched_was_key, sub_path, matched, content 

536 return None 

537 

538 

539_COMPLETION_HINT_KEY = "___COMPLETE:" 

540_COMPLETION_HINT_VALUE = "___COMPLETE" 

541 

542 

543def resolve_keyword( 

544 current_parser: Union[DeclarativeInputParser[Any], DispatchingParserBase], 

545 current_plugin: DebputyPluginMetadata, 

546 segments: List[Union[str, int]], 

547 segment_idx: int, 

548 parser_generator: ParserGenerator, 

549 *, 

550 is_completion_attempt: bool = False, 

551) -> Optional[ 

552 Tuple[ 

553 Union[DeclarativeInputParser[Any], DispatchingParserBase], 

554 DebputyPluginMetadata, 

555 int, 

556 ] 

557]: 

558 if segment_idx >= len(segments): 

559 return current_parser, current_plugin, segment_idx 

560 current_segment = segments[segment_idx] 

561 if isinstance(current_parser, ListWrappedDeclarativeInputParser): 

562 if isinstance(current_segment, int): 562 ↛ 569line 562 didn't jump to line 569, because the condition on line 562 was never false

563 current_parser = current_parser.delegate 

564 segment_idx += 1 

565 if segment_idx >= len(segments): 565 ↛ 566line 565 didn't jump to line 566, because the condition on line 565 was never true

566 return current_parser, current_plugin, segment_idx 

567 current_segment = segments[segment_idx] 

568 

569 if not isinstance(current_segment, str): 569 ↛ 570line 569 didn't jump to line 570, because the condition on line 569 was never true

570 return None 

571 

572 if is_completion_attempt and current_segment.endswith( 

573 (_COMPLETION_HINT_KEY, _COMPLETION_HINT_VALUE) 

574 ): 

575 return current_parser, current_plugin, segment_idx 

576 

577 if isinstance(current_parser, InPackageContextParser): 

578 return resolve_keyword( 

579 current_parser.delegate, 

580 current_plugin, 

581 segments, 

582 segment_idx + 1, 

583 parser_generator, 

584 is_completion_attempt=is_completion_attempt, 

585 ) 

586 elif isinstance(current_parser, DispatchingParserBase): 

587 if not current_parser.is_known_keyword(current_segment): 587 ↛ 588line 587 didn't jump to line 588, because the condition on line 587 was never true

588 if is_completion_attempt: 

589 return current_parser, current_plugin, segment_idx 

590 return None 

591 subparser = current_parser.parser_for(current_segment) 

592 segment_idx += 1 

593 if segment_idx < len(segments): 

594 return resolve_keyword( 

595 subparser.parser, 

596 subparser.plugin_metadata, 

597 segments, 

598 segment_idx, 

599 parser_generator, 

600 is_completion_attempt=is_completion_attempt, 

601 ) 

602 return subparser.parser, subparser.plugin_metadata, segment_idx 

603 elif isinstance(current_parser, DeclarativeMappingInputParser): 603 ↛ 625line 603 didn't jump to line 625, because the condition on line 603 was never false

604 attr = current_parser.manifest_attributes.get(current_segment) 

605 attr_type = attr.attribute_type if attr is not None else None 

606 if ( 

607 attr_type is not None 

608 and isinstance(attr_type, type) 

609 and issubclass(attr_type, DebputyDispatchableType) 

610 ): 

611 subparser = parser_generator.dispatch_parser_table_for(attr_type) 

612 if subparser is not None and ( 

613 is_completion_attempt or segment_idx + 1 < len(segments) 

614 ): 

615 return resolve_keyword( 

616 subparser, 

617 current_plugin, 

618 segments, 

619 segment_idx + 1, 

620 parser_generator, 

621 is_completion_attempt=is_completion_attempt, 

622 ) 

623 return current_parser, current_plugin, segment_idx 

624 else: 

625 _info(f"Unknown parser: {current_parser.__class__}") 

626 return None 

627 

628 

629def _render_param_doc( 

630 rule_name: str, 

631 declarative_parser: DeclarativeMappingInputParser, 

632 plugin_metadata: DebputyPluginMetadata, 

633 attribute: str, 

634) -> Optional[str]: 

635 attr = declarative_parser.source_attributes.get(attribute) 

636 if attr is None: 636 ↛ 637line 636 didn't jump to line 637, because the condition on line 636 was never true

637 return None 

638 

639 doc_args, parser_doc = doc_args_for_parser_doc( 

640 rule_name, 

641 declarative_parser, 

642 plugin_metadata, 

643 ) 

644 rendered_docs = render_attribute_doc( 

645 declarative_parser, 

646 declarative_parser.source_attributes, 

647 declarative_parser.input_time_required_parameters, 

648 declarative_parser.at_least_one_of, 

649 parser_doc, 

650 doc_args, 

651 is_interactive=True, 

652 rule_name=rule_name, 

653 ) 

654 

655 for attributes, rendered_doc in rendered_docs: 655 ↛ 664line 655 didn't jump to line 664, because the loop on line 655 didn't complete

656 if attribute in attributes: 

657 full_doc = [ 

658 f"# Attribute `{attribute}`", 

659 "", 

660 ] 

661 full_doc.extend(rendered_doc) 

662 

663 return "\n".join(full_doc) 

664 return None 

665 

666 

667DEBPUTY_PLUGIN_METADATA = plugin_metadata_for_debputys_own_plugin() 

668 

669 

670def _guess_rule_name(segments: List[Union[str, int]], idx: int) -> str: 

671 orig_idx = idx 

672 idx -= 1 

673 while idx >= 0: 673 ↛ 678line 673 didn't jump to line 678, because the condition on line 673 was never false

674 segment = segments[idx] 

675 if isinstance(segment, str): 

676 return segment 

677 idx -= 1 

678 _warn(f"Unable to derive rule name from {segments} [{orig_idx}]") 

679 return "<Bug: unknown rule name>" 

680 

681 

682def _escape(v: str) -> str: 

683 return '"' + v.replace("\n", "\\n") + '"' 

684 

685 

686def _insert_snippet(lines: List[str], server_position: Position) -> bool: 

687 _info(f"Complete at {server_position}") 

688 line_no = server_position.line 

689 line = lines[line_no] 

690 pos_rhs = line[server_position.character :] 

691 if pos_rhs and not pos_rhs.isspace(): 691 ↛ 692line 691 didn't jump to line 692, because the condition on line 691 was never true

692 _info(f"No insertion: {_escape(line[server_position.character:])}") 

693 return False 

694 lhs_ws = line[: server_position.character] 

695 lhs = lhs_ws.strip() 

696 if lhs.endswith(":"): 

697 _info("Insertion of value (key seen)") 

698 new_line = line[: server_position.character] + _COMPLETION_HINT_VALUE 

699 elif lhs.startswith("-"): 

700 _info("Insertion of key or value (list item)") 

701 # Respect the provided indentation 

702 snippet = _COMPLETION_HINT_KEY if ":" not in lhs else _COMPLETION_HINT_VALUE 

703 new_line = line[: server_position.character] + snippet 

704 elif not lhs or (lhs_ws and not lhs_ws[0].isspace()): 

705 _info(f"Insertion of key or value: {_escape(line[server_position.character:])}") 

706 # Respect the provided indentation 

707 snippet = _COMPLETION_HINT_KEY if ":" not in lhs else _COMPLETION_HINT_VALUE 

708 new_line = line[: server_position.character] + snippet 

709 elif lhs.isalpha() and ":" not in lhs: 

710 _info(f"Expanding value to a key: {_escape(line[server_position.character:])}") 

711 # Respect the provided indentation 

712 new_line = line[: server_position.character] + _COMPLETION_HINT_KEY 

713 else: 

714 c = line[server_position.character] 

715 _info(f"Not touching line: {_escape(line)} -- {_escape(c)}") 

716 return False 

717 _info(f'Evaluating complete on synthetic line: "{new_line}"') 

718 lines[line_no] = new_line 

719 return True 

720 

721 

722@lsp_completer(_LANGUAGE_IDS) 

723def debputy_manifest_completer( 

724 ls: "DebputyLanguageServer", 

725 params: CompletionParams, 

726) -> Optional[Union[CompletionList, Sequence[CompletionItem]]]: 

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

728 if not is_valid_file(doc.path): 728 ↛ 729line 728 didn't jump to line 729, because the condition on line 728 was never true

729 return None 

730 lines = doc.lines 

731 server_position = doc.position_codec.position_from_client_units( 

732 lines, params.position 

733 ) 

734 attribute_root_path = AttributePath.root_path() 

735 added_key = _insert_snippet(lines, server_position) 

736 attempts = 1 if added_key else 2 

737 content = None 

738 

739 while attempts > 0: 739 ↛ 767line 739 didn't jump to line 767, because the condition on line 739 was never false

740 attempts -= 1 

741 try: 

742 content = MANIFEST_YAML.load("".join(lines)) 

743 break 

744 except MarkedYAMLError as e: 

745 context_line = ( 

746 e.context_mark.line if e.context_mark else e.problem_mark.line 

747 ) 

748 if ( 

749 e.problem_mark.line != server_position.line 

750 and context_line != server_position.line 

751 ): 

752 l_data = ( 

753 lines[e.problem_mark.line].rstrip() 

754 if e.problem_mark.line < len(lines) 

755 else "N/A (OOB)" 

756 ) 

757 

758 _info(f"Parse error on line: {e.problem_mark.line}: {l_data}") 

759 return None 

760 

761 if attempts > 0: 

762 # Try to make it a key and see if that fixes the problem 

763 new_line = lines[server_position.line].rstrip() + _COMPLETION_HINT_KEY 

764 lines[server_position.line] = new_line 

765 except YAMLError: 

766 break 

767 if content is None: 767 ↛ 768line 767 didn't jump to line 768, because the condition on line 767 was never true

768 context = lines[server_position.line].replace("\n", "\\n") 

769 _info(f"Completion failed: parse error: Line in question: {context}") 

770 return None 

771 m = _trace_cursor(content, attribute_root_path, server_position) 

772 

773 if m is None: 773 ↛ 774line 773 didn't jump to line 774, because the condition on line 773 was never true

774 _info("No match") 

775 return None 

776 matched_key, attr_path, matched, parent = m 

777 _info(f"Matched path: {matched} (path: {attr_path.path}) [{matched_key=}]") 

778 feature_set = ls.plugin_feature_set 

779 root_parser = feature_set.manifest_parser_generator.dispatchable_object_parsers[ 

780 OPARSER_MANIFEST_ROOT 

781 ] 

782 segments = list(attr_path.path_segments()) 

783 km = resolve_keyword( 

784 root_parser, 

785 DEBPUTY_PLUGIN_METADATA, 

786 segments, 

787 0, 

788 feature_set.manifest_parser_generator, 

789 is_completion_attempt=True, 

790 ) 

791 if km is None: 791 ↛ 792line 791 didn't jump to line 792, because the condition on line 791 was never true

792 return None 

793 parser, _, at_depth_idx = km 

794 _info(f"Match leaf parser {at_depth_idx} -- {parser.__class__}") 

795 items = [] 

796 if at_depth_idx + 1 >= len(segments): 796 ↛ 859line 796 didn't jump to line 859, because the condition on line 796 was never false

797 if isinstance(parser, DispatchingParserBase): 

798 if matched_key: 

799 items = [ 

800 CompletionItem(f"{k}:") 

801 for k in parser.registered_keywords() 

802 if k not in parent 

803 and not isinstance( 

804 parser.parser_for(k).parser, 

805 DeclarativeValuelessKeywordInputParser, 

806 ) 

807 ] 

808 else: 

809 items = [ 

810 CompletionItem(k) 

811 for k in parser.registered_keywords() 

812 if k not in parent 

813 and isinstance( 

814 parser.parser_for(k).parser, 

815 DeclarativeValuelessKeywordInputParser, 

816 ) 

817 ] 

818 elif isinstance(parser, InPackageContextParser): 818 ↛ 820line 818 didn't jump to line 820, because the condition on line 818 was never true

819 # doc = ls.workspace.get_text_document(params.text_document.uri) 

820 _info(f"TODO: Match package - {parent} -- {matched} -- {matched_key=}") 

821 elif isinstance(parser, DeclarativeMappingInputParser): 

822 if matched_key: 

823 _info("Match attributes") 

824 locked = set(parent) 

825 for mx in parser.mutually_exclusive_attributes: 

826 if not mx.isdisjoint(parent.keys()): 

827 locked.update(mx) 

828 for attr_name, attr in parser.manifest_attributes.items(): 

829 if not attr.conflicting_attributes.isdisjoint(parent.keys()): 

830 locked.add(attr_name) 

831 break 

832 items = [ 

833 CompletionItem(f"{k}:") 

834 for k in parser.manifest_attributes 

835 if k not in locked 

836 ] 

837 else: 

838 # Value 

839 key = segments[at_depth_idx] if len(segments) > at_depth_idx else None 

840 attr = parser.manifest_attributes.get(key) 

841 if attr is not None: 841 ↛ 849line 841 didn't jump to line 849, because the condition on line 841 was never false

842 _info(f"Expand value / key: {key} -- {attr.attribute_type}") 

843 items = _completion_from_attr( 

844 attr, 

845 feature_set.manifest_parser_generator, 

846 matched, 

847 ) 

848 else: 

849 _info( 

850 f"Expand value / key: {key} -- !! {list(parser.manifest_attributes)}" 

851 ) 

852 elif isinstance(parser, DeclarativeNonMappingInputParser): 852 ↛ 859line 852 didn't jump to line 859, because the condition on line 852 was never false

853 attr = parser.alt_form_parser 

854 items = _completion_from_attr( 

855 attr, 

856 feature_set.manifest_parser_generator, 

857 matched, 

858 ) 

859 return items 

860 

861 

862def _completion_from_attr( 

863 attr: AttributeDescription, 

864 pg: ParserGenerator, 

865 matched: Any, 

866) -> Optional[Union[CompletionList, Sequence[CompletionItem]]]: 

867 orig = get_origin(attr.attribute_type) 

868 valid_values: Sequence[Any] = tuple() 

869 if orig == Literal: