Coverage for test / idm_viaf_test.py: 99%

1# SPDX-FileCopyrightText: 2023 Arianna Moretti <arianna.moretti4@unibo.it> 

2# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from oc_ds_converter.oc_idmanager.viaf import ViafManager 

8import json 

9import sqlite3 

10import os.path 

11import unittest 

12from os import makedirs 

13from os.path import exists, join 

14 

15import xmltodict 

16from oc_ds_converter.oc_idmanager import * 

17from oc_ds_converter.oc_idmanager.base import IdentifierManager 

18from requests import ReadTimeout, get 

19from requests.exceptions import ConnectionError 

20from oc_ds_converter.oc_idmanager.oc_data_storage.storage_manager import StorageManager 

21from oc_ds_converter.oc_idmanager.oc_data_storage.in_memory_manager import InMemoryStorageManager 

22from oc_ds_converter.oc_idmanager.oc_data_storage.sqlite_manager import SqliteStorageManager 

23from oc_ds_converter.oc_idmanager.oc_data_storage.redis_manager import RedisStorageManager 

24 

25class viafIdentifierManagerTest(unittest.TestCase): 

26 """This class aim at testing identifiers manager.""" 

27 

28 def setUp(self): 

29 if not exists("tmp"): 

30 makedirs("tmp") 

31 

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

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

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

35 self.data = json.load(fp) 

36 

37 self.valid_viaf_1 = "5604148947771454950004" 

38 self.valid_viaf_2 = "234145033" 

39 self.valid_viaf_3 = "56752857" 

40 self.invalid_viaf_1 = "012517637138" 

41 self.invalid_viaf_2 = "5604148947771454953333" 

42 self.valid_viaf_3 = "56752857abc" 

43 self.invalid_viaf_4 = "5604148999953333" 

44 

45 

46 def test_exists(self): 

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

48 viaf_manager = ViafManager() 

49 output = viaf_manager.exists(self.valid_viaf_3, get_extra_info=True, allow_extra_api=None) 

50 expected_output = (True, {'id': viaf_manager.normalise(self.valid_viaf_3), 'valid': True}) 

51 self.assertEqual(output, expected_output) 

52 with self.subTest(msg="get_extra_info=False, allow_extra_api=None"): 

53 viaf_manager = ViafManager() 

54 output = viaf_manager.exists(self.valid_viaf_3, get_extra_info=False, allow_extra_api=None) 

55 expected_output = True 

56 self.assertEqual(output, expected_output) 

57 with self.subTest(msg="get_extra_info=False, allow_extra_api='None'"): 

58 viaf_manager = ViafManager() 

59 output = viaf_manager.exists(self.valid_viaf_2, get_extra_info=False, allow_extra_api=None) 

60 expected_output = True 

61 self.assertEqual(output, expected_output) 

62 with self.subTest(msg="get_extra_info=False, allow_extra_api='None'"): 

63 viaf_manager = ViafManager() 

64 output = viaf_manager.exists(self.invalid_viaf_1, get_extra_info=False, allow_extra_api=None) 

65 expected_output = False 

66 self.assertEqual(output, expected_output) 

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

68 viaf_manager = ViafManager() 

69 output = viaf_manager.exists(self.invalid_viaf_1, get_extra_info=True, allow_extra_api=None) 

70 expected_output = (False, {'id': viaf_manager.normalise(self.invalid_viaf_1), 'valid': False}) 

71 self.assertEqual(output, expected_output) 

72 

73 

74 def test_viaf_normalise(self): 

75 vm = ViafManager() 

76 self.assertEqual( 

77 vm.normalise(self.valid_viaf_1), 

78 vm.normalise(' ' + self.valid_viaf_1), 

79 ) 

80 self.assertEqual( 

81 vm.normalise(self.valid_viaf_1), 

82 vm.normalise('abc' + self.valid_viaf_1), 

83 ) 

84 self.assertEqual( 

85 vm.normalise(self.valid_viaf_1), 

86 vm.normalise('\n' + self.valid_viaf_1), 

87 ) 

88 

89 def test_viaf_is_valid(self): 

90 vm = ViafManager() 

91 self.assertTrue(vm.is_valid(self.valid_viaf_1)) 

92 self.assertTrue(vm.is_valid(self.valid_viaf_2)) 

93 self.assertFalse(vm.is_valid(self.invalid_viaf_1)) 

94 

95 

96 dm_file = ViafManager(storage_manager=InMemoryStorageManager(self.test_json_path), use_api_service=False) 

97 self.assertTrue(dm_file.normalise(self.valid_viaf_1, include_prefix=True) in self.data) 

98 self.assertTrue(dm_file.normalise(self.invalid_viaf_2, include_prefix=True) in self.data) 

