Coverage for src/debputy/linting/lint_util.py: 54%

1import collections 

2import contextlib 

3import dataclasses 

4import datetime 

5import os 

6import time 

7from collections import defaultdict, Counter 

8from enum import IntEnum 

9from functools import lru_cache 

10from typing import ( 

11 Optional, 

12 TYPE_CHECKING, 

13 cast, 

14 Self, 

15 TypeVar, 

16) 

17from collections.abc import ( 

18 Callable, 

19 Mapping, 

20 Sequence, 

21 Iterable, 

22 Awaitable, 

23 AsyncIterable, 

24) 

25 

26 

27import debputy.l10n as l10n 

28from debputy.commands.debputy_cmd.output import IOBasedOutputStyling 

29from debputy.dh.dh_assistant import ( 

30 extract_dh_addons_from_control, 

31 DhSequencerData, 

32 parse_drules_for_addons, 

33) 

34from debputy.exceptions import PureVirtualPathError, DebputyRuntimeError 

35from debputy.filesystem_scan import VirtualPathBase 

36from debputy.integration_detection import determine_debputy_integration_mode 

37from debputy.l10n import Translations 

38from debputy.lsp.config.debputy_config import DebputyConfig 

39from debputy.lsp.diagnostics import ( 

40 LintSeverity, 

41 LINT_SEVERITY2LSP_SEVERITY, 

42 DiagnosticData, 

43 NATIVELY_LSP_SUPPORTED_SEVERITIES, 

44) 

45from debputy.lsp.spellchecking import Spellchecker, default_spellchecker 

46from debian._deb822_repro import Deb822FileElement, parse_deb822_file 

47from debputy.packages import SourcePackage, BinaryPackage 

48from debputy.plugin.api.feature_set import PluginProvidedFeatureSet 

49from debputy.plugin.api.spec import DebputyIntegrationMode 

50from debputy.util import _warn, T 

51from debian._deb822_repro.locatable import ( 

52 Range as TERange, 

53 Position as TEPosition, 

54 Locatable, 

55 START_POSITION, 

56) 

57 

58if TYPE_CHECKING: 

59 import lsprotocol.types as types 

60 from debputy.lsp.text_util import LintCapablePositionCodec 

61 from debputy.lsp.maint_prefs import ( 

62 MaintainerPreferenceTable, 

63 EffectiveFormattingPreference, 

64 ) 

65 

66else: 

67 import debputy.lsprotocol.types as types 

68 

69 

70L = TypeVar("L", bound=Locatable) 

71 

72 

73AsyncLinterImpl = Callable[["LintState"], Awaitable[None]] 

74FormatterImpl = Callable[["LintState"], Optional[Sequence[types.TextEdit]]] 

75 

76 

77class AbortTaskError(DebputyRuntimeError): 

78 pass 

79 

80 

81# If you add a new one to this set, remember to mention it in the docs of `LintState.emit_diagnostic` 

82DIAG_SOURCE_WITHOUT_SECTIONS: frozenset[str] = frozenset( 

83 { 

84 "debputy", 

85 "dpkg", 

86 } 

87) 

88 

89# If you add a new one to this set, remember to mention it in the docs of `LintState.emit_diagnostic` 

90DIAG_SOURCE_WITH_SECTIONS: frozenset[str] = frozenset( 

91 { 

92 "Policy", 

93 "DevRef", 

94 } 

95) 

96 

97 

98def te_position_to_lsp(te_position: "TEPosition") -> types.Position: 

99 return types.Position( 

100 te_position.line_position, 

101 te_position.cursor_position, 

102 ) 

103 

104 

105def te_range_to_lsp(te_range: "TERange") -> types.Range: 

106 return types.Range( 

107 te_position_to_lsp(te_range.start_pos), 

108 te_position_to_lsp(te_range.end_pos), 

109 ) 

110 

111 

112def with_range_in_continuous_parts( 

113 iterable: Iterable["L"], 

114 *, 

115 start_relative_to: "TEPosition" = START_POSITION, 

116) -> Iterable[tuple["TERange", "L"]]: 

117 current_pos = start_relative_to 

118 for part in iterable: 

119 part_range = part.size().relative_to(current_pos) 

