Coverage for tests/test_bedset.py: 100%

1import os 

2 

3import pytest 

4from sqlalchemy.orm import Session 

5from sqlalchemy.sql import select 

6 

7from bbconf.db_utils import BedSets 

8from bbconf.exceptions import BedbaseS3ConnectionError, BedSetNotFoundError 

9 

10from .conftest import DATA_PATH, SERVICE_UNAVAILABLE 

11from .utils import BED_TEST_ID, BEDSET_TEST_ID, ContextManagerDBTesting 

12 

13 

14@pytest.mark.skipif(SERVICE_UNAVAILABLE, reason="Database is not available") 

15class TestBedset: 

16 def test_calculate_stats(self, bbagent_obj): 

17 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): 

18 results = bbagent_obj.bedset._calculate_statistics([BED_TEST_ID]) 

19 

20 assert results is not None 

21 assert results.sd is not None 

22 assert results.mean is not None 

23 

24 def test_crate_bedset_all(self, bbagent_obj, mocker): 

25 with ContextManagerDBTesting( 

26 config=bbagent_obj.config, add_data=True, bedset=False 

27 ): 

28 mocker.patch( 

29 "bbconf.config_parser.bedbaseconfig.BedBaseConfig.upload_s3", 

30 return_value=True, 

31 ) 

32 bbagent_obj.bedset.create( 

33 "testinoo", 

34 "test_name", 

35 description="this is test description", 

36 bedid_list=[ 

37 BED_TEST_ID, 

38 ], 

39 plots={ 

40 "region_commonality": { 

41 "name": "region_commonality", 

42 "description": "Regions distribution over chromosomes", 

43 "title": "Regions distribution over chromosomes", 

44 "path": os.path.join( 

45 DATA_PATH, 

46 "plots/bbad85f21962bb8d972444f7f9a3a932_chrombins.pdf", 

47 ), 

48 "path_thumbnail": os.path.join( 

49 DATA_PATH, 

50 "/plots/bbad85f21962bb8d972444f7f9a3a932_chrombins.png", 

51 ), 

52 }, 

53 }, 

54 statistics=True, 

55 upload_s3=True, 

56 no_fail=True, 

57 ) 

58 with Session(bbagent_obj.config.db_engine.engine) as session: 

59 result = session.scalar(select(BedSets).where(BedSets.id == "testinoo")) 

60 assert result is not None 

61 assert result.name == "test_name" 

62 assert len([k for k in result.files]) == 1 

63 assert result.bedfile_count == 1 

64 

65 def test_get_metadata_full(self, bbagent_obj): 

66 with ContextManagerDBTesting( 

67 config=bbagent_obj.config, add_data=True, bedset=True 

68 ): 

69 result = bbagent_obj.bedset.get(BEDSET_TEST_ID, full=True) 

70 

71 assert result.id == BEDSET_TEST_ID 

72 assert result.md5sum == "bbad0000000000000000000000000000" 

73 assert result.statistics.sd is not None 

74 assert result.statistics.mean is not None 

75 assert result.plots is not None 

76 assert result.bedfile_count == 1 

77 

78 def test_get_metadata_not_full(self, bbagent_obj): 

79 with ContextManagerDBTesting( 

80 config=bbagent_obj.config, add_data=True, bedset=True 

81 ): 

82 result = bbagent_obj.bedset.get(BEDSET_TEST_ID, full=False) 

83 

84 assert result.id == BEDSET_TEST_ID 

85 assert result.md5sum == "bbad0000000000000000000000000000" 

86 assert result.statistics is None 

87 assert result.plots is None 

88 assert result.bedfile_count == 1 

89 

90 def test_get_not_found(self, bbagent_obj): 

91 with ContextManagerDBTesting( 

92 config=bbagent_obj.config, add_data=True, bedset=True 

93 ): 

94 with pytest.raises(BedSetNotFoundError): 

95 bbagent_obj.bedset.get("not_uid", full=True) 

96 

97 def test_get_object(self, bbagent_obj): 

98 with ContextManagerDBTesting( 

99 config=bbagent_obj.config, add_data=True, bedset=True 

100 ): 

101 result = bbagent_obj.bedset.get_objects(BEDSET_TEST_ID) 

102 

103 assert len(result) == 1 

104 

105 def test_get_plots(self, bbagent_obj): 

106 with ContextManagerDBTesting( 

107 config=bbagent_obj.config, add_data=True, bedset=True 

108 ): 

109 result = bbagent_obj.bedset.get_plots(BEDSET_TEST_ID) 

110 

111 assert result is not None 

112 

113 def test_get_stats(self, bbagent_obj): 

114 with ContextManagerDBTesting( 

115 config=bbagent_obj.config, add_data=True, bedset=True 

116 ): 

117 result = bbagent_obj.bedset.get_statistics(BEDSET_TEST_ID) 

118 

119 assert result.sd is not None 

120 assert result.mean is not None 

121 

122 def test_get_bedset_list(self, bbagent_obj): 

123 with ContextManagerDBTesting( 

124 config=bbagent_obj.config, add_data=True, bedset=True 

125 ): 

126 result = bbagent_obj.bedset.get_ids_list(limit=100, offset=0) 

127 

128 assert result.count == 1 

