Coverage for test / idm_ror_test.py: 99%

1# SPDX-FileCopyrightText: 2024 Elia Rizzetto <elia.rizzetto2@unibo.it> 

2# SPDX-FileCopyrightText: 2026 Marta Soricetti <marta.soricetti@unibo.it> 

3# 

4# SPDX-License-Identifier: ISC 

5 

6import json 

7import sqlite3 

8import os.path 

9import unittest 

10from os import makedirs 

11from os.path import exists, join 

12 

13from oc_ds_converter.oc_idmanager import * 

14from oc_ds_converter.oc_idmanager.ror import RORManager 

15from oc_ds_converter.oc_idmanager.oc_data_storage.sqlite_manager import SqliteStorageManager 

16from oc_ds_converter.oc_idmanager.oc_data_storage.in_memory_manager import InMemoryStorageManager 

17from oc_ds_converter.oc_idmanager.oc_data_storage.redis_manager import RedisStorageManager 

18 

19class RORIdentifierManagerTest(unittest.TestCase): 

20 """This class aim at testing ror identifiers manager.""" 

21 

22 def setUp(self): 

23 if not exists("tmp"): 

24 makedirs("tmp") 

25 

26 self.test_dir = os.path.join("test","data") 

27 self.test_json_path = join(self.test_dir, "glob.json") 

28 with open(self.test_json_path, encoding="utf-8") as fp: 

29 self.data = json.load(fp) 

30 

31 self.valid_ror_1 = "https://ror.org/040jc3p57" 

32 self.valid_ror_2 = "01111rn36" 

33 self.invalid_ror_1 = "la673822" 

34 self.invalid_ror_2 = ".org/560jc3p57" 

35 

36 

37 def test_ror_normalise(self): 

38 rm = RORManager() 

39 self.assertEqual( 

40 rm.normalise(self.valid_ror_1), 

41 rm.normalise(self.valid_ror_1.replace("https://", "")), 

42 ) 

43 self.assertEqual( 

44 rm.normalise(self.valid_ror_2), 

45 rm.normalise("https://ror.org/" + self.valid_ror_2), 

46 ) 

47 

48 def test_ror_is_valid(self): 

49 rm = RORManager() 

50 self.assertTrue(rm.is_valid(self.valid_ror_1)) 

51 self.assertTrue(rm.is_valid(self.valid_ror_2)) 

52 self.assertFalse(rm.is_valid(self.invalid_ror_1)) 

53 self.assertFalse(rm.is_valid(self.invalid_ror_2)) 

54 

55 rm_file = RORManager(storage_manager=InMemoryStorageManager(self.test_json_path)) 

56 self.assertTrue(rm_file.normalise(self.valid_ror_1, include_prefix=True) in self.data) 

57 self.assertTrue(rm_file.normalise(self.valid_ror_2, include_prefix=True) in self.data) 

58 

59 self.assertTrue(rm_file.is_valid(rm_file.normalise(self.valid_ror_1, include_prefix=True))) 

60 self.assertTrue(rm_file.is_valid(rm_file.normalise(self.valid_ror_2, include_prefix=True))) 

61 self.assertFalse(rm_file.is_valid(rm_file.normalise(self.invalid_ror_2, include_prefix=True))) 

62 

63 rm_nofile_noapi = RORManager(storage_manager=InMemoryStorageManager(self.test_json_path), use_api_service=False) 

64 self.assertTrue(rm_nofile_noapi.is_valid(self.valid_ror_1)) 

65 self.assertFalse(rm_nofile_noapi.is_valid(self.invalid_ror_1)) 

66 

67 def test_ror_exists(self): 

68 with self.subTest(msg="get_extra_info = True, allow_extra_api=None"): 

69 jm = RORManager() 

70 output = jm.exists(self.valid_ror_1, get_extra_info=True, allow_extra_api=None) 

71 expected_output = (True, {"valid": True}) 

72 self.assertEqual(output, expected_output) 