120 yield part_range, part 

121 current_pos = part_range.end_pos 

122 

123 

124@lru_cache 

125def _check_diagnostic_source(source: str) -> None: 

126 if source in DIAG_SOURCE_WITHOUT_SECTIONS: 

127 return 

128 parts = source.split(" ") 

129 s = parts[0] 

130 if s not in DIAG_SOURCE_WITH_SECTIONS: 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true

131 raise ValueError( 

132 f'Unknown diagnostic source: "{source}". If you are adding a new source, update lint_util.py' 

133 ) 

134 if len(parts) != 2: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 raise ValueError( 

136 f'The diagnostic source "{source}" should have exactly one section associated with it.' 

137 ) 

138 

139 

140@dataclasses.dataclass(slots=True) 

141class DebputyMetadata: 

142 debputy_integration_mode: DebputyIntegrationMode | None 

143 

144 @classmethod 

145 def from_data( 

146 cls, 

147 source_fields: Mapping[str, str], 

148 dh_sequencer_data: DhSequencerData, 

149 ) -> Self: 

150 integration_mode = determine_debputy_integration_mode( 

151 source_fields, 

152 dh_sequencer_data.sequences, 

153 ) 

154 return cls(integration_mode) 

155 

156 

157@dataclasses.dataclass(slots=True, frozen=True) 

158class RelatedDiagnosticInformation: 

159 text_range: "TERange" 

160 message: str 

161 doc_uri: str 

162 

163 def to_lsp(self, lint_state: "LintState") -> types.DiagnosticRelatedInformation: 

164 return types.DiagnosticRelatedInformation( 

165 types.Location( 

166 self.doc_uri, 

167 lint_state.position_codec.range_to_client_units( 

168 lint_state.lines, 

169 te_range_to_lsp(self.text_range), 

170 ), 

171 ), 

172 self.message, 

173 ) 

174 

175 

176@dataclasses.dataclass(frozen=True, slots=True) 

177class WorkspaceTextEditSupport: 

178 supports_document_changes: bool = False 

179 supported_resource_operation_edit_kinds: Sequence[types.ResourceOperationKind] = ( 

180 dataclasses.field(default_factory=list) 

181 ) 

182 

183 @property 

184 def supports_versioned_text_edits(self) -> bool: 

185 return self.supports_document_changes or bool( 

186 self.supported_resource_operation_edit_kinds 

187 ) 

188 

189 

190class LintState: 

191 

192 @property 

193 def plugin_feature_set(self) -> PluginProvidedFeatureSet: 

194 """The plugin features known to the running instance of `debputy` 

195 

196 This is mostly only relevant when working with `debputy.manifest` 

197 """ 

198 raise NotImplementedError 

199 

200 @property 

201 def doc_uri(self) -> str: 

202 """The URI for the document being scanned. 

203 

204 This can be useful for providing related location ranges. 

205 """ 

206 raise NotImplementedError 

207 

208 @property 

209 def doc_version(self) -> int | None: 

210 raise NotImplementedError 

211 

212 @property 

213 def source_root(self) -> VirtualPathBase | None: 

214 """The path to the unpacked source root directory if available 

215 

216 This is the directory that would contain the `debian/` directory. Note, if you need the 

217 `debian/` directory, please use `debian_dir` instead. There may be cases where the source 

218 root is unavailable but the `debian/` directory is not. 

219 """ 

220 raise NotImplementedError 

221 

222 @property 

223 def debian_dir(self) -> VirtualPathBase | None: 

224 """The path to the `debian/` directory if available""" 

225 raise NotImplementedError 

226 

227 @property 

228 def path(self) -> str: 

229 """The filename or path of the file being scanned. 

230 

231 Note this path may or may not be accessible to the running `debputy` instance. Nor is it guaranteed 

232 that the file on the file system (even if accessible) has correct contents. When doing diagnostics 

233 for an editor, the editor often requests diagnostics for unsaved changes. 

234 """ 

235 raise NotImplementedError 

236 

237 @property 

238 def content(self) -> str: 

239 """The full contents of the file being checked""" 

240 raise NotImplementedError 

241 

242 @property 

