Coverage for tests/test_bedfile.py: 25%

1import pytest 

2from sqlalchemy.orm import Session 

3from sqlalchemy.sql import select 

4 

5from bbconf.bbagent import BedBaseAgent 

6from bbconf.db_utils import Bed, Files 

7from bbconf.exceptions import BedFIleExistsError, BEDFileNotFoundError 

8from bbconf.const import DEFAULT_LICENSE 

9 

10from .conftest import SERVICE_UNAVAILABLE, get_bbagent 

11from .utils import BED_TEST_ID, ContextManagerDBTesting 

12 

13 

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

15def test_bb_database(): 

16 agent = get_bbagent() 

17 assert isinstance(agent, BedBaseAgent) 

18 

19 

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

21class Test_BedFile_Agent: 

22 def test_upload(self, bbagent_obj, example_dict, mocker): 

23 upload_s3_mock = mocker.patch( 

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

25 return_value=True, 

26 ) 

27 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): 

28 bbagent_obj.bed.add(**example_dict) 

29 

30 assert upload_s3_mock.called 

31 assert bbagent_obj.bed.exists(example_dict["identifier"]) 

32 

33 def test_upload_exists(self, bbagent_obj, example_dict, mocker): 

34 mocker.patch( 

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

36 return_value=True, 

37 ) 

38 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): 

39 bbagent_obj.bed.add(**example_dict) 

40 with pytest.raises(BedFIleExistsError): 

41 bbagent_obj.bed.add(**example_dict) 

42 

43 def test_add_nofail(self, bbagent_obj, example_dict, mocker): 

44 mocker.patch( 

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

46 return_value=True, 

47 ) 

48 

49 example_dict["nofail"] = True 

50 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): 

51 bbagent_obj.bed.add(**example_dict) 

52 bbagent_obj.bed.add(**example_dict) 

53 assert bbagent_obj.bed.exists(example_dict["identifier"]) 

54 

55 def test_get_all(self, bbagent_obj, mocked_phc): 

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

57 return_result = bbagent_obj.bed.get(BED_TEST_ID, full=True) 

58 assert return_result is not None 

59 assert return_result.files is not None 

60 assert return_result.plots is not None 

61 assert return_result.raw_metadata is not None 

62 

63 assert return_result.genome_alias == "hg38" 

64 assert return_result.stats.number_of_regions == 1 

65 

66 assert return_result.files.bed_file is not None 

67 assert return_result.plots.chrombins is not None 

68 assert return_result.license_id == DEFAULT_LICENSE 

69 

70 def test_get_all_not_found(self, bbagent_obj): 

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

72 return_result = bbagent_obj.bed.get(BED_TEST_ID, full=False) 

73 

74 assert return_result is not None 

75 assert return_result.files is None 

76 assert return_result.plots is None 

77 assert return_result.raw_metadata is None 

78 assert return_result.stats is None 

79 

80 assert return_result.genome_alias == "hg38" 

81 assert return_result.id == BED_TEST_ID 

82 

83 def test_get_raw_metadata(self, bbagent_obj, mocked_phc): 

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

85 return_result = bbagent_obj.bed.get_raw_metadata(BED_TEST_ID) 

86 

87 assert return_result is not None 

88 assert return_result.sample_name == BED_TEST_ID 

89 

90 def test_get_stats(self, bbagent_obj): 

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

92 return_result = bbagent_obj.bed.get_stats(BED_TEST_ID) 

93 

94 assert return_result is not None 

95 assert return_result.number_of_regions == 1 

96 

97 def test_get_files(self, bbagent_obj): 

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

99 return_result = bbagent_obj.bed.get_files(BED_TEST_ID) 

100 

101 assert return_result is not None 

102 assert return_result.bed_file.path is not None 

103 

104 def test_get_plots(self, bbagent_obj): 

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

106 return_result = bbagent_obj.bed.get_plots(BED_TEST_ID) 

107 

108 assert return_result is not None 

109 assert return_result.chrombins is not None 

110 

111 def test_get_objects(self, bbagent_obj): 

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

113 return_result = bbagent_obj.bed.get_objects(BED_TEST_ID) 

114 

115 assert "bed_file" in return_result 

116 assert "chrombins" in return_result 

117 

118 def test_get_classification(self, bbagent_obj): 

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

120 return_result = bbagent_obj.bed.get_classification(BED_TEST_ID) 

121 

122 assert return_result is not None 

123 assert return_result.bed_type == "bed6+4" 