73 with self.subTest(msg="get_extra_info = True, allow_extra_api=None"): 

74 jm = RORManager() 

75 output = jm.exists(self.valid_ror_2, get_extra_info=True, allow_extra_api=None) 

76 expected_output = (True, {"valid": True}) 

77 self.assertEqual(output, expected_output) 

78 with self.subTest(msg="get_extra_info = True, allow_extra_api=None"): 

79 jm = RORManager() 

80 output = jm.exists(self.invalid_ror_1, get_extra_info=True, allow_extra_api=None) 

81 expected_output = (False, {"valid": False}) 

82 self.assertEqual(output, expected_output) 

83 

84 def test_ror_default(self): 

85 jm_nofile = RORManager() 

86 # No support files (it generates it) 

87 # Default storage manager : in Memory + generates file on method call (not automatically) 

88 # uses API 

89 self.assertTrue(jm_nofile.is_valid(self.valid_ror_1)) 

90 self.assertTrue(jm_nofile.is_valid(self.valid_ror_2)) 

91 

92 self.assertFalse(jm_nofile.is_valid(self.invalid_ror_1)) 

93 self.assertFalse(jm_nofile.is_valid(self.invalid_ror_2)) 

94 jm_nofile.storage_manager.store_file() 

95 validated_ids = [self.valid_ror_1, self.valid_ror_2, self.invalid_ror_1, self.invalid_ror_2] 

96 # check that the support file was correctly created 

97 self.assertTrue(os.path.exists("storage/id_value.json")) 

98 lj = open("storage/id_value.json") 

99 load_dict = json.load(lj) 

100 lj.close() 

101 # check that all the validated ids are stored in the json file 

102 self.assertTrue(all(jm_nofile.normalise(x, include_prefix=True) in load_dict for x in validated_ids)) 

103 jm_nofile.storage_manager.delete_storage() 

104 # check that the support file was correctly deleted 

105 self.assertFalse(os.path.exists("storage/id_value.json")) 

106 

107 ##### IN MEMORY STORAGE MANAGER 

108 

109 def test_ror_memory_file_noapi(self): 

110 # Uses support file (without updating it) 

111 # Uses InMemoryStorageManager storage manager 

112 # does not use API (so a syntactically correct id is considered to be valid) 

113 rm_file = RORManager(storage_manager=InMemoryStorageManager(self.test_json_path), use_api_service=False) 

114 self.assertTrue(rm_file.normalise(self.valid_ror_1, include_prefix=True) in self.data) 

115 self.assertTrue(rm_file.normalise(self.valid_ror_2, include_prefix=True) in self.data) 

116 self.assertTrue(rm_file.is_valid(self.valid_ror_1)) 

117 self.assertFalse(rm_file.is_valid(self.invalid_ror_1)) # is stored in support file as invalid 

118 self.assertTrue(rm_file.is_valid( 

119 "ror:03ztgj036")) # is not stored in support file as invalid, does not exist but has correct syntax 

120 

121 def test_ror_memory_file_api(self): 

122 # Uses support file (without updating it) 

123 # Uses InMemoryStorageManager storage manager 

124 # uses API (so a syntactically correct id which is not valid is considered to be invalid) 

125 rm_file = RORManager(storage_manager=InMemoryStorageManager(self.test_json_path), use_api_service=True) 

126 self.assertFalse(rm_file.is_valid("ror:03ztgj036")) 

127 

128 def test_ror_memory_nofile_noapi(self): 

129 # Does not use support file 

130 # Uses InMemoryStorageManager storage manager 

131 # Does not use API (so a syntactically correct id which is not valid is considered to be valid) 

132 rm_nofile_noapi = RORManager(storage_manager=InMemoryStorageManager(), use_api_service=False) 

133 self.assertTrue(rm_nofile_noapi.is_valid(self.valid_ror_1)) 

134 self.assertTrue(rm_nofile_noapi.is_valid("ror:03ztgj036")) 

