Coverage for tests/test_project_history.py: 100%

1import peprs 

2import pytest 

3 

4from pepdbagent.const import PEPHUB_SAMPLE_ID_KEY 

5from pepdbagent.exceptions import HistoryNotFoundError 

6 

7from .utils import PEPDBAgentContextManager 

8 

9 

10@pytest.mark.skipif( 

11 not PEPDBAgentContextManager().db_setup(), 

12 reason="DB is not setup", 

13) 

14class TestProjectHistory: 

15 """ 

16 Test project methods 

17 """ 

18 

19 @pytest.mark.parametrize( 

20 "namespace, name, sample_name", 

21 [ 

22 ["namespace1", "amendments1", "pig_0h"], 

23 ], 

24 ) 

25 def test_get_add_history_all_annotation(self, namespace, name, sample_name): 

26 with PEPDBAgentContextManager(add_data=True) as agent: 

27 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

28 

29 prj["samples"][0]["sample_name"] = "new_sample_name" 

30 

31 del prj["samples"][1] 

32 del prj["samples"][2] 

33 new_sample1 = { 

34 "sample_name": "new_sample", 

35 "protocol": "new_protocol", 

36 PEPHUB_SAMPLE_ID_KEY: None, 

37 } 

38 new_sample2 = { 

39 "sample_name": "new_sample2", 

40 "protocol": "new_protocol2", 

41 PEPHUB_SAMPLE_ID_KEY: None, 

42 } 

43 

44 prj["samples"].append(new_sample1.copy()) 

45 prj["samples"].append(new_sample2.copy()) 

46 

47 agent.project.update( 

48 namespace=namespace, 

49 name=name, 

50 tag="default", 

51 update_dict={"project": peprs.Project.from_dict(prj)}, 

52 ) 

53 

54 project_history = agent.project.get_history(namespace, name, tag="default") 

55 

56 assert len(project_history.history) == 1 

57 

58 @pytest.mark.parametrize( 

59 "namespace, name, sample_name", 

60 [ 

61 ["namespace1", "amendments1", "pig_0h"], 

62 ], 

63 ) 

64 def test_get_add_history_all_project(self, namespace, name, sample_name): 

65 with PEPDBAgentContextManager(add_data=True) as agent: 

66 prj_init = agent.project.get(namespace, name, tag="default", raw=False) 

67 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

68 

69 # prj["samples"][0]["sample_name"] = "new_sample_name" 

70 

71 del prj["samples"][1] 

72 del prj["samples"][2] 

73 new_sample1 = { 

74 "sample_name": "new_sample", 

75 "protocol": "new_protocol", 

76 PEPHUB_SAMPLE_ID_KEY: None, 

77 } 

78 new_sample2 = { 

79 "sample_name": "new_sample2", 

80 "protocol": "new_protocol2", 

81 PEPHUB_SAMPLE_ID_KEY: None, 

82 } 

83 

84 prj["samples"].append(new_sample1.copy()) 

85 prj["samples"].append(new_sample2.copy()) 

86 

87 agent.project.update( 

88 namespace=namespace, 

89 name=name, 

90 tag="default", 

91 update_dict={"project": peprs.Project.from_dict(prj)}, 

92 ) 

93 

94 history_prj = agent.project.get_project_from_history( 

95 namespace, name, tag="default", history_id=1, raw=False 

96 ) 

97 assert prj_init == history_prj 

98 

99 @pytest.mark.parametrize( 

100 "namespace, name, sample_name", 

101 [ 

102 ["namespace1", "amendments1", "pig_0h"], 

103 ], 

104 ) 

105 def test_get_history_multiple_changes(self, namespace, name, sample_name): 

106 with PEPDBAgentContextManager(add_data=True) as agent: 

107 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

108 

109 del prj["samples"][1] 

110 

111 agent.project.update( 

112 namespace=namespace, 

113 name=name, 

114 tag="default", 

115 update_dict={"project": peprs.Project.from_dict(prj)}, 

116 ) 

117 

118 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

119 

120 new_sample1 = { 

121 "sample_name": "new_sample", 

122 "protocol": "new_protocol", 

123 PEPHUB_SAMPLE_ID_KEY: None, 

124 } 

125 prj["samples"].append(new_sample1.copy()) 

126 

127 agent.project.update( 

128 namespace=namespace, 

129 name=name, 

130 tag="default", 

131 update_dict={"project": peprs.Project.from_dict(prj)}, 

132 ) 

133 

134 history = agent.project.get_history(namespace, name, tag="default") 

135 

136 assert len(history.history) == 2 

137 

138 @pytest.mark.parametrize( 

139 "namespace, name, sample_name", 

140 [ 

141 ["namespace1", "amendments1", "pig_0h"], 

142 ], 

143 ) 

144 def test_get_project_incorrect_history_id(self, namespace, name, sample_name): 

145 with PEPDBAgentContextManager(add_data=True) as agent: 

146 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

147 del prj["samples"][1] 

148 

149 agent.project.update( 

150 namespace=namespace, 

151 name=name, 

152 tag="default", 

153 update_dict={"project": peprs.Project.from_dict(prj)}, 

154 ) 