243 def lines(self) -> list[str]: 

244 # FIXME: Replace with `Sequence[str]` if possible 

245 """The contents of the file being checked as a list of lines 

246 

247 Do **not** change the contents of this list as it may be cached. 

248 """ 

249 raise NotImplementedError 

250 

251 @property 

252 def position_codec(self) -> "LintCapablePositionCodec": 

253 raise NotImplementedError 

254 

255 @property 

256 def parsed_deb822_file_content(self) -> Deb822FileElement | None: 

257 """The contents of the file being checked as a parsed deb822 file 

258 

259 This can sometimes use a cached version of the parsed file and is therefore preferable to 

260 parsing the file manually from `content` or `lines`. 

261 

262 Do **not** change the contents of this as it may be cached. 

263 """ 

264 raise NotImplementedError 

265 

266 @property 

267 def source_package(self) -> SourcePackage | None: 

268 """The source package (source stanza of `debian/control`). 

269 

270 Will be `None` if the `debian/control` file cannot be parsed as a deb822 file, or if the 

271 source stanza is not available. 

272 """ 

273 raise NotImplementedError 

274 

275 @property 

276 def binary_packages(self) -> Mapping[str, BinaryPackage] | None: 

277 """The binary packages (the Package stanzas of `debian/control`). 

278 

279 Will be `None` if the `debian/control` file cannot be parsed, or if no Package stanzas are 

280 available. 

281 """ 

282 raise NotImplementedError 

283 

284 @property 

285 def maint_preference_table(self) -> "MaintainerPreferenceTable": 

286 # TODO: Visible only for tests. 

287 raise NotImplementedError 

288 

289 @property 

290 def effective_preference(self) -> Optional["EffectiveFormattingPreference"]: 

291 raise NotImplementedError 

292 

293 @property 

294 def debputy_metadata(self) -> DebputyMetadata: 

295 """Information about `debputy` usage such as which integration mode is being used.""" 

296 src_pkg = self.source_package 

297 src_fields = src_pkg.fields if src_pkg else {} 

298 return DebputyMetadata.from_data( 

299 src_fields, 

300 self.dh_sequencer_data, 

301 ) 

302 

303 @property 

304 def dh_sequencer_data(self) -> DhSequencerData: 

305 """Information about the use of the `dh` sequencer 

306 

307 This includes which sequences are being used and whether the `dh` sequencer is used at all. 

308 """ 

309 raise NotImplementedError 

310 

311 @property 

312 def workspace_text_edit_support(self) -> WorkspaceTextEditSupport: 

313 raise NotImplementedError 

314 

315 @property 

316 def debputy_config(self) -> DebputyConfig: 

317 raise NotImplementedError 

318 

319 def spellchecker(self) -> "Spellchecker": 

320 checker = default_spellchecker() 

321 ignored_words = set() 

322 source_package = self.source_package 

323 binary_packages = self.binary_packages 

324 if source_package and (name := source_package.fields.get("Source")) is not None: 

325 ignored_words.add(name) 

326 if binary_packages: 

327 ignored_words.update(binary_packages.keys()) 

328 return checker.context_ignored_words(ignored_words) 

329 

330 async def slow_iter( 

331 self, 

332 iterable: Iterable[T], 

333 *, 

334 # Stub implement, arg present to mirror real implementation. 

335 yield_every: int = 100, # noqa: S1172 

336 ) -> AsyncIterable[T]: 

337 for value in iterable: 

338 yield value 

339 

340 def translation(self, domain: str) -> Translations: 

341 return l10n.translation( 

342 domain, 

343 ) 

344 

345 def related_diagnostic_information( 

346 self, 

347 text_range: "TERange", 

348 message: str, 

349 *, 

350 doc_uri: str | None = None, 

351 ) -> RelatedDiagnosticInformation: 

352 """Provide a related context for the diagnostic 

353 

354 The related diagnostic information is typically highlighted with the diagnostic. As an example, 

355 `debputy lint`'s terminal output will display the message and display the selected range after 

356 the diagnostic itself. 

357 

358 :param text_range: The text range to highlight. 

359 :param message: The message to associate with the provided text range. 

360 :param doc_uri: The URI of the document that the context is from. When omitted, the text range is 

361 assumed to be from the "current" file (the `doc_uri` attribute), which is also the default file 

362 for ranges passed to `emit_diagnostic`. 

363 :return: 

364 """ 

