Coverage for tests/test_bedset.py: 25%

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 upload_pephub=False, 

57 no_fail=True, 

58 ) 

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

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

61 assert result is not None 

62 assert result.name == "test_name" 

63 assert len([k for k in result.files]) == 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 

77 def test_get_metadata_not_full(self, bbagent_obj): 

78 with ContextManagerDBTesting( 

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

80 ): 

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

82 

83 assert result.id == BEDSET_TEST_ID 

84 assert result.md5sum == "bbad0000000000000000000000000000" 

85 assert result.statistics is None 

86 assert result.plots is None 

87 

88 def test_get_not_found(self, bbagent_obj): 

89 with ContextManagerDBTesting( 

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

91 ): 

92 with pytest.raises(BedSetNotFoundError): 

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

94 

95 def test_get_object(self, bbagent_obj): 

96 with ContextManagerDBTesting( 

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

98 ): 

99 result = bbagent_obj.bedset.get_objects(BEDSET_TEST_ID) 

100 

101 assert len(result) == 1 

102 

103 def test_get_plots(self, bbagent_obj): 

104 with ContextManagerDBTesting( 

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

106 ): 

107 result = bbagent_obj.bedset.get_plots(BEDSET_TEST_ID) 

108 

109 assert result is not None 

110 

111 def test_get_stats(self, bbagent_obj): 

112 with ContextManagerDBTesting( 

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

114 ): 

115 result = bbagent_obj.bedset.get_statistics(BEDSET_TEST_ID) 

116 

117 assert result.sd is not None 

118 assert result.mean is not None 

119 

120 def test_get_bedset_list(self, bbagent_obj): 

121 with ContextManagerDBTesting( 

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

123 ): 

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

125 

126 assert result.count == 1 

127 assert result.limit == 100 

128 assert result.offset == 0 

129 assert len(result.results) == 1 

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

131 

132 def test_get_bedset_list_offset(self, bbagent_obj): 

133 with ContextManagerDBTesting( 

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

135 ): 

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

137 

138 # assert result.count == 1 

139 assert result.limit == 100 

140 assert result.offset == 1 

141 assert len(result.results) == 0 

142 

143 def test_get_idset_list_query_found(self, bbagent_obj): 

144 with ContextManagerDBTesting( 

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

146 ): 

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

148 

149 assert result.count == 1 

150 assert result.limit == 100 

151 assert result.offset == 0 

152 assert len(result.results) == 1 

153 

154 def test_get_idset_list_query_fail(self, bbagent_obj): 

155 with ContextManagerDBTesting( 

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

157 ): 

158 result = bbagent_obj.bedset.get_ids_list( 

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

160 ) 

161 

162 assert result.count == 0 

163 assert result.limit == 100 

164 assert result.offset == 0 

165 assert len(result.results) == 0 

166 

167 def test_get_get_bedset_bedfiles(self, bbagent_obj): 

168 with ContextManagerDBTesting( 

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

170 ): 

171 result = bbagent_obj.bedset.get_bedset_bedfiles(BEDSET_TEST_ID) 

172 

173 assert result.count == 1 

174 assert len(result.results) == 1 

175 

176 def test_delete(self, bbagent_obj, mocker): 

177 with ContextManagerDBTesting( 

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

179 ): 

180 mocker.patch( 

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

182 return_value=True, 

183 ) 

184 bbagent_obj.bedset.delete(BEDSET_TEST_ID) 

185 

186 assert not bbagent_obj.bedset.exists(BEDSET_TEST_ID) 

187 

188 def test_delete_none(self, bbagent_obj, mocker): 

189 with ContextManagerDBTesting( 

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

191 ): 

192 mocker.patch( 

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

194 return_value=True, 

195 ) 

196 bbagent_obj.bedset.delete(BEDSET_TEST_ID) 

197 

198 with pytest.raises(BedSetNotFoundError): 

199 bbagent_obj.bedset.delete(BEDSET_TEST_ID) 

200 

201 def test_delete_s3_error(self, bbagent_obj): 

202 with ContextManagerDBTesting( 

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

204 ): 

205 with pytest.raises(BedbaseS3ConnectionError): 

206 bbagent_obj.bedset.delete(BEDSET_TEST_ID)