Coverage for tests\test_request.py: 100%

1import os 

2from contextlib import redirect_stdout 

3from io import StringIO 

4from tempfile import TemporaryDirectory 

5 

6import pytest 

7import pandas 

8 

9import receptiviti 

10 

11receptiviti.readin_env() 

12 

13 

14@pytest.mark.skipif(os.getenv("RECEPTIVITI_KEY") is None, reason="no API key") 

15class TestRequest: 

16 def test_unreachable(self): 

17 url = "http://localhost:0/not_served" 

18 assert receptiviti.status(url) is None 

19 with pytest.raises(RuntimeError, match="URL is not reachable"): 

20 receptiviti.request("a text", url=url) 

21 

22 def test_single_text(self): 

23 res = receptiviti.request("text to score") 

24 assert res["summary.word_count"][0] == 3 

25 

26 def test_invalid_text(self): 

27 with pytest.raises(RuntimeError, match="one of your texts is over the bundle size limit"): 

28 receptiviti.request(" " * int(1e7)) 

29 with pytest.raises(RuntimeError, match="no valid texts to process"): 

30 receptiviti.request("") 

31 

32 def test_multi_text(self): 

33 res = receptiviti.request(["text to score", float("nan"), "another text", "another text"]) 

34 assert str(res["summary.word_count"][1]) == "nan" 

35 assert res["summary.word_count"].iloc[pandas.Index([0, 2, 3])].to_list() == [3, 2, 2] 

36 

37 def test_framework_selection(self): 

38 res = receptiviti.request(["text to score", "another text"], frameworks="summary") 

39 assert res.shape == (2, 10) 

40 assert res["word_count"].to_list() == [3, 2] 

41 

42 def test_framework_prefix(self): 

43 res = receptiviti.request( 

44 ["text to score", "another text"], frameworks="summary", framework_prefix=True 

45 ) 

46 assert res["summary.word_count"].to_list() == [3, 2] 

47 

48 def test_id_text(self): 

49 res = receptiviti.request( 

50 ["text to score", "another text"], ids=["a", "b"], return_text=True 

51 ) 

52 assert res["text"].to_list() == ["text to score", "another text"] 

53 assert res["id"].to_list() == ["a", "b"] 

54 

55 def test_verbose(self): 

56 with redirect_stdout(StringIO()) as out: 

57 receptiviti.request("text to score", frameworks=["summary", "sallee"], verbose=True) 

58 messages = out.getvalue().split("\n") 

59 expected = ["prep"] * 3 + ["requ", "done", "prep", "sele", "done", ""] 

60 assert len(messages) == len(expected) and all( 

61 line[:4] == expected[i] for i, line in enumerate(messages) 

62 ) 

63 

64 def test_cache_initialization(self): 

65 with TemporaryDirectory() as tempdir: 

66 receptiviti.request("a text to score", cache=tempdir, clear_cache=True) 

67 assert os.path.isdir(tempdir + "/bin=h") 

68 

69 def test_id_assignment(self): 

70 text = ["text to score", "another text"] 

71 with TemporaryDirectory() as tempdir: 

72 txt_file = f"{tempdir}/text.txt" 

73 with open(txt_file, "w", encoding="utf-8") as txt: 

74 txt.write("\n".join(text)) 

75 csv_file = f"{tempdir}/text.csv" 

76 pandas.DataFrame({"text": text}).to_csv(csv_file) 

77 assert receptiviti.request(txt_file)["id"].to_list() == [ 

78 txt_file + "0", 

79 txt_file + "1", 

80 ] 

81 assert receptiviti.request(csv_file, text_column="text")["id"].to_list() == [ 

82 csv_file + "0", 

83 csv_file + "1", 

84 ] 

85 

86 @pytest.mark.skipif(not os.path.isfile("../data.txt"), reason="no txt test file present") 

87 def test_from_directory(self): 

88 with TemporaryDirectory() as tempdir: 

89 res_single = receptiviti.request("../data.txt") 

90 nth_text = 0 

91 txt_files = [] 

92 csv_files = [] 

93 with open("../data.txt", encoding="utf-8") as texts: 

94 for text in texts: 

95 nth_text += 1 

96 txt_file = f"{tempdir}/{nth_text}.txt" 

97 txt_files.append(txt_file) 

98 with open(txt_file, "w", encoding="utf-8") as txt: 

99 txt.write(text) 

100 csv_file = f"{tempdir}/{nth_text}.csv" 

101 csv_files.append(csv_file) 

102 pandas.DataFrame({"text": [text]}).to_csv(csv_file) 

103 res_multi = receptiviti.request(tempdir) 

104 res_multi_txt = receptiviti.request(txt_files) 

105 res_multi_csv = receptiviti.request(csv_files, text_column="text") 

106 assert res_single["summary.word_count"].sum() == res_multi["summary.word_count"].sum() 

107 assert res_multi["summary.word_count"].sum() == res_multi_txt["summary.word_count"].sum() 

108 assert res_multi["summary.word_count"].sum() == res_multi_csv["summary.word_count"].sum() 

109 

110 @pytest.mark.skipif(not os.path.isfile("../data.csv"), reason="no csv test file present") 

111 def test_from_file(self): 

112 with TemporaryDirectory() as tempdir: 

113 res_parallel = receptiviti.request( 

114 "../data.csv", 

115 text_column="texts", 

116 id_column="id", 

117 bundle_size=20, 

118 cache=tempdir, 

119 in_memory=False, 

120 ) 

121 res_serial = receptiviti.request( 

122 "../data.csv", 

123 text_column="texts", 

124 id_column="id", 

125 bundle_size=20, 

126 cores=1, 

127 cache=tempdir, 

128 ) 

129 assert res_parallel["summary.word_count"].sum() == res_serial["summary.word_count"].sum() 

130 

131 @pytest.mark.skipif( 

132 receptiviti.status(os.getenv("RECEPTIVITI_URL_TEST")) is None, 

133 reason="test API is not reachable", 

134 ) 

135 def test_endpoint_version(self): 

136 with TemporaryDirectory() as tempdir: 

137 receptiviti.request( 

138 "text to process", 

139 url=os.getenv("RECEPTIVITI_URL_TEST"), 

140 key=os.getenv("RECEPTIVITI_KEY_TEST"), 

141 secret=os.getenv("RECEPTIVITI_SECRET_TEST"), 

142 cache=tempdir, 

143 ) 

144 with redirect_stdout(StringIO()) as out: 

145 receptiviti.request( 

146 "text to process", 

147 url=os.getenv("RECEPTIVITI_URL_TEST") + "v2/taxonomies", 

148 key=os.getenv("RECEPTIVITI_KEY_TEST"), 

149 secret=os.getenv("RECEPTIVITI_SECRET_TEST"), 

150 cache=tempdir, 

151 verbose=True, 

152 ) 

153 messages = out.getvalue().split("\n") 

154 expected = ["prep"] * 3 + ["requ", "done", "clea", "addi", "prep", "done", ""] 

155 assert len(messages) == len(expected) and all( 

156 line[:4] == expected[i] for i, line in enumerate(messages) 

157 )