124 

125 def test_get_list(self, bbagent_obj): 

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

127 return_result = bbagent_obj.bed.get_ids_list(limit=100, offset=0) 

128 

129 assert len(return_result.results) == 1 

130 assert return_result.count == 1 

131 assert return_result.results[0].id == BED_TEST_ID 

132 assert return_result.limit == 100 

133 assert return_result.offset == 0 

134 

135 def test_get_list_genome_true(self, bbagent_obj): 

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

137 return_result = bbagent_obj.bed.get_ids_list( 

138 limit=100, offset=0, genome="hg38" 

139 ) 

140 

141 assert len(return_result.results) == 1 

142 assert return_result.count == 1 

143 assert return_result.results[0].id == BED_TEST_ID 

144 assert return_result.limit == 100 

145 assert return_result.offset == 0 

146 

147 def test_get_list_genome_false(self, bbagent_obj): 

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

149 return_result = bbagent_obj.bed.get_ids_list( 

150 limit=100, offset=0, genome="hg381" 

151 ) 

152 

153 assert len(return_result.results) == 0 

154 assert return_result.count == 0 

155 

156 def test_get_list_bed_type_true(self, bbagent_obj): 

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

158 return_result = bbagent_obj.bed.get_ids_list( 

159 limit=100, offset=0, bed_type="bed6+4" 

160 ) 

161 

162 assert len(return_result.results) == 1 

163 assert return_result.count == 1 

164 assert return_result.results[0].id == BED_TEST_ID 

165 assert return_result.limit == 100 

166 assert return_result.offset == 0 

167 

168 def test_get_list_bed_type_false(self, bbagent_obj): 

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

170 return_result = bbagent_obj.bed.get_ids_list( 

171 limit=100, offset=0, bed_type="bed6+5" 

172 ) 

173 

174 assert len(return_result.results) == 0 

175 assert return_result.count == 0 

176 

177 def test_get_list_bed_offset(self, bbagent_obj): 

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

179 return_result = bbagent_obj.bed.get_ids_list( 

180 limit=100, 

181 offset=1, 

182 ) 

183 

184 assert len(return_result.results) == 0 

185 assert return_result.count == 1 

186 assert return_result.offset == 1 

187 

188 def test_bed_delete(self, bbagent_obj, mocker): 

189 mocker.patch("bbconf.config_parser.bedbaseconfig.BedBaseConfig.delete_s3") 

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

191 bbagent_obj.bed.delete(BED_TEST_ID) 

192 

193 assert not bbagent_obj.bed.exists(BED_TEST_ID) 

194 

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

196 result = session.scalar(select(Bed).where(Bed.id == BED_TEST_ID)) 

197 assert result is None 

198 

199 result = session.scalars(select(Files)) 

200 assert len([k for k in result]) == 0 

201 

202 def test_bed_delete_not_found(self, bbagent_obj): 

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

204 with pytest.raises(BEDFileNotFoundError): 

205 bbagent_obj.bed.delete("not_found") 

206 

207 @pytest.mark.skip("Skipped, not fully implemented") 

208 def test_bed_update(self): 

209 # agent = BedBaseAgent(config=config) 

210 # ff = agent.bed.update("91b2754c8ff01769bacfc80e6923c46e", {"number_of_regions": 44}) 

211 # print(ff) 

212 # assert ff != None 

213 pass 

214 

215 

216@pytest.mark.skip("Skipped, because ML models and qdrant needed") 

217class TestVectorSearch: 

218 def test_qdrant_search(self, bbagent_obj, mocker): 

219 mocker.patch( 

220 "geniml.text2bednn.text2bednn.Text2BEDSearchInterface.nl_vec_search", 

221 return_value={ 

222 "id": BED_TEST_ID, 

223 "payload": {"bed_id": "39b686ec08206b92b540ed434266ec9b"}, 

224 "score": 0.2146723, 

225 }, 

226 ) 

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

228 return_result = bbagent_obj.bed.text_to_bed_search("something") 

229 assert return_result 

230 

231 def test_delete_qdrant_point(self, bbagent_obj): 

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

233 bbagent_obj.bed.delete_qdrant_point(BED_TEST_ID) 

234 

235 def test_create_qdrant_collection(self): 

236 agent = BedBaseAgent( 

237 config="/home/bnt4me/virginia/repos/bbuploader/config_db_local.yaml" 

238 ) 

239 ff = agent.bed.create_qdrant_collection() 

240 ff 

241 assert True