Coverage for tests/test_schema.py: 100%

1import pytest 

2 

3from pepdbagent.models import UpdateSchemaRecordFields, UpdateSchemaVersionFields 

4 

5from .utils import PEPDBAgentContextManager 

6 

7DEFAULT_SCHEMA_VERSION = "1.0.0" 

8 

9 

10@pytest.mark.skipif( 

11 not PEPDBAgentContextManager().db_setup(), 

12 reason="DB is not setup", 

13) 

14class TestSchemas: 

15 @pytest.mark.parametrize( 

16 "namespace, name", 

17 [ 

18 ["namespace1", "2.0.0"], 

19 ], 

20 ) 

21 def test_get(self, namespace, name): 

22 with PEPDBAgentContextManager(add_schemas=True) as agent: 

23 assert agent.schema.schema_exist(namespace=namespace, name=name) 

24 assert agent.schema.version_exist( 

25 namespace=namespace, name=name, version=DEFAULT_SCHEMA_VERSION 

26 ) 

27 schema = agent.schema.get( 

28 namespace=namespace, name=name, version=DEFAULT_SCHEMA_VERSION 

29 ) 

30 assert schema 

31 

32 def test_update_schema(self): 

33 with PEPDBAgentContextManager(add_schemas=True) as agent: 

34 new_maintainers = "New Maintainer" 

35 new_lifecycle_stage = "New Stage" 

36 new_private = True 

37 new_name = "new_schema_name" 

38 

39 agent.schema.update_schema_record( 

40 "namespace1", 

41 "2.0.0", 

42 UpdateSchemaRecordFields( 

43 maintainers=new_maintainers, 

44 lifecycle_stage=new_lifecycle_stage, 

45 private=new_private, 

46 name=new_name, 

47 ), 

48 ) 

49 result = agent.schema.get_schema_info("namespace1", new_name) 

50 assert result.maintainers == new_maintainers 

51 assert result.lifecycle_stage == new_lifecycle_stage 

52 assert result.private == new_private 

53 assert result.schema_name == new_name 

54 

55 def test_update_schema_update_date(self): 

56 with PEPDBAgentContextManager(add_schemas=True) as agent: 

57 version_name = "2.0.0" 

58 version_schema = { 

59 "type": "object", 

60 "properties": { 

61 "name": {"type": "string"}, 

62 "age": {"type": "integer"}, 

63 }, 

64 } 

65 

66 first_time = agent.schema.get_schema_info( 

67 "namespace1", "2.0.0" 

68 ).last_update_date 

69 

70 agent.schema.add_version( 

71 "namespace1", 

72 version_name, 

73 "pablo1", 

74 schema_value=version_schema, 

75 contributors="Teddy", 

76 release_notes="Initial release", 

77 ) 

78 result = agent.schema.get_schema_info("namespace1", "2.0.0") 

79 assert result.last_update_date != first_time 

80 

81 def test_add_schema_version(self): 

82 with PEPDBAgentContextManager(add_schemas=True) as agent: 

83 new_schema = { 

84 "type": "object", 

85 "properties": { 

86 "name": {"type": "string"}, 

87 "age": {"type": "integer"}, 

88 }, 

89 } 

90 new_contributors = "New Maintainer" 

91 new_release_notes = "New release" 

92 agent.schema.update_schema_version( 

93 "namespace1", 

94 "2.0.0", 

95 DEFAULT_SCHEMA_VERSION, 

96 UpdateSchemaVersionFields( 

97 schema_value=new_schema, 

98 contributors=new_contributors, 

99 release_notes=new_release_notes, 

100 ), 

101 ) 

102 result = agent.schema.get_version_info( 

103 "namespace1", "2.0.0", DEFAULT_SCHEMA_VERSION 

104 ) 

105 assert result.contributors == new_contributors 

106 assert result.release_notes == new_release_notes 

107 assert ( 

108 agent.schema.get("namespace1", "2.0.0", DEFAULT_SCHEMA_VERSION) 

109 == new_schema 

110 ) 

111 

112 def test_search_schema_version(self): 

113 with PEPDBAgentContextManager(add_schemas=True) as agent: 

114 result = agent.schema.query_schema_version("namespace1", "2.0.0") 

115 assert result.pagination.total == 1 

116 assert len(result.results) == 1 

117 

118 def test_search_schema_version_with_tags(self): 

119 with PEPDBAgentContextManager(add_schemas=True) as agent: 

120 schema1 = { 

121 "type": "object", 

122 "properties": { 

123 "name1": {"type": "string"}, 

124 "age1": {"type": "integer"}, 

125 }, 

126 } 

127 schema2 = { 

128 "type": "object", 

129 "properties": { 

130 "name2": {"type": "string"}, 

131 "age2": {"type": "integer"}, 

132 }, 

133 } 

134 

135 agent.schema.add_version( 

136 "namespace1", 

137 "2.0.0", 

138 "bino1", 

139 schema_value=schema1, 

140 tags=["tag1", "bioinfo"], 

141 release_notes="computer change", 

142 ) 

143 agent.schema.add_version( 

144 "namespace1", 

145 "2.0.0", 

146 "bino2", 

147 schema_value=schema2, 

148 tags=["bioinfo"], 

149 release_notes="language", 

150 ) 

151 

152 result = agent.schema.query_schema_version( 

153 "namespace1", "2.0.0", tag="tag1" 

154 ) 