135 rm_nofile_noapi.storage_manager.delete_storage() 

136 

137 #### SQLITE STORAGE MANAGER 

138 

139 def test_ror_sqlite_nofile_api(self): 

140 # No support files (it generates it) 

141 # Uses SqliteStorageManager 

142 # uses API 

143 sql_rm_nofile = RORManager(storage_manager=SqliteStorageManager()) 

144 self.assertTrue(sql_rm_nofile.is_valid(self.valid_ror_1)) 

145 self.assertTrue(sql_rm_nofile.is_valid(self.valid_ror_2)) 

146 

147 self.assertFalse(sql_rm_nofile.is_valid(self.invalid_ror_1)) 

148 self.assertFalse(sql_rm_nofile.is_valid(self.invalid_ror_2)) 

149 

150 # check that the support db was correctly created and that it contains all the validated ids 

151 self.assertTrue(os.path.exists("storage/id_valid_dict.db")) 

152 validated_ids = [self.valid_ror_1, self.valid_ror_2, self.invalid_ror_1, self.invalid_ror_2] 

153 all_ids_stored = sql_rm_nofile.storage_manager.get_all_keys() 

154 # check that all the validated ids are stored in the json file 

155 self.assertTrue(all(sql_rm_nofile.normalise(x, include_prefix=True) in all_ids_stored for x in validated_ids)) 

156 sql_rm_nofile.storage_manager.delete_storage() 

157 # check that the support file was correctly deleted 

158 self.assertFalse(os.path.exists("storage/id_valid_dict.db")) 

159 

160 def test_ror_sqlite_file_api(self): 

161 # Uses support file 

162 # Uses SqliteStorageManager 

163 # does not use API (so a syntactically correct id is considered to be valid) 

164 # db creation 

165 test_sqlite_db = os.path.join(self.test_dir, "database.db") 

166 if os.path.exists(test_sqlite_db): 

167 os.remove(test_sqlite_db) 

168 to_insert = [self.invalid_ror_1, self.valid_ror_1, self.valid_ror_2] 

169 sql_file = RORManager(storage_manager=SqliteStorageManager(test_sqlite_db), use_api_service=True) 

170 for id in to_insert: 

171 norm_id = sql_file.normalise(id, include_prefix=True) 

172 is_valid = 1 if sql_file.is_valid(norm_id) else 0 

173 insert_tup = (norm_id, is_valid) 

174 sql_file.storage_manager.cur.execute(f"INSERT OR REPLACE INTO info VALUES (?,?)", insert_tup) 

175 sql_file.storage_manager.con.commit() 

176 sql_file.storage_manager.con.close() 

177 

178 sql_no_api = RORManager(storage_manager=SqliteStorageManager(test_sqlite_db), use_api_service=False) 

179 all_db_keys = sql_no_api.storage_manager.get_all_keys() 

180 # check that all the normalised ind in the list were correctly inserted in the db 

181 self.assertTrue(all(sql_no_api.normalise(x, include_prefix=True) in all_db_keys for x in to_insert)) 

182 self.assertTrue(sql_no_api.is_valid(self.valid_ror_1)) # is stored in support file as valid 

183 self.assertTrue(sql_no_api.is_valid(self.valid_ror_2)) # is stored in support file as valid 

184 self.assertFalse(sql_no_api.is_valid(self.invalid_ror_1)) # is stored in support file as invalid 

185 self.assertTrue(sql_no_api.is_valid( 

186 "ror:03ztgj036")) # is not stored in support file as invalid, does not exist but has correct syntax 

187 sql_no_api.storage_manager.delete_storage() 

188 

189 def test_ror_sqlite_nofile_noapi(self): 

190 # Does not use support file 

191 # Uses SqliteStorageManager 

192 # Does not use API (so a syntactically correct id which is not valid is considered to be valid) 

193 rm_nofile_noapi = RORManager(storage_manager=SqliteStorageManager(), use_api_service=False) 