99 self.assertTrue(dm_file.is_valid(self.valid_viaf_1)) 

100 self.assertFalse(dm_file.is_valid(self.invalid_viaf_2)) 

101 

102 

103 def test_viaf_default(self): 

104 am_nofile = ViafManager(storage_manager=InMemoryStorageManager("storage/id_value.json")) 

105 # No support files (it generates it) 

106 # Uses InMemoryStorageManager + generates file on method call (not automatically) 

107 # uses API 

108 self.assertTrue(am_nofile.is_valid(self.valid_viaf_1)) 

109 self.assertTrue(am_nofile.is_valid(self.valid_viaf_2)) 

110 self.assertFalse(am_nofile.is_valid(self.invalid_viaf_2)) 

111 self.assertFalse(am_nofile.is_valid(self.invalid_viaf_1)) 

112 am_nofile.storage_manager.store_file() 

113 validated_ids = [self.valid_viaf_1, self.valid_viaf_2, self.invalid_viaf_1, self.invalid_viaf_2] 

114 # check that the support file was correctly created 

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

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

117 load_dict = json.load(lj) 

118 lj.close() 

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

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

121 am_nofile.storage_manager.delete_storage() 

122 # check that the support file was correctly deleted 

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

124 

125 def test_viaf_memory_file_noapi(self): 

126 # Uses support file (without updating it) 

127 # Uses InMemoryStorageManager storage manager 

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

129 am_file = ViafManager(storage_manager=InMemoryStorageManager(self.test_json_path), use_api_service=False) 

130 self.assertTrue(am_file.normalise(self.valid_viaf_1, include_prefix=True) in self.data) 

131 self.assertTrue(am_file.normalise(self.invalid_viaf_2, include_prefix=True) in self.data) 

132 self.assertFalse(am_file.is_valid(self.invalid_viaf_2)) # is stored in support file as invalid 

133 self.assertTrue(am_file.is_valid(am_file.normalise(self.invalid_viaf_4, include_prefix=True))) # is not stored in support file as invalid, does not exist but has correct syntax 

134 

135 def test_viaf_memory_file_api(self): 

136 # Uses support file (without updating it) 

137 # Uses InMemoryStorageManager storage manager 

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

139 am_file = ViafManager(storage_manager=InMemoryStorageManager(self.test_json_path), use_api_service=True) 

140 self.assertFalse(am_file.is_valid(self.invalid_viaf_1)) 

141 

142 def test_viaf_memory_nofile_noapi(self): 

143 # Does not use support file 

144 # Uses InMemoryStorageManager storage manager 

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

146 am_nofile_noapi = ViafManager(storage_manager=InMemoryStorageManager(), use_api_service=False) 

147 self.assertTrue(am_nofile_noapi.is_valid(self.valid_viaf_1)) 

148 self.assertTrue(am_nofile_noapi.is_valid(self.invalid_viaf_2)) 

149 am_nofile_noapi.storage_manager.delete_storage() 

150 

151 def test_viaf_sqlite_nofile_api(self): 

152 # No support files (it generates it) 

153 # storage manager : SqliteStorageManager 

154 # uses API 

155 sql_am_nofile = ViafManager(storage_manager=SqliteStorageManager()) 

156 self.assertTrue(sql_am_nofile.is_valid(self.valid_viaf_1)) 

157 self.assertTrue(sql_am_nofile.is_valid(self.valid_viaf_2)) 

158 self.assertFalse(sql_am_nofile.is_valid(self.invalid_viaf_1)) 

159 self.assertFalse(sql_am_nofile.is_valid(self.invalid_viaf_2)) 

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

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

162 validated_ids = [self.valid_viaf_1, self.valid_viaf_2, self.invalid_viaf_1, self.invalid_viaf_2] 

163 all_ids_stored = sql_am_nofile.storage_manager.get_all_keys() 

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

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

166 

167 sql_am_nofile.storage_manager.delete_storage() 

168 # check that the support file was correctly deleted 

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

170 

171 def test_viaf_sqlite_file_api(self): 

172 # Uses support file 

173 # Uses SqliteStorageManager storage manager 

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

175 # db creation 

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

177 if os.path.exists(test_sqlite_db): 

178 os.remove(test_sqlite_db) 

179 #con = sqlite3.connect(test_sqlite_db) 

180 #cur = con.cursor() 

181 to_insert = [self.invalid_viaf_1, self.valid_viaf_1] 

182 sql_file = ViafManager(storage_manager=SqliteStorageManager(test_sqlite_db), use_api_service=True) 

183 for id in to_insert: 

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

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

186 insert_tup = (norm_id, is_valid) 

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