129 assert result.limit == 100 

130 assert result.offset == 0 

131 assert len(result.results) == 1 

132 assert result.results[0].id == BEDSET_TEST_ID 

133 # bed_ids is intentionally left unpopulated in list results to 

134 # avoid lazy-loading full bedfile membership for every row 

135 assert result.results[0].bed_ids is None 

136 assert result.results[0].bedfile_count == 1 

137 

138 def test_get_bedset_list_offset(self, bbagent_obj): 

139 with ContextManagerDBTesting( 

140 config=bbagent_obj.config, add_data=True, bedset=True 

141 ): 

142 result = bbagent_obj.bedset.get_ids_list(limit=100, offset=1) 

143 

144 # assert result.count == 1 

145 assert result.limit == 100 

146 assert result.offset == 1 

147 assert len(result.results) == 0 

148 

149 def test_get_bedset_list_stable_order(self, bbagent_obj): 

150 extra_ids = [f"bedset_{i}" for i in range(9, 0, -1)] 

151 with ContextManagerDBTesting( 

152 config=bbagent_obj.config, add_data=True, bedset=True 

153 ): 

154 with Session(bbagent_obj.config.db_engine.engine) as session: 

155 for bedset_id in extra_ids: 

156 session.add( 

157 BedSets( 

158 id=bedset_id, 

159 name=bedset_id, 

160 description="random desc", 

161 md5sum="bbad0000000000000000000000000000", 

162 processed=False, 

163 bedfile_count=0, 

164 ) 

165 ) 

166 session.commit() 

167 

168 first = [r.id for r in bbagent_obj.bedset.get_ids_list(limit=100).results] 

169 second = [r.id for r in bbagent_obj.bedset.get_ids_list(limit=100).results] 

170 

171 # paged walk must cover every id exactly once 

172 paged = [] 

173 for offset in range(0, 10, 4): 

174 page = bbagent_obj.bedset.get_ids_list(limit=4, offset=offset) 

175 paged.extend(r.id for r in page.results) 

176 

177 assert first == second 

178 assert first == sorted(extra_ids + [BEDSET_TEST_ID]) 

179 assert paged == first 

180 

181 def test_get_idset_list_query_found(self, bbagent_obj): 

182 with ContextManagerDBTesting( 

183 config=bbagent_obj.config, add_data=True, bedset=True 

184 ): 

185 result = bbagent_obj.bedset.get_ids_list(query="rando", limit=100, offset=0) 

186 

187 assert result.count == 1 

188 assert result.limit == 100 

189 assert result.offset == 0 

190 assert len(result.results) == 1 

191 

192 def test_get_idset_list_query_fail(self, bbagent_obj): 

193 with ContextManagerDBTesting( 

194 config=bbagent_obj.config, add_data=True, bedset=True 

195 ): 

196 result = bbagent_obj.bedset.get_ids_list( 

197 query="rando1", limit=100, offset=0 

198 ) 

199 

200 assert result.count == 0 

201 assert result.limit == 100 

202 assert result.offset == 0 

203 assert len(result.results) == 0 

204 

205 def test_get_get_bedset_bedfiles(self, bbagent_obj): 

206 with ContextManagerDBTesting( 

207 config=bbagent_obj.config, add_data=True, bedset=True 

208 ): 

209 result = bbagent_obj.bedset.get_bedset_bedfiles(BEDSET_TEST_ID) 

210 

211 assert result.count == 1 

212 assert len(result.results) == 1 

213 

214 def test_delete(self, bbagent_obj, mocker): 

215 with ContextManagerDBTesting( 

216 config=bbagent_obj.config, add_data=True, bedset=True 

217 ): 

218 mocker.patch( 

219 "bbconf.config_parser.bedbaseconfig.BedBaseConfig.delete_s3", 

220 return_value=True, 

221 ) 

222 bbagent_obj.bedset.delete(BEDSET_TEST_ID) 

223 

224 assert not bbagent_obj.bedset.exists(BEDSET_TEST_ID) 

225 

226 def test_delete_none(self, bbagent_obj, mocker): 

227 with ContextManagerDBTesting( 

228 config=bbagent_obj.config, add_data=True, bedset=True 

229 ): 

230 mocker.patch( 

231 "bbconf.config_parser.bedbaseconfig.BedBaseConfig.delete_s3", 

232 return_value=True, 

233 ) 

234 bbagent_obj.bedset.delete(BEDSET_TEST_ID) 

235 

236 with pytest.raises(BedSetNotFoundError): 

237 bbagent_obj.bedset.delete(BEDSET_TEST_ID) 

238 

239 def test_delete_s3_error(self, bbagent_obj): 

240 with ContextManagerDBTesting( 

241 config=bbagent_obj.config, add_data=True, bedset=True 

242 ): 

243 with pytest.raises(BedbaseS3ConnectionError): 

244 bbagent_obj.bedset.delete(BEDSET_TEST_ID) 

245 

246 def test_retrieve_unprocessed(self, bbagent_obj): 

247 with ContextManagerDBTesting( 

248 config=bbagent_obj.config, add_data=True, bedset=True 

249 ): 

250 result = bbagent_obj.bedset.get_unprocessed() 

251 assert result.count == 1