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

1"""Tests for stream_json_lines function.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7import pytest 

8from assertpy import assert_that 

9 

10from lintro.parsers.streaming import stream_json_lines 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Callable 

14 

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

16 

17 

18def test_parses_string( 

19 parse_test_item: Callable[[dict[str, object]], SimpleIssue | None], 

20) -> None: 

21 """Parse newline-separated JSON objects from string. 

22 

23 Args: 

24 parse_test_item: Fixture providing a parser function for test items. 

25 """ 

26 output = ( 

27 '{"file": "a.py", "message": "error1"}\n{"file": "b.py", "message": "error2"}\n' 

28 ) 

29 results = list(stream_json_lines(output, parse_test_item)) 

30 

31 assert_that(results).is_length(2) 

32 assert_that(results[0].file).is_equal_to("a.py") 

33 assert_that(results[1].file).is_equal_to("b.py") 

34 

35 

36def test_parses_iterable( 

37 parse_test_item: Callable[[dict[str, object]], SimpleIssue | None], 

38) -> None: 

39 """Parse JSON objects from line iterable. 

40 

41 Args: 

42 parse_test_item: Fixture providing a parser function for test items. 

43 """ 

44 lines = [ 

45 '{"file": "c.py", "message": "msg1"}', 

46 '{"file": "d.py", "message": "msg2"}', 

47 ] 

48 results = list(stream_json_lines(lines, parse_test_item)) 

49 

50 assert_that(results).is_length(2) 

51 assert_that(results[0].file).is_equal_to("c.py") 

52 assert_that(results[1].file).is_equal_to("d.py") 

53 

54 

55def test_skips_empty_lines( 

56 parse_test_item: Callable[[dict[str, object]], SimpleIssue | None], 

57) -> None: 

58 """Skip empty lines in output. 

59 

60 Args: 

61 parse_test_item: Fixture providing a parser function for test items. 

62 """ 

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

64 results = list(stream_json_lines(output, parse_test_item)) 

65 

66 assert_that(results).is_length(2) 

67 

68 

69def test_skips_non_json_lines( 

70 parse_test_item: Callable[[dict[str, object]], SimpleIssue | None], 

71) -> None: 

72 """Skip lines that don't start with opening brace. 

73 

74 Args: 

75 parse_test_item: Fixture providing a parser function for test items. 

76 """ 

77 output = 'some info\n{"file": "a.py"}\nmore text\n' 

78 results = list(stream_json_lines(output, parse_test_item)) 

79 

80 assert_that(results).is_length(1) 

81 assert_that(results[0].file).is_equal_to("a.py") 

82 

83 

84def test_handles_invalid_json( 

85 parse_test_item: Callable[[dict[str, object]], SimpleIssue | None], 

86) -> None: 

87 """Handle invalid JSON gracefully without raising. 

88 

89 Args: 

90 parse_test_item: Fixture providing a parser function for test items. 

91 """ 

92 output = '{"file": "a.py"}\n{invalid json}\n{"file": "b.py"}\n' 

93 results = list(stream_json_lines(output, parse_test_item)) 

94 

95 assert_that(results).is_length(2) 

96 

97 

98@pytest.mark.parametrize( 

99 "output", 

100 [ 

101 "", 

102 "\n", 

103 "\n\n\n", 

104 ], 

105) 

106def test_empty_output_yields_nothing( 

107 output: str, 

108 parse_test_item: Callable[[dict[str, object]], SimpleIssue | None], 

109) -> None: 

110 """Empty or whitespace-only output yields no results. 

111 

112 Args: 

113 output: Parameterized output string to parse. 

114 parse_test_item: Fixture providing a parser function for test items. 

115 """ 

116 results = list(stream_json_lines(output, parse_test_item)) 

117 

118 assert_that(results).is_empty() 

119 

120 

121def test_skips_items_where_parser_returns_none( 

122 parse_test_item: Callable[[dict[str, object]], SimpleIssue | None], 

123) -> None: 

124 """Skip items when parse function returns None. 

125 

126 Args: 

127 parse_test_item: Fixture providing a parser function for test items. 

128 """ 

129 

130 def selective_parser(item: dict[str, object]) -> SimpleIssue | None: 

131 if item.get("skip"): 

132 return None 

133 return parse_test_item(item) 

134 

135 output = '{"file": "a.py"}\n{"file": "b.py", "skip": true}\n{"file": "c.py"}\n' 

136 results = list(stream_json_lines(output, selective_parser)) 

137 

138 assert_that(results).is_length(2) 

139 assert_that(results[0].file).is_equal_to("a.py") 

140 assert_that(results[1].file).is_equal_to("c.py")