188 sql_file.storage_manager.con.commit() 

189 sql_file.storage_manager.con.close() 

190 

191 sql_no_api = ViafManager(storage_manager=SqliteStorageManager(test_sqlite_db), use_api_service=False) 

192 all_db_keys = sql_no_api.storage_manager.get_all_keys() 

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

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

195 self.assertTrue(sql_no_api.is_valid(self.valid_viaf_1)) # is stored in support file as valid 

196 self.assertFalse(sql_no_api.is_valid(self.invalid_viaf_1)) # is stored in support file as invalid 

197 self.assertTrue(sql_no_api.is_valid(sql_no_api.normalise("10.1109/5.771073FAKE_ID", include_prefix=True))) # is not stored in support file as invalid, does not exist but has correct syntax 

198 sql_no_api.storage_manager.delete_storage() 

199 

200 def test_viaf_sqlite_nofile_noapi(self): 

201 # Does not use support file 

202 # Uses SqliteStorageManager storage manager 

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

204 am_nofile_noapi = ViafManager(storage_manager=SqliteStorageManager(), use_api_service=False) 

205 self.assertTrue(am_nofile_noapi.is_valid(self.valid_viaf_1)) 

206 self.assertTrue(am_nofile_noapi.is_valid(self.invalid_viaf_2)) 

207 am_nofile_noapi.storage_manager.delete_storage() 

208 

209 #### REDIS STORAGE MANAGER 

210 def test_viaf_redis_nofile_api(self): 

211 # No available data in redis db 

212 # Storage manager : RedisStorageManager 

213 # uses API 

214 vm_nofile = ViafManager(storage_manager=RedisStorageManager(testing=True)) 

215 self.assertTrue(vm_nofile.is_valid(self.valid_viaf_1)) 

216 self.assertTrue(vm_nofile.is_valid(self.valid_viaf_2)) 

217 

218 self.assertFalse(vm_nofile.is_valid(self.invalid_viaf_1)) 

219 self.assertFalse(vm_nofile.is_valid(self.invalid_viaf_2)) 

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

221 

222 validated_ids = {self.valid_viaf_1, self.valid_viaf_2, self.invalid_viaf_1, self.invalid_viaf_2} 

223 validated_ids = {vm_nofile.normalise(x, include_prefix=True) for x in validated_ids} 

224 all_ids_stored = vm_nofile.storage_manager.get_all_keys() 

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

226 self.assertEqual(validated_ids, all_ids_stored) 

227 vm_nofile.storage_manager.delete_storage() 

228 # check that the support file was correctly deleted 

229 self.assertEqual(vm_nofile.storage_manager.get_all_keys(), set()) 

230 

231 def test_viaf_redis_file_api(self): 

232 # Uses data in redis db 

233 # Uses RedisStorageManager 

234 # fills db 

235 

236 # use API to save validity values 

237 to_insert = [self.invalid_viaf_1, self.valid_viaf_1, self.valid_viaf_3] 

238 storage_manager = RedisStorageManager(testing=True) 

239 redis_file = ViafManager(storage_manager=storage_manager, use_api_service=True) 

240 for id in to_insert: 

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

242 is_valid = redis_file.is_valid(norm_id) 

243 # insert_tup = (norm_id, is_valid) 

244 redis_file.storage_manager.set_value(norm_id, is_valid) 

245 

246 # does not use API, retrieve values from DB 

247 redis_no_api = ViafManager(storage_manager=storage_manager, use_api_service=False) 

248 all_db_keys = redis_no_api.storage_manager.get_all_keys() 

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

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

251 self.assertTrue(redis_no_api.is_valid(self.valid_viaf_1)) # is stored in support file as valid 

252 self.assertTrue(redis_no_api.is_valid(self.valid_viaf_2)) # is stored in support file as valid 

253 self.assertFalse(redis_no_api.is_valid(self.invalid_viaf_1)) # is stored in support file as invalid 

254 self.assertTrue(redis_no_api.is_valid( 

255 self.invalid_viaf_2)) # is not stored in support file as invalid, does not exist but has correct syntax 

256 redis_no_api.storage_manager.delete_storage() 

257 

258 def test_viaf_redis_nofile_noapi(self): 

259 # No data in redis db 

260 # Uses RedisStorageManager 

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

262 vm_nofile_noapi = ViafManager(storage_manager=RedisStorageManager(testing=True), use_api_service=False) 

263 self.assertTrue(vm_nofile_noapi.is_valid(self.valid_viaf_1)) 

264 self.assertTrue(vm_nofile_noapi.is_valid(self.invalid_viaf_2)) 

265 

266 vm_nofile_noapi.storage_manager.delete_storage()