155 

156 with pytest.raises(HistoryNotFoundError): 

157 agent.project.get_project_from_history( 

158 namespace, "amendments2", tag="default", history_id=1, raw=False 

159 ) 

160 

161 @pytest.mark.parametrize( 

162 "namespace, name, sample_name", 

163 [ 

164 ["namespace1", "amendments1", "pig_0h"], 

165 ], 

166 ) 

167 def test_get_history_none(self, namespace, name, sample_name): 

168 with PEPDBAgentContextManager(add_data=True) as agent: 

169 history_annot = agent.project.get_history(namespace, name, tag="default") 

170 assert len(history_annot.history) == 0 

171 

172 @pytest.mark.parametrize( 

173 "namespace, name, sample_name", 

174 [ 

175 ["namespace1", "amendments1", "pig_0h"], 

176 ], 

177 ) 

178 def test_delete_all_history(self, namespace, name, sample_name): 

179 with PEPDBAgentContextManager(add_data=True) as agent: 

180 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

181 

182 del prj["samples"][1] 

183 

184 agent.project.update( 

185 namespace=namespace, 

186 name=name, 

187 tag="default", 

188 update_dict={"project": peprs.Project.from_dict(prj)}, 

189 ) 

190 

191 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

192 

193 new_sample1 = { 

194 "sample_name": "new_sample", 

195 "protocol": "new_protocol", 

196 PEPHUB_SAMPLE_ID_KEY: None, 

197 } 

198 prj["samples"].append(new_sample1.copy()) 

199 

200 agent.project.update( 

201 namespace=namespace, 

202 name=name, 

203 tag="default", 

204 update_dict={"project": peprs.Project.from_dict(prj)}, 

205 ) 

206 

207 history = agent.project.get_history(namespace, name, tag="default") 

208 

209 assert len(history.history) == 2 

210 

211 agent.project.delete_history( 

212 namespace, name, tag="default", history_id=None 

213 ) 

214 

215 history = agent.project.get_history(namespace, name, tag="default") 

216 assert len(history.history) == 0 

217 

218 project_exists = agent.project.exists(namespace, name, tag="default") 

219 assert project_exists 

220 

221 @pytest.mark.parametrize( 

222 "namespace, name, sample_name", 

223 [ 

224 ["namespace1", "amendments1", "pig_0h"], 

225 ], 

226 ) 

227 def test_delete_one_history(self, namespace, name, sample_name): 

228 with PEPDBAgentContextManager(add_data=True) as agent: 

229 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

230 

231 del prj["samples"][1] 

232 

233 agent.project.update( 

234 namespace=namespace, 

235 name=name, 

236 tag="default", 

237 update_dict={"project": peprs.Project.from_dict(prj)}, 

238 ) 

239 

240 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

241 

242 new_sample1 = { 

243 "sample_name": "new_sample", 

244 "protocol": "new_protocol", 

245 PEPHUB_SAMPLE_ID_KEY: None, 

246 } 

247 prj["samples"].append(new_sample1.copy()) 

248 

249 agent.project.update( 

250 namespace=namespace, 

251 name=name, 

252 tag="default", 

253 update_dict={"project": peprs.Project.from_dict(prj)}, 

254 ) 

255 

256 history = agent.project.get_history(namespace, name, tag="default") 

257 

258 assert len(history.history) == 2 

259 

260 agent.project.delete_history(namespace, name, tag="default", history_id=1) 

261 

262 history = agent.project.get_history(namespace, name, tag="default") 

263 

264 assert len(history.history) == 1 

265 assert history.history[0].change_id == 2 

266 

267 @pytest.mark.parametrize( 

268 "namespace, name, sample_name", 

269 [ 

270 ["namespace1", "amendments1", "pig_0h"], 

271 ], 

272 ) 

273 def test_restore_project(self, namespace, name, sample_name): 

274 with PEPDBAgentContextManager(add_data=True) as agent: 

275 prj_org = agent.project.get(namespace, name, tag="default", with_id=False) 

276 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

277 

278 del prj["samples"][1] 

279 

280 agent.project.update( 

281 namespace=namespace, 

282 name=name, 

283 tag="default", 

284 update_dict={"project": peprs.Project.from_dict(prj)}, 

285 ) 

286 

287 prj = agent.project.get(namespace, name, tag="default", with_id=True) 

288 

289 new_sample1 = { 

290 "sample_name": "new_sample", 

291 "protocol": "new_protocol", 

292 PEPHUB_SAMPLE_ID_KEY: None, 

293 } 

294 prj["samples"].append(new_sample1.copy()) 

295 

296 agent.project.update( 

297 namespace=namespace, 

298 name=name, 

299 tag="default", 

300 update_dict={"project": peprs.Project.from_dict(prj)}, 

301 ) 

302 

303 agent.project.restore(namespace, name, tag="default", history_id=1) 

304 

305 restored_project = agent.project.get( 

306 namespace, name, tag="default", with_id=False 

307 ) 

308 

309 assert prj_org == restored_project