Coverage for nlp_manager/tests/nlp_parser_tests.py: 97%

1""" 

2crate_anon/nlp_manager/tests/nlp_parser_tests.py 

3 

4=============================================================================== 

5 

6 Copyright (C) 2015, University of Cambridge, Department of Psychiatry. 

7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk). 

8 

9 This file is part of CRATE. 

10 

11 CRATE is free software: you can redistribute it and/or modify 

12 it under the terms of the GNU General Public License as published by 

13 the Free Software Foundation, either version 3 of the License, or 

14 (at your option) any later version. 

15 

16 CRATE is distributed in the hope that it will be useful, 

17 but WITHOUT ANY WARRANTY; without even the implied warranty of 

18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

19 GNU General Public License for more details. 

20 

21 You should have received a copy of the GNU General Public License 

22 along with CRATE. If not, see <https://www.gnu.org/licenses/>. 

23 

24=============================================================================== 

25 

26""" 

27 

28import logging 

29import sys 

30from typing import Any, Dict, Generator, List, Tuple 

31from unittest import mock, TestCase 

32 

33from sqlalchemy.exc import OperationalError 

34from sqlalchemy.schema import Column 

35 

36from crate_anon.nlp_manager.base_nlp_parser import BaseNlpParser 

37 

38 

39class FruitParser(BaseNlpParser): 

40 is_test_nlp_parser = True 

41 

42 def test(self, verbose: bool = False) -> None: 

43 pass 

44 

45 def dest_tables_columns(self) -> Dict[str, List[Column]]: 

46 return {} 

47 

48 def parse( 

49 self, text: str 

50 ) -> Generator[Tuple[str, Dict[str, Any]], None, None]: 

51 fruits = ("apple", "banana", "cherry", "fig") 

52 

53 for word in text.split(" "): 

54 if word.lower() in fruits: 

55 yield "output", {"fruit": word.lower()} 

56 

57 

58class NlpParserProcessTests(TestCase): 

59 def setUp(self) -> None: 

60 self.parser = FruitParser(None, None) 

61 

62 self.mock_execute_method = mock.Mock() 

63 self.mock_session = mock.Mock(execute=self.mock_execute_method) 

64 self.mock_db = mock.Mock(session=self.mock_session) 

65 

66 # can't set name attribute in constructor here as it has special 

67 # meaning 

68 mock_column = mock.Mock() 

69 mock_column.name = "fruit" # so set it here 

70 

71 self.mock_values_method = mock.Mock() 

72 mock_insert_object = mock.Mock(values=self.mock_values_method) 

73 mock_insert_method = mock.Mock(return_value=mock_insert_object) 

74 mock_sqla_table = mock.Mock( 

75 columns=[mock_column], insert=mock_insert_method 

76 ) 

77 self.mock_get_table = mock.Mock(return_value=mock_sqla_table) 

78 

79 self.mock_notify_transaction_method = mock.Mock() 

80 self.mock_nlpdef = mock.Mock( 

81 notify_transaction=self.mock_notify_transaction_method 

82 ) 

83 self.mock_nlpdef.name = "fruitdef" 

84 

85 def test_inserts_values(self) -> None: 

86 with self.assertLogs(level=logging.DEBUG) as logging_cm: 

87 with mock.patch.multiple( 

88 self.parser, 

89 _nlpdef=self.mock_nlpdef, 

90 _destdb=self.mock_db, 

91 get_table=self.mock_get_table, 

92 _friendly_name="Fruit", 

93 ): 

94 

95 starting_fields_values = {} 

96 

97 self.parser.process( 

98 "Apple Banana Cabbage Dandelion Edelweiss Fig", 

99 starting_fields_values, 

100 ) 

101 

102 self.mock_values_method.assert_any_call({"fruit": "apple"}) 

103 self.mock_values_method.assert_any_call({"fruit": "banana"}) 

104 self.mock_values_method.assert_any_call({"fruit": "fig"}) 

105 self.assertEqual(self.mock_values_method.call_count, 3) 

106 self.assertEqual(self.mock_execute_method.call_count, 3) 

107 

108 self.mock_notify_transaction_method.assert_any_call( 

109 self.mock_session, 

110 n_rows=1, 

111 n_bytes=sys.getsizeof({"fruit": "apple"}), 

112 force_commit=mock.ANY, 

113 ) 

114 self.mock_notify_transaction_method.assert_any_call( 

115 self.mock_session, 

116 n_rows=1, 

117 n_bytes=sys.getsizeof({"fruit": "banana"}), 

118 force_commit=mock.ANY, 

119 ) 

120 self.mock_notify_transaction_method.assert_any_call( 

121 self.mock_session, 

122 n_rows=1, 

123 n_bytes=sys.getsizeof({"fruit": "fig"}), 

124 force_commit=mock.ANY, 

125 ) 

126 self.assertEqual(self.mock_notify_transaction_method.call_count, 3) 

127 

128 logger_name = "crate_anon.nlp_manager.base_nlp_parser" 

129 self.assertIn( 

130 f"DEBUG:{logger_name}:NLP processor fruitdef/Fruit: found 3 values", # noqa: E501 

131 logging_cm.output, 

132 ) 

133 

134 def test_handles_failed_insert(self) -> None: 

135 self.mock_execute_method.side_effect = OperationalError( 

136 "Insert failed", None, None, None 

137 ) 

138 with self.assertLogs(level=logging.ERROR) as logging_cm: 

139 with mock.patch.multiple( 

140 self.parser, 

141 _nlpdef=self.mock_nlpdef, 

142 _destdb=self.mock_db, 

143 get_table=self.mock_get_table, 

144 _friendly_name="Fruit", 

145 ): 

146 

147 starting_fields_values = {} 

148 

149 self.parser.process("Apple", starting_fields_values) 

150 

151 self.mock_notify_transaction_method.assert_any_call( 

152 self.mock_session, 

153 n_rows=1, 

154 n_bytes=sys.getsizeof({"fruit": "apple"}), 

155 force_commit=mock.ANY, 

156 ) 

157 logger_name = "crate_anon.nlp_manager.base_nlp_parser" 

158 

159 self.assertIn(f"ERROR:{logger_name}", logging_cm.output[0]) 

160 self.assertIn("Insert failed", logging_cm.output[0])