Coverage for tests/test_samples.py: 100%

1import peprs 

2import pytest 

3 

4from pepdbagent.exceptions import SampleNotFoundError 

5 

6from .utils import PEPDBAgentContextManager 

7 

8 

9@pytest.mark.skipif( 

10 not PEPDBAgentContextManager().db_setup(), 

11 reason="DB is not setup", 

12) 

13class TestSamples: 

14 @pytest.mark.parametrize( 

15 "namespace, name, sample_name", 

16 [ 

17 ["namespace1", "amendments1", "pig_0h"], 

18 ], 

19 ) 

20 def test_retrieve_one_sample(self, namespace, name, sample_name): 

21 with PEPDBAgentContextManager(add_data=True) as agent: 

22 one_sample = agent.sample.get(namespace, name, sample_name, raw=False) 

23 assert isinstance(one_sample, peprs.Sample) 

24 assert one_sample.sample_name == sample_name 

25 

26 @pytest.mark.parametrize( 

27 "namespace, name, sample_name", 

28 [ 

29 ["namespace1", "amendments1", "pig_0h"], 

30 ], 

31 ) 

32 def test_retrieve_raw_sample(self, namespace, name, sample_name): 

33 with PEPDBAgentContextManager(add_data=True) as agent: 

34 one_sample = agent.sample.get(namespace, name, sample_name) 

35 assert isinstance(one_sample, dict) 

36 assert one_sample["sample_name"] == sample_name 

37 

38 @pytest.mark.parametrize( 

39 "namespace, name, sample_name", 

40 [ 

41 ["namespace2", "custom_index", "frog_1"], 

42 ], 

43 ) 

44 def test_retrieve_sample_with_modified_sample_id( 

45 self, namespace, name, sample_name 

46 ): 

47 with PEPDBAgentContextManager(add_data=True) as agent: 

48 one_sample = agent.sample.get(namespace, name, sample_name, raw=False) 

49 assert isinstance(one_sample, peprs.Sample) 

50 assert one_sample.sample_id == "frog_1" 

51 

52 @pytest.mark.parametrize( 

53 "namespace, name, sample_name", 

54 [ 

55 ["namespace1", "amendments1", "pig_0h"], 

56 ], 

57 ) 

58 def test_update(self, namespace, name, sample_name): 

59 with PEPDBAgentContextManager(add_data=True) as agent: 

60 agent.sample.update( 

61 namespace=namespace, 

62 name=name, 

63 tag="default", 

64 sample_name=sample_name, 

65 update_dict={"organism": "butterfly"}, 

66 ) 

67 one_sample = agent.sample.get(namespace, name, sample_name, raw=False) 

68 assert one_sample.organism == "butterfly" 

69 

70 @pytest.mark.parametrize( 

71 "namespace, name, sample_name", 

72 [ 

73 ["namespace1", "amendments1", "pig_0h"], 

74 ], 

75 ) 

76 def test_update_sample_name(self, namespace, name, sample_name): 

77 with PEPDBAgentContextManager(add_data=True) as agent: 

78 agent.sample.update( 

79 namespace=namespace, 

80 name=name, 

81 tag="default", 

82 sample_name=sample_name, 

83 update_dict={"sample_name": "butterfly"}, 

84 ) 

85 one_sample = agent.sample.get(namespace, name, "butterfly", raw=False) 

86 assert one_sample.sample_name == "butterfly" 

87 

88 @pytest.mark.parametrize( 

89 "namespace, name, sample_name", 

90 [ 

91 ["namespace2", "custom_index", "frog_1"], 

92 ], 

93 ) 

94 def test_update_custom_sample_id(self, namespace, name, sample_name): 

95 with PEPDBAgentContextManager(add_data=True) as agent: 

96 agent.sample.update( 

97 namespace=namespace, 

98 name=name, 

99 tag="default", 

100 sample_name=sample_name, 

101 update_dict={"sample_id": "butterfly"}, 

102 ) 

103 one_sample = agent.sample.get(namespace, name, "butterfly", raw=False) 

104 assert one_sample.sample_id == "butterfly" 

105 

106 @pytest.mark.parametrize( 

107 "namespace, name, sample_name", 

108 [ 

109 ["namespace1", "amendments1", "pig_0h"], 

110 ], 

111 ) 

112 def test_add_new_attributes(self, namespace, name, sample_name): 

113 with PEPDBAgentContextManager(add_data=True) as agent: 

114 agent.sample.update( 

115 namespace=namespace, 

116 name=name, 

117 tag="default", 

118 sample_name=sample_name, 

119 update_dict={"new_attr": "butterfly"}, 

120 ) 