365 return RelatedDiagnosticInformation( 

366 text_range, 

367 message, 

368 doc_uri=doc_uri if doc_uri is not None else self.doc_uri, 

369 ) 

370 

371 def emit_diagnostic( 

372 self, 

373 text_range: "TERange", 

374 diagnostic_msg: str, 

375 severity: LintSeverity, 

376 authority_reference: str, 

377 *, 

378 quickfixes: Sequence[Mapping] | None = None, 

379 tags: list[types.DiagnosticTag] | None = None, 

380 related_information: list[RelatedDiagnosticInformation] | None = None, 

381 diagnostic_applies_to_another_file: str | None = None, 

382 enable_non_interactive_auto_fix: bool = True, 

383 ) -> None: 

384 """Emit a diagnostic for an issue detected in the current file. 

385 

386 :param text_range: The text range to highlight in the file. 

387 :param diagnostic_msg: The message to show to the user for this diagnostic 

388 :param severity: The severity to associate with the diagnostic. 

389 :param authority_reference: A reference to the authority / guide that this diagnostic is a violation of. 

390 

391 Use: 

392 * "Policy 3.4.1" for Debian Policy Manual section 3.4.1 

393 (replace the section number with the relevant number for your case) 

394 * "DevRef 6.2.2" for the Debian Developer Reference section 6.2.2 

395 (replace the section number with the relevant number for your case) 

396 * "debputy" for diagnostics without a reference or where `debputy` is the authority. 

397 (This is also used for cases where `debputy` filters the result. Like with spellchecking 

398 via hunspell, where `debputy` provides its own ignore list on top) 

399 

400 If you need a new reference, feel free to add it to this list. 

401 :param quickfixes: If provided, this is a list of possible fixes for this problem. 

402 Use the quickfixes provided in `debputy.lsp.quickfixes` such as `propose_correct_text_quick_fix`. 

403 :param tags: TODO: Not yet specified (currently uses LSP format). 

404 :param related_information: Provide additional context to the diagnostic. This can be used to define 

405 the source of a conflict. As an example, for duplicate definitions, this can be used to show where 

406 the definitions are. 

407 

408 Every item should be created via the `related_diagnostic_information` method. 

409 :param enable_non_interactive_auto_fix: Allow non-interactive auto-fixing (such as via 

410 `debputy lint --auto-fix`) of this issue. Set to `False` if the check is likely to have false 

411 positives. 

412 :param diagnostic_applies_to_another_file: Special-case parameter for flagging invalid file names. 

413 Leave this one at `None`, unless you know you need it. 

414 

415 It has non-obvious semantics and is primarily useful for reporting typos of filenames such as 

416 `debian/install`, etc. 

417 """ 

418 _check_diagnostic_source(authority_reference) 

419 lsp_severity = LINT_SEVERITY2LSP_SEVERITY[severity] 

420 diag_data: DiagnosticData = { 

421 "enable_non_interactive_auto_fix": enable_non_interactive_auto_fix, 

422 } 

423 

424 if severity not in NATIVELY_LSP_SUPPORTED_SEVERITIES: 

425 diag_data["lint_severity"] = severity 

426 if quickfixes: 

427 diag_data["quickfixes"] = quickfixes 

428 if diagnostic_applies_to_another_file is not None: 

429 diag_data["report_for_related_file"] = diagnostic_applies_to_another_file 

430 

431 lsp_range_client_units = self.position_codec.range_to_client_units( 

432 self.lines, 

433 te_range_to_lsp(text_range), 

434 ) 

435 

436 if related_information and any( 436 ↛ 439line 436 didn't jump to line 439 because the condition on line 436 was never true

437 i.doc_uri != self.doc_uri for i in related_information 

438 ): 

439 raise NotImplementedError("Ranges from another document will be wrong") 

440 

441 related_lsp_format = ( 

442 [i.to_lsp(self) for i in related_information] 

443 if related_information 

444 else None 

445 ) 