155 

156 assert result.pagination.total == 1 

157 assert len(result.results) == 1 

158 

159 result = agent.schema.query_schema_version( 

160 "namespace1", "2.0.0", tag="bioinfo" 

161 ) 

162 

163 assert result.pagination.total == 2 

164 assert len(result.results) == 2 

165 

166 def test_search_schema(self): 

167 with PEPDBAgentContextManager(add_schemas=True) as agent: 

168 result = agent.schema.query_schemas(search_str="bed") 

169 assert result.pagination.total == 3 

170 assert len(result.results) == 3 

171 

172 def test_search_schema_namespace(self): 

173 with PEPDBAgentContextManager(add_schemas=True) as agent: 

174 result = agent.schema.query_schemas("namespace1") 

175 assert result.pagination.total == 2 

176 assert "namespace1" in [f.namespace for f in result.results] 

177 assert "2.0.0" in [f.schema_name for f in result.results] 

178 

179 def test_search_schema_page_number(self): 

180 with PEPDBAgentContextManager(add_schemas=True) as agent: 

181 result = agent.schema.query_schemas("namespace2", page_size=2, page=1) 

182 assert result.pagination.total == 3 

183 assert result.pagination.page == 1 

184 assert result.pagination.page_size == 2 

185 assert len(result.results) == 1 

186 

187 @pytest.mark.parametrize( 

188 "namespace, name", 

189 [ 

190 ["namespace2", "bedmaker"], 

191 ], 

192 ) 

193 def test_schema_delete(self, namespace, name): 

194 with PEPDBAgentContextManager(add_schemas=True) as agent: 

195 assert agent.schema.schema_exist(namespace=namespace, name=name) 

196 agent.schema.delete_schema(namespace=namespace, name=name) 

197 assert not agent.schema.version_exist( 

198 namespace=namespace, name=name, version=DEFAULT_SCHEMA_VERSION 

199 ) 

200 assert not agent.schema.schema_exist(namespace=namespace, name=name) 

201 

202 @pytest.mark.parametrize( 

203 "namespace, name", 

204 [ 

205 ["namespace2", "bedmaker"], 

206 ], 

207 ) 

208 def test_schema_version_delete(self, namespace, name): 

209 with PEPDBAgentContextManager(add_schemas=True) as agent: 

210 assert agent.schema.version_exist( 

211 namespace=namespace, name=name, version=DEFAULT_SCHEMA_VERSION 

212 ) 

213 agent.schema.delete_version( 

214 namespace=namespace, name=name, version=DEFAULT_SCHEMA_VERSION 

215 ) 

216 assert not agent.schema.version_exist( 

217 namespace=namespace, name=name, version=DEFAULT_SCHEMA_VERSION 

218 ) 

219 assert agent.schema.schema_exist(namespace=namespace, name=name) 

220 

221 def test_number_of_schemas_in_namespace(self): 

222 with PEPDBAgentContextManager(add_schemas=True) as agent: 

223 for k in agent.namespace.info().results: 

224 if k.namespace_name == "namespace1": 

225 assert k.number_of_schemas == 2 

226 if k.namespace_name == "namespace2": 

227 assert k.number_of_schemas == 3 

228 

229 

230class TestSchemaTags: 

231 def test_insert_tags(self): 

232 with PEPDBAgentContextManager(add_schemas=True) as agent: 

233 new_tag1 = "new_tag" 

234 new_tag2 = "tag2" 

235 agent.schema.add_tag_to_schema( 

236 "namespace1", "2.0.0", DEFAULT_SCHEMA_VERSION, tag=[new_tag1, new_tag2] 

237 ) 

238 

239 result = agent.schema.get_version_info( 

240 "namespace1", "2.0.0", DEFAULT_SCHEMA_VERSION 

241 ) 

242 

243 assert new_tag1 in result.tags 

244 assert new_tag2 in result.tags 

245 

246 def test_insert_one_tag(self): 

247 with PEPDBAgentContextManager(add_schemas=True) as agent: 

248 new_tag1 = "new_tag" 

249 agent.schema.add_tag_to_schema( 

250 "namespace1", "2.0.0", DEFAULT_SCHEMA_VERSION, tag=new_tag1 

251 ) 

252 result = agent.schema.get_version_info( 

253 "namespace1", "2.0.0", DEFAULT_SCHEMA_VERSION 

254 ) 

255 assert new_tag1 in result.tags 

256 

257 @pytest.mark.parametrize( 

258 "namespace, name", 

259 [ 

260 ["namespace2", "bedmaker"], 

261 ], 

262 ) 

263 def test_delete_tag(self, namespace, name): 

264 with PEPDBAgentContextManager(add_schemas=True) as agent: 

265 new_tag1 = "new_tag" 

266 agent.schema.add_tag_to_schema( 

267 namespace, name, DEFAULT_SCHEMA_VERSION, tag=new_tag1 

268 ) 

269 result = agent.schema.get_version_info( 

270 namespace, name, DEFAULT_SCHEMA_VERSION 

271 ) 

272 assert new_tag1 in result.tags 

273 agent.schema.remove_tag_from_schema( 

274 namespace, name, DEFAULT_SCHEMA_VERSION, tag=new_tag1 

275 ) 

276 result = agent.schema.get_version_info( 

277 namespace, name, DEFAULT_SCHEMA_VERSION 

278 ) 

279 assert new_tag1 not in result.tags