Coverage for tests / unit / parsers / streaming / test_parser_class.py: 100%

1"""Tests for StreamingParser class.""" 

2 

3from __future__ import annotations 

4 

5import pytest 

6from assertpy import assert_that 

7 

8from lintro.parsers.base_issue import BaseIssue 

9from lintro.parsers.streaming import StreamingParser 

10from tests.unit.parsers.streaming.conftest import SimpleIssue 

11 

12 

13def test_stream_json_lines_uses_parse_item() -> None: 

14 """StreamingParser.stream_json_lines uses parse_item method.""" 

15 

16 class TestParser(StreamingParser): 

17 def parse_item(self, item: dict[str, object]) -> BaseIssue | None: 

18 file_val = item.get("file") 

19 return SimpleIssue( 

20 file=str(file_val) if file_val else "", 

21 ) 

22 

23 parser = TestParser("test") 

24 output = '{"file": "a.py"}\n{"file": "b.py"}\n' 

25 results: list[BaseIssue] = list(parser.stream_json_lines(output)) 

26 

27 assert_that(results).is_length(2) 

28 

29 

30def test_stream_text_lines_uses_parse_line() -> None: 

31 """StreamingParser.stream_text_lines uses parse_line method.""" 

32 

33 class TestParser(StreamingParser): 

34 def parse_line(self, line: str) -> BaseIssue | None: 

35 if line.startswith("E:"): 

36 return SimpleIssue(message=line[2:].strip()) 

37 return None 

38 

39 parser = TestParser("test") 

40 output = "I: info\nE: error\nI: more\n" 

41 results: list[BaseIssue] = list(parser.stream_text_lines(output)) 

42 

43 assert_that(results).is_length(1) 

44 assert_that(results[0].message).is_equal_to("error") 

45 

46 

47def test_stream_json_array_uses_parse_item() -> None: 

48 """StreamingParser.stream_json_array uses parse_item method.""" 

49 

50 class TestParser(StreamingParser): 

51 def parse_item(self, item: dict[str, object]) -> BaseIssue | None: 

52 file_val = item.get("file") 

53 return SimpleIssue( 

54 file=str(file_val) if file_val else "", 

55 ) 

56 

57 parser = TestParser("test") 

58 output = '[{"file": "a.py"}]' 

59 results: list[BaseIssue] = list(parser.stream_json_array(output)) 

60 

61 assert_that(results).is_length(1) 

62 

63 

64@pytest.mark.parametrize( 

65 ("method", "args"), 

66 [ 

67 ("parse_item", ({},)), 

68 ("parse_line", ("test",)), 

69 ], 

70) 

71def test_methods_raise_not_implemented( 

72 method: str, 

73 args: tuple[object, ...], 

74) -> None: 

75 """parse_item and parse_line raise NotImplementedError by default. 

76 

77 Args: 

78 method: The method name to test. 

79 args: Arguments to pass to the method. 

80 """ 

81 parser = StreamingParser("test") 

82 

83 with pytest.raises(NotImplementedError): 

84 getattr(parser, method)(*args)