446 diag = types.Diagnostic( 

447 lsp_range_client_units, 

448 diagnostic_msg, 

449 severity=lsp_severity, 

450 source=authority_reference, 

451 data=diag_data if diag_data else None, 

452 tags=tags, 

453 related_information=related_lsp_format, 

454 ) 

455 self._emit_diagnostic(diag) 

456 

457 def _emit_diagnostic(self, diagnostic: types.Diagnostic) -> None: 

458 raise NotImplementedError 

459 

460 

461CLI_WORKSPACE_TEXT_EDIT_SUPPORT = WorkspaceTextEditSupport( 

462 supports_document_changes=True, 

463) 

464 

465 

466@dataclasses.dataclass(slots=True) 

467class LintStateImpl(LintState): 

468 plugin_feature_set: PluginProvidedFeatureSet = dataclasses.field(repr=False) 

469 maint_preference_table: "MaintainerPreferenceTable" = dataclasses.field(repr=False) 

470 source_root: VirtualPathBase | None 

471 debian_dir: VirtualPathBase | None 

472 path: str 

473 content: str 

474 lines: list[str] 

475 debputy_config: DebputyConfig 

476 source_package: SourcePackage | None = None 

477 binary_packages: Mapping[str, BinaryPackage] | None = None 

478 effective_preference: Optional["EffectiveFormattingPreference"] = None 

479 lint_implementation: Optional["AsyncLinterImpl"] = None 

480 _parsed_cache: Deb822FileElement | None = None 

481 _dh_sequencer_cache: DhSequencerData | None = None 

482 _diagnostics: list[types.Diagnostic] | None = None 

483 

484 @property 

485 def doc_uri(self) -> str: 

486 path = self.path 

487 abs_path = os.path.join(os.path.curdir, path) 

488 return f"file://{abs_path}" 

489 

490 @property 

491 def doc_version(self) -> int | None: 

492 return None 

493 

494 @property 

495 def position_codec(self) -> "LintCapablePositionCodec": 

496 return LINTER_POSITION_CODEC 

497 

498 @property 

499 def parsed_deb822_file_content(self) -> Deb822FileElement | None: 

500 cache = self._parsed_cache 

501 if cache is None: 

502 cache = parse_deb822_file( 

503 self.lines, 

504 accept_files_with_error_tokens=True, 

505 accept_files_with_duplicated_fields=True, 

506 ) 

507 self._parsed_cache = cache 

508 return cache 

509 

510 @property 

511 def dh_sequencer_data(self) -> DhSequencerData: 

512 dh_sequencer_cache = self._dh_sequencer_cache 

513 if dh_sequencer_cache is None: 

514 debian_dir = self.debian_dir 

515 dh_sequences: set[str] = set() 

516 saw_dh = False 

517 src_pkg = self.source_package 

518 drules = debian_dir.get("rules") if debian_dir is not None else None 

519 if drules and drules.is_file: 

520 try: 

521 with drules.open() as fd: 

522 saw_dh = parse_drules_for_addons(fd, dh_sequences) 

523 except PureVirtualPathError: 

524 pass 

525 if src_pkg: 

526 extract_dh_addons_from_control(src_pkg.fields, dh_sequences) 

527 

528 dh_sequencer_cache = DhSequencerData( 

529 frozenset(dh_sequences), 

530 saw_dh, 

531 ) 

532 self._dh_sequencer_cache = dh_sequencer_cache 

533 return dh_sequencer_cache 

534 

535 @property 

536 def workspace_text_edit_support(self) -> WorkspaceTextEditSupport: 

537 return CLI_WORKSPACE_TEXT_EDIT_SUPPORT 

538 

539 async def gather_diagnostics(self) -> list[types.Diagnostic]: 

540 if self._diagnostics is not None: 540 ↛ 541line 540 didn't jump to line 541 because the condition on line 540 was never true

541 raise RuntimeError( 

542 "run_diagnostics cannot be run while it is already running" 

543 ) 

544 linter = self.lint_implementation 

545 if linter is None: 545 ↛ 546line 545 didn't jump to line 546 because the condition on line 545 was never true