194 self.assertTrue(rm_nofile_noapi.is_valid(self.valid_ror_1)) 

195 self.assertTrue(rm_nofile_noapi.is_valid("ror:03ztgj036")) 

196 rm_nofile_noapi.storage_manager.delete_storage() 

197 

198 #### REDIS STORAGE MANAGER 

199 def test_ror_redis_nofile_api(self): 

200 # No available data in redis db 

201 # Storage manager : RedisStorageManager 

202 # uses API 

203 rm_nofile = RORManager(storage_manager=RedisStorageManager(testing=True)) 

204 self.assertTrue(rm_nofile.is_valid(self.valid_ror_1)) 

205 self.assertTrue(rm_nofile.is_valid(self.valid_ror_2)) 

206 

207 self.assertFalse(rm_nofile.is_valid(self.invalid_ror_1)) 

208 self.assertFalse(rm_nofile.is_valid(self.invalid_ror_2)) 

209 # check that the redis db was correctly filled and that it contains all the validated ids 

210 

211 validated_ids = {self.valid_ror_1, self.valid_ror_2, self.invalid_ror_1, self.invalid_ror_2} 

212 validated_ids = {rm_nofile.normalise(x, include_prefix=True) for x in validated_ids} 

213 all_ids_stored = rm_nofile.storage_manager.get_all_keys() 

214 # check that all the validated ids are stored in the json file 

215 self.assertEqual(validated_ids, all_ids_stored) 

216 rm_nofile.storage_manager.delete_storage() 

217 # check that the support file was correctly deleted 

218 self.assertEqual(rm_nofile.storage_manager.get_all_keys(), set()) 

219 

220 def test_ror_redis_file_api(self): 

221 # Uses data in redis db 

222 # Uses RedisStorageManager 

223 # fills db 

224 

225 # use API to save validity values 

226 to_insert = [self.invalid_ror_1, self.valid_ror_1, self.valid_ror_2] 

227 storage_manager = RedisStorageManager(testing=True) 

228 redis_file = RORManager(storage_manager=storage_manager, use_api_service=True) 

229 for id in to_insert: 

230 norm_id = redis_file.normalise(id, include_prefix=True) 

231 is_valid = redis_file.is_valid(norm_id) 

232 # insert_tup = (norm_id, is_valid) 

233 redis_file.storage_manager.set_value(norm_id, is_valid) 

234 

235 # does not use API, retrieve values from DB 

236 redis_no_api = RORManager(storage_manager=storage_manager, use_api_service=False) 

237 all_db_keys = redis_no_api.storage_manager.get_all_keys() 

238 # check that all the normalised ids in the list were correctly inserted in the db 

239 self.assertTrue(all(redis_no_api.normalise(x, include_prefix=True) in all_db_keys for x in to_insert)) 

240 self.assertTrue(redis_no_api.is_valid(self.valid_ror_1)) # is stored in support file as valid 

241 self.assertTrue(redis_no_api.is_valid(self.valid_ror_2)) # is stored in support file as valid 

242 self.assertFalse(redis_no_api.is_valid(self.invalid_ror_1)) # is stored in support file as invalid 

243 self.assertTrue(redis_no_api.is_valid("ror:03ztgj036")) # is not stored in support file as invalid, does not exist but has correct syntax 

244 redis_no_api.storage_manager.delete_storage() 

245 

246 def test_ror_redis_nofile_noapi(self): 

247 # No data in redis db 

248 # Uses RedisStorageManager 

249 # Does not use API (so a syntactically correct id which is not valid is considered to be valid) 

250 rm_nofile_noapi = RORManager(storage_manager=RedisStorageManager(testing=True), use_api_service=False) 

251 self.assertTrue(rm_nofile_noapi.is_valid(self.valid_ror_1)) 

252 self.assertTrue(rm_nofile_noapi.is_valid("ror:03ztgj036")) 

253 

254 rm_nofile_noapi.storage_manager.delete_storage()