121 prj = agent.project.get(namespace, name, raw=False) 

122 

123 assert prj.get_sample(sample_name).new_attr == "butterfly" 

124 

125 @pytest.mark.parametrize( 

126 "namespace, name, sample_name", 

127 [ 

128 ["namespace1", "amendments1", "pig_0h"], 

129 ], 

130 ) 

131 def test_project_timestamp_was_changed(self, namespace, name, sample_name): 

132 with PEPDBAgentContextManager(add_data=True) as agent: 

133 annotation1 = agent.annotation.get(namespace, name, "default") 

134 import time 

135 

136 time.sleep(0.2) 

137 agent.sample.update( 

138 namespace=namespace, 

139 name=name, 

140 tag="default", 

141 sample_name=sample_name, 

142 update_dict={"new_attr": "butterfly"}, 

143 ) 

144 annotation2 = agent.annotation.get(namespace, name, "default") 

145 

146 assert ( 

147 annotation1.results[0].last_update_date 

148 != annotation2.results[0].last_update_date 

149 ) 

150 

151 @pytest.mark.parametrize( 

152 "namespace, name, sample_name", 

153 [ 

154 ["namespace1", "amendments1", "pig_0h"], 

155 ], 

156 ) 

157 def test_delete_sample(self, namespace, name, sample_name): 

158 with PEPDBAgentContextManager(add_data=True) as agent: 

159 one_sample = agent.sample.get(namespace, name, sample_name, raw=False) 

160 assert isinstance(one_sample, peprs.Sample) 

161 

162 agent.sample.delete(namespace, name, tag="default", sample_name=sample_name) 

163 

164 with pytest.raises(SampleNotFoundError): 

165 agent.sample.get( 

166 namespace, name, tag="default", sample_name=sample_name, raw=False 

167 ) 

168 

169 @pytest.mark.parametrize( 

170 "namespace, name, tag, sample_dict", 

171 [ 

172 [ 

173 "namespace1", 

174 "amendments1", 

175 "default", 

176 { 

177 "sample_name": "new_sample", 

178 "time": "new_time", 

179 }, 

180 ], 

181 ], 

182 ) 

183 def test_add_sample(self, namespace, name, tag, sample_dict): 

184 with PEPDBAgentContextManager(add_data=True) as agent: 

185 prj = agent.project.get(namespace, name, raw=False) 

186 agent.sample.add(namespace, name, tag, sample_dict) 

187 

188 prj2 = agent.project.get(namespace, name, raw=False) 

189 

190 assert len(prj.samples) + 1 == len(prj2.samples) 

191 assert prj2.samples[-1].sample_name == sample_dict["sample_name"] 

192 

193 @pytest.mark.parametrize( 

194 "namespace, name, tag, sample_dict", 

195 [ 

196 [ 

197 "namespace1", 

198 "amendments1", 

199 "default", 

200 { 

201 "sample_name": "pig_0h", 

202 "time": "new_time", 

203 }, 

204 ], 

205 ], 

206 ) 

207 def test_overwrite_sample(self, namespace, name, tag, sample_dict): 

208 with PEPDBAgentContextManager(add_data=True) as agent: 

209 # peprs/polars infers numeric values from the sample table, so the 

210 # original time is loaded as int (not string). 

211 assert ( 

212 agent.project.get(namespace, name, raw=False).get_sample("pig_0h").time 

213 == 0 

214 ) 

215 agent.sample.add(namespace, name, tag, sample_dict, overwrite=True) 

216 

217 assert ( 

218 agent.project.get(namespace, name, raw=False).get_sample("pig_0h").time 

219 == "new_time" 

220 ) 

221 

222 @pytest.mark.parametrize( 

223 "namespace, name, tag, sample_dict", 

224 [ 

225 [ 

226 "namespace1", 

227 "amendments1", 

228 "default", 

229 { 

230 "sample_name": "new_sample", 

231 "time": "new_time", 

232 }, 

233 ], 

234 ], 

235 ) 

236 def test_delete_and_add(self, namespace, name, tag, sample_dict): 

237 with PEPDBAgentContextManager(add_data=True) as agent: 

238 prj = agent.project.get(namespace, name, raw=False) 

239 sample_dict = agent.sample.get(namespace, name, "pig_0h", raw=True) 

240 agent.sample.delete(namespace, name, tag, "pig_0h") 

241 agent.sample.add(namespace, name, tag, sample_dict) 

242 prj2 = agent.project.get(namespace, name, raw=False) 

243 assert ( 

244 prj.get_sample("pig_0h").to_dict() 

245 == prj2.get_sample("pig_0h").to_dict() 

246 )