546 raise TypeError( 

547 "run_diagnostics cannot be run:" 

548 " LintState was created without a lint implementation (such as for reformat-only)" 

549 ) 

550 diagnostics: list[types.Diagnostic] = [] 

551 self._diagnostics = diagnostics 

552 

553 await linter(self) 

554 

555 self._diagnostics = None 

556 return diagnostics 

557 

558 def clear_cache(self) -> None: 

559 self._parsed_cache = None 

560 self._dh_sequencer_cache = None 

561 

562 def _emit_diagnostic(self, diagnostic: types.Diagnostic) -> None: 

563 diagnostics = self._diagnostics 

564 if diagnostics is None: 564 ↛ 565line 564 didn't jump to line 565 because the condition on line 564 was never true

565 raise TypeError("Cannot run emit_diagnostic outside of gather_diagnostics") 

566 diagnostics.append(diagnostic) 

567 

568 

569class LintDiagnosticResultState(IntEnum): 

570 REPORTED = 1 

571 MANUAL_FIXABLE = 2 

572 AUTO_FIXABLE = 3 

573 FIXED = 4 

574 

575 

576@dataclasses.dataclass(slots=True, frozen=True) 

577class LintDiagnosticResult: 

578 diagnostic: types.Diagnostic 

579 result_state: LintDiagnosticResultState 

580 invalid_marker: RuntimeError | None 

581 is_file_level_diagnostic: bool 

582 has_broken_range: bool 

583 missing_severity: bool 

584 discovered_in: str 

585 report_for_related_file: str | None 

586 

587 

588class LintReport: 

589 

590 def __init__(self) -> None: 

591 self.diagnostics_count: Counter[types.DiagnosticSeverity] = Counter() 

592 self.diagnostics_by_file: Mapping[str, list[LintDiagnosticResult]] = ( 

593 defaultdict(list) 

594 ) 

595 self.number_of_invalid_diagnostics: int = 0 

596 self.number_of_broken_diagnostics: int = 0 

597 self.lint_state: LintState | None = None 

598 self.start_timestamp = datetime.datetime.now() 

599 self.durations: dict[str, float] = collections.defaultdict(float) 

600 self._timer = time.perf_counter() 

601 

602 @contextlib.contextmanager 

603 def line_state(self, lint_state: LintState) -> collections.abc.Iterator[None]: 

604 previous = self.lint_state 

605 if previous is not None: 

606 path = previous.path 

607 duration = time.perf_counter() - self._timer 

608 self.durations[path] += duration 

609 

610 self.lint_state = lint_state 

611 

612 try: 

613 self._timer = time.perf_counter() 

614 yield 

615 finally: 

616 now = time.perf_counter() 

617 duration = now - self._timer 

618 self.durations[lint_state.path] += duration 

619 self._timer = now 

620 self.lint_state = previous 

621 

622 def report_diagnostic( 

623 self, 

624 diagnostic: types.Diagnostic, 

625 *, 

626 result_state: LintDiagnosticResultState = LintDiagnosticResultState.REPORTED, 

627 in_file: str | None = None, 

628 ) -> None: 

629 lint_state = self.lint_state 

630 assert lint_state is not None 

631 if in_file is None: 

632 in_file = lint_state.path 

633 assert in_file is not None 

634 discovered_in_file = in_file 

635 severity = diagnostic.severity 

636 missing_severity = False 

637 error_marker: RuntimeError | None = None 

638 if severity is None: 

639 self.number_of_invalid_diagnostics += 1 

640 severity = types.DiagnosticSeverity.Warning 

641 diagnostic.severity = severity 

642 missing_severity = True 

643 

644 lines = lint_state.lines 

645 diag_range = diagnostic.range 

646 start_pos = diag_range.start 

647 end_pos = diag_range.end 

648 diag_data = diagnostic.data 

649 if isinstance(diag_data, dict): 

650 report_for_related_file = diag_data.get("report_for_related_file") 

651 if report_for_related_file is None or not isinstance( 

652 report_for_related_file, str 

653 ): 

654 report_for_related_file = None 

655 else: 

656 in_file = report_for_related_file 

657 # Force it to exist in self.durations, since subclasses can use .items() or "foo" in self.durations. 

658 if in_file not in self.durations: 

