Coverage for tests/utils.py: 90%

1from typing import Union 

2 

3from sqlalchemy.orm import Session 

4 

5from bbconf.config_parser.bedbaseconfig import BedBaseConfig 

6from bbconf.db_utils import Bed, BedFileBedSetRelation, BedSets, BedStats, Files 

7 

8BED_TEST_ID = "bbad85f21962bb8d972444f7f9a3a932" 

9BEDSET_TEST_ID = "test_bedset_id" 

10 

11 

12stats = { 

13 "id": BED_TEST_ID, 

14 "number_of_regions": 1, 

15 "median_tss_dist": 2, 

16 "mean_region_width": 3, 

17 "exon_frequency": 4, 

18 "exon_percentage": 5, 

19 "intron_frequency": 6, 

20 "intron_percentage": 7, 

21 "intergenic_percentage": 8, 

22 "intergenic_frequency": 9, 

23 "promotercore_frequency": 10, 

24 "promotercore_percentage": 11, 

25 "fiveutr_frequency": 12, 

26 "fiveutr_percentage": 13, 

27 "threeutr_frequency": 14, 

28 "threeutr_percentage": 15, 

29 "promoterprox_frequency": 16, 

30 "promoterprox_percentage": 17, 

31} 

32 

33 

34def get_example_dict() -> dict: 

35 value = { 

36 "id": BED_TEST_ID, 

37 "data_format": "encode_narrowpeak", 

38 "bed_compliance": "bed6+4", 

39 "genome_alias": "hg38", 

40 "genome_digest": "2230c535660fb4774114bfa966a62f823fdb6d21acf138d4", 

41 "name": "random_name", 

42 "processed": False, 

43 } 

44 return value 

45 

46 

47def get_bedset_files() -> dict: 

48 return { 

49 "title": "region_commonality", 

50 "name": "region_commonality", 

51 "path": "data/files/bbad85f21962bb8d972444f7f9a3a932.bed.gz", 

52 "description": "Bfffffff", 

53 "bedset_id": BEDSET_TEST_ID, 

54 } 

55 

56 

57def get_files() -> dict: 

58 return { 

59 "title": "Bed file", 

60 "name": "bed_file", 

61 "path": "data/files/bbad85f21962bb8d972444f7f9a3a932.bed.gz", 

62 "description": "Bed file with regions", 

63 "bedfile_id": BED_TEST_ID, 

64 } 

65 

66 

67def get_plots() -> dict: 

68 return { 

69 "name": "chrombins", 

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

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

72 "path": "data/plots/bbad85f21962bb8d972444f7f9a3a932_chrombins.pdf", 

73 "path_thumbnail": "data/plots/bbad85f21962bb8d972444f7f9a3a932_chrombins.png", 

74 "bedfile_id": BED_TEST_ID, 

75 } 

76 

77 

78class ContextManagerDBTesting: 

79 """ 

80 Creates context manager to connect to database at db_url adds data and drop everything from the database upon exit to ensure 

81 the db is empty for each new test. 

82 """ 

83 

84 def __init__( 

85 self, 

86 config: Union[str, BedBaseConfig], 

87 add_data: bool = False, 

88 bedset: bool = False, 

89 ): 

90 """ 

91 :param config: config object 

92 :param add_data: add data to the database 

93 :param bedset: add bedset data to the database 

94 """ 

95 if isinstance(config, BedBaseConfig): 

96 self.config = config 

97 else: 

98 self.config = BedBaseConfig(config) 

99 

100 self.add_data = add_data 

101 self.bedset = bedset 

102 

103 self.db_engine = self.config.db_engine 

104 self.db_engine.create_schema() 

105 

106 def __enter__(self): 

107 if self.add_data: 

108 self._add_data() 

109 if self.bedset: 

110 self._add_bedset_data() 

111 

112 def __exit__(self, exc_type, exc_value, exc_traceback): 

113 # If we want to keep data, and schema, comment out the following line 

114 self.db_engine.delete_schema() 

115 pass 

116 

117 def _add_data(self): 

118 with Session(self.db_engine.engine) as session: 

119 new_bed = Bed(**get_example_dict()) 

120 new_files = Files(**get_files()) 

121 new_plots = Files(**get_plots()) 

122 new_stats = BedStats(**stats) 

123 

124 session.add(new_bed) 

125 session.add(new_files) 

126 session.add(new_plots) 

127 session.add(new_stats) 

128 

129 session.commit() 

130 

131 def _add_bedset_data(self): 

132 with Session(self.db_engine.engine) as session: 

133 new_bedset = BedSets( 

134 id=BEDSET_TEST_ID, 

135 name=BEDSET_TEST_ID, 

136 description="random desc", 

137 bedset_means=stats, 

138 bedset_standard_deviation=stats, 

139 md5sum="bbad0000000000000000000000000000", 

140 processed=False, 

141 bedfile_count=1, 

142 ) 

143 new_bed_bedset = BedFileBedSetRelation( 

144 bedfile_id=BED_TEST_ID, 

145 bedset_id=BEDSET_TEST_ID, 

146 ) 

147 new_files = Files(**get_bedset_files()) 

148 

149 session.add(new_bedset) 

150 session.add(new_bed_bedset) 

151 session.add(new_files) 

152 

153 session.commit() 

154 

155 def _add_universe(self): 

156 from bbconf.db_utils import Universes 

157 

158 with Session(self.db_engine.engine) as session: 

159 new_univ = Universes( 

160 id=BED_TEST_ID, 

161 ) 

162 session.add(new_univ) 

163 session.commit()