659 self.durations[in_file] = 0 

660 else: 

661 report_for_related_file = None 

662 if report_for_related_file is not None: 

663 is_file_level_diagnostic = True 

664 else: 

665 is_file_level_diagnostic = _is_file_level_diagnostic( 

666 lines, 

667 start_pos.line, 

668 start_pos.character, 

669 end_pos.line, 

670 end_pos.character, 

671 ) 

672 has_broken_range = not is_file_level_diagnostic and ( 

673 end_pos.line > len(lines) or start_pos.line < 0 

674 ) 

675 

676 if has_broken_range or missing_severity: 

677 error_marker = RuntimeError("Registration Marker for invalid diagnostic") 

678 

679 diagnostic_result = LintDiagnosticResult( 

680 diagnostic, 

681 result_state, 

682 error_marker, 

683 is_file_level_diagnostic, 

684 has_broken_range, 

685 missing_severity, 

686 report_for_related_file=report_for_related_file, 

687 discovered_in=discovered_in_file, 

688 ) 

689 

690 self.diagnostics_by_file[in_file].append(diagnostic_result) 

691 self.diagnostics_count[severity] += 1 

692 self.process_diagnostic(in_file, lint_state, diagnostic_result) 

693 

694 def process_diagnostic( 

695 self, 

696 filename: str, 

697 lint_state: LintState, 

698 diagnostic_result: LintDiagnosticResult, 

699 ) -> None: 

700 # Subclass hook 

701 pass 

702 

703 def finish_report(self) -> None: 

704 # Subclass hook 

705 pass 

706 

707 

708_LS2DEBPUTY_SEVERITY: Mapping[types.DiagnosticSeverity, LintSeverity] = { 

709 types.DiagnosticSeverity.Error: "error", 

710 types.DiagnosticSeverity.Warning: "warning", 

711 types.DiagnosticSeverity.Information: "informational", 

712 types.DiagnosticSeverity.Hint: "pedantic", 

713} 

714 

715 

716_TERM_SEVERITY2TAG = { 

717 types.DiagnosticSeverity.Error: lambda fo, lint_tag=None: fo.colored( 

718 lint_tag if lint_tag else "error", 

719 fg="red", 

720 bg="black", 

721 style="bold", 

722 ), 

723 types.DiagnosticSeverity.Warning: lambda fo, lint_tag=None: fo.colored( 

724 lint_tag if lint_tag else "warning", 

725 fg="yellow", 

726 bg="black", 

727 style="bold", 

728 ), 

729 types.DiagnosticSeverity.Information: lambda fo, lint_tag=None: fo.colored( 

730 lint_tag if lint_tag else "informational", 

731 fg="blue", 

732 bg="black", 

733 style="bold", 

734 ), 

735 types.DiagnosticSeverity.Hint: lambda fo, lint_tag=None: fo.colored( 

736 lint_tag if lint_tag else "pedantic", 

737 fg="green", 

738 bg="black", 

739 style="bold", 

740 ), 

741} 

742 

743 

744def debputy_severity(diagnostic: types.Diagnostic) -> LintSeverity: 

745 lint_tag: LintSeverity | None = None 

746 if isinstance(diagnostic.data, dict): 

747 lint_tag = cast("LintSeverity", diagnostic.data.get("lint_severity")) 

748 

749 if lint_tag is not None: 

750 return lint_tag 

751 severity = diagnostic.severity 

752 if severity is None: 

753 return "warning" 

754 return _LS2DEBPUTY_SEVERITY.get(severity, "warning") 

755 

756 

757class TermLintReport(LintReport): 

758 

759 def __init__(self, fo: IOBasedOutputStyling) -> None: 

760 super().__init__() 

761 self.fo = fo 

762 

763 def finish_report(self) -> None: 

764 # Nothing to do for now 

765 pass 

766 

767 def process_diagnostic( 

768 self, 

769 filename: str, 

770 lint_state: LintState, 

771 diagnostic_result: LintDiagnosticResult, 

772 ) -> None: 

773 diagnostic = diagnostic_result.diagnostic 

774 fo = self.fo 

775 severity = diagnostic.severity 

776 assert severity is not None 

777 if diagnostic_result.result_state != LintDiagnosticResultState.FIXED: 

778 tag_unresolved = _TERM_SEVERITY2TAG[severity] 

779 lint_tag: LintSeverity | None = debputy_severity(diagnostic) 

780 tag = tag_unresolved(fo, lint_tag) 

781 else: 

782 tag = fo.colored( 

783 "auto-fixing", 

784 fg="magenta", 

785 bg="black", 

786 style="bold", 

787 ) 

788 

789 if diagnostic_result.is_file_level_diagnostic: 

790 start_line = 0 

791 start_position = 0 

792 end_line = 0 

793 end_position = 0 

794 else: 

795 start_line = diagnostic.range.start.line 

796 start_position = diagnostic.range.start.character 

797 end_line = diagnostic.range.end.line 

798 end_position = diagnostic.range.end.character 

799 

800 authority = diagnostic.source 

801 assert authority is not None 

802 diag_tags = f" [{authority}]" 

803 lines = lint_state.lines 

804 line_no_format_width = len(str(len(lines))) 

805 

806 if diagnostic_result.result_state == LintDiagnosticResultState.AUTO_FIXABLE: 

807 diag_tags += "[Correctable via --auto-fix]" 

808 elif diagnostic_result.result_state == LintDiagnosticResultState.MANUAL_FIXABLE: 

809 diag_tags += "[LSP interactive quickfix]" 

810 

811 code = f"[{diagnostic.code}]: " if diagnostic.code else "" 

812 msg = f"{code}{diagnostic.message}" 

813 

814 print( 

815 f"{tag}: File: {filename}:{start_line+1}:{start_position}:{end_line+1}:{end_position}: {msg}{diag_tags}", 

816 ) 

817 if diagnostic_result.missing_severity: 

818 _warn( 

819 " This warning did not have an explicit severity; Used Warning as a fallback!" 

820 ) 

821 if diagnostic_result.result_state == LintDiagnosticResultState.FIXED: 

822 # If it is fixed, there is no reason to show additional context. 

823 return 

824 if diagnostic_result.is_file_level_diagnostic: 

825 print(" File-level diagnostic") 

826 return 

827 if diagnostic_result.has_broken_range: 

828 _warn( 

829 "Bug in the underlying linter: The line numbers of the warning does not fit in the file..." 

830 ) 

831 return 

832 self._print_range_context(diagnostic.range, lines, line_no_format_width) 

833 related_info_list = diagnostic.related_information or [] 

834 hint_tag = fo.colored( 

835 "Related information", 

836 fg="magenta", 

837 bg="black", 

838 style="bold", 

839 ) 

840 for related_info in related_info_list: 

841 if related_info.location.uri != lint_state.doc_uri: 

842 continue 

843 print(f" {hint_tag}: {related_info.message}") 

844 self._print_range_context( 

845 related_info.location.range, lines, line_no_format_width 

846 ) 

847 

848 def _print_range_context( 

849 self, 

850 print_range: types.Range, 

851 lines: list[str], 

852 line_no_format_width: int, 

853 ) -> None: 

854 lines_to_print = _lines_to_print(print_range) 

855 fo = self.fo 

856 start_line = print_range.start.line 

857 for line_no in range(start_line, start_line + lines_to_print): 

858 line = _highlight_range(fo, lines[line_no], line_no, print_range) 

859 print(f" {line_no + 1:{line_no_format_width}}: {line}") 

860 

861 

862class LinterPositionCodec: 

863 

864 def client_num_units(self, chars: str): 

865 return len(chars) 

866 

867 def position_from_client_units( 

868 self, 

869 lines: list[str], 

870 position: types.Position, 

871 ) -> types.Position: 

872 

873 if len(lines) == 0: 

874 return types.Position(0, 0) 

875 if position.line >= len(lines): 

876 return types.Position(len(lines) - 1, self.client_num_units(lines[-1])) 

877 return position 

878 

879 def position_to_client_units( 

880 self, 

881 _lines: list[str], 

882 position: types.Position, 

883 ) -> types.Position: 

884 return position 

885 

886 def range_from_client_units( 

887 self, _lines: list[str], range: types.Range