Coverage for test / orcid_index_test.py: 99%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import os 

6import shutil 

7import tempfile 

8import unittest 

9 

10from oc_ds_converter.datasource.orcid_index import ( 

11 OrcidIndexRedis, 

12 PublishersRedis, 

13 load_orcid_index_to_redis, 

14 load_publishers_to_redis, 

15) 

16 

17 

18class TestOrcidIndexRedis(unittest.TestCase): 

19 def setUp(self) -> None: 

20 self.orcid_index = OrcidIndexRedis(testing=True) 

21 

22 def test_get_value_empty(self) -> None: 

23 result = self.orcid_index.get_value("doi:10.1234/nonexistent") 

24 self.assertIsNone(result) 

25 

26 def test_has_data_empty(self) -> None: 

27 self.assertFalse(self.orcid_index.has_data()) 

28 

29 def test_add_values_batch_and_get_value(self) -> None: 

30 data = { 

31 "doi:10.1234/test1": {"Author One orcid:0000-0001-2345-6789", "Author Two orcid:0000-0002-3456-7890"}, 

32 "doi:10.1234/test2": {"Author Three orcid:0000-0003-4567-8901"}, 

33 } 

34 self.orcid_index.add_values_batch(data) 

35 

36 result1 = self.orcid_index.get_value("doi:10.1234/test1") 

37 self.assertEqual(result1, {"Author One orcid:0000-0001-2345-6789", "Author Two orcid:0000-0002-3456-7890"}) 

38 

39 result2 = self.orcid_index.get_value("doi:10.1234/test2") 

40 self.assertEqual(result2, {"Author Three orcid:0000-0003-4567-8901"}) 

41 

42 def test_has_data_after_insert(self) -> None: 

43 self.orcid_index.add_values_batch({"doi:10.1234/test": {"value"}}) 

44 self.assertTrue(self.orcid_index.has_data()) 

45 

46 def test_clear(self) -> None: 

47 self.orcid_index.add_values_batch({"doi:10.1234/test": {"value"}}) 

48 self.assertTrue(self.orcid_index.has_data()) 

49 self.orcid_index.clear() 

50 self.assertFalse(self.orcid_index.has_data()) 

51 

52 

53class TestLoadOrcidIndexToRedis(unittest.TestCase): 

54 def setUp(self) -> None: 

55 self.orcid_index = OrcidIndexRedis(testing=True) 

56 self.temp_dir = tempfile.mkdtemp() 

57 

58 def tearDown(self) -> None: 

59 shutil.rmtree(self.temp_dir) 

60 

61 def test_load_from_directory(self) -> None: 

62 csv_content = '''"id","value" 

63"10.1234/article1","Smith, John orcid:0000-0001-2345-6789" 

64"10.1234/article1","Doe, Jane orcid:0000-0002-3456-7890" 

65"10.1234/article2","Brown, Alice orcid:0000-0003-4567-8901" 

66''' 

67 csv_path = os.path.join(self.temp_dir, "orcid_index.csv") 

68 with open(csv_path, "w", encoding="utf-8") as f: 

69 f.write(csv_content) 

70 

71 load_orcid_index_to_redis(self.temp_dir, self.orcid_index) 

72 

73 result1 = self.orcid_index.get_value("doi:10.1234/article1") 

74 self.assertEqual(result1, {"Smith, John orcid:0000-0001-2345-6789", "Doe, Jane orcid:0000-0002-3456-7890"}) 

75 

76 result2 = self.orcid_index.get_value("doi:10.1234/article2") 

77 self.assertEqual(result2, {"Brown, Alice orcid:0000-0003-4567-8901"}) 

78 

79 def test_load_nonexistent_directory(self) -> None: 

80 load_orcid_index_to_redis("/nonexistent/path", self.orcid_index) 

81 self.assertFalse(self.orcid_index.has_data()) 

82 

83 

84class TestPublishersRedis(unittest.TestCase): 

85 def setUp(self) -> None: 

86 self.publishers_redis = PublishersRedis(testing=True) 

87 

88 def test_get_by_member_empty(self) -> None: 

89 result = self.publishers_redis.get_by_member("999") 

90 self.assertIsNone(result) 

91 

92 def test_set_and_get_by_member(self) -> None: 

93 self.publishers_redis.set_publisher("123", "Test Publisher", {"10.1234", "10.5678"}) 

94 result = self.publishers_redis.get_by_member("123") 

95 assert result is not None 

96 self.assertEqual(result["name"], "Test Publisher") 

97 self.assertEqual(result["prefixes"], {"10.1234", "10.5678"}) 

98 

99 def test_get_by_prefix(self) -> None: 

100 self.publishers_redis.set_publisher("123", "Test Publisher", {"10.1234"}) 

101 result = self.publishers_redis.get_by_prefix("10.1234") 

102 assert result is not None 

103 self.assertEqual(result["name"], "Test Publisher") 

104 

105 def test_get_by_prefix_not_found(self) -> None: 

106 result = self.publishers_redis.get_by_prefix("10.9999") 

107 self.assertIsNone(result) 

108 

109 def test_has_data_empty(self) -> None: 

110 self.assertFalse(self.publishers_redis.has_data()) 

111 

112 def test_has_data_after_insert(self) -> None: 

113 self.publishers_redis.set_publisher("1", "Publisher", {"10.1000"}) 

114 self.assertTrue(self.publishers_redis.has_data()) 

115 

116 def test_clear(self) -> None: 

117 self.publishers_redis.set_publisher("1", "Publisher", {"10.1000"}) 

118 self.assertTrue(self.publishers_redis.has_data()) 

119 self.publishers_redis.clear() 

120 self.assertFalse(self.publishers_redis.has_data()) 

121 

122 def test_set_publishers_batch(self) -> None: 

123 publishers = { 

124 "1": {"name": "Publisher A", "prefixes": {"10.1111"}}, 

125 "2": {"name": "Publisher B", "prefixes": {"10.2222", "10.3333"}}, 

126 } 

127 self.publishers_redis.set_publishers_batch(publishers) 

128 

129 result_a = self.publishers_redis.get_by_member("1") 

130 assert result_a is not None 

131 self.assertEqual(result_a["name"], "Publisher A") 

132 self.assertEqual(result_a["prefixes"], {"10.1111"}) 

133 

134 result_b = self.publishers_redis.get_by_prefix("10.2222") 

135 assert result_b is not None 

136 self.assertEqual(result_b["name"], "Publisher B") 

137 

138 result_c = self.publishers_redis.get_by_prefix("10.3333") 

139 assert result_c is not None 

140 self.assertEqual(result_c["name"], "Publisher B") 

141 

142 

143class TestLoadPublishersToRedis(unittest.TestCase): 

144 def setUp(self) -> None: 

145 self.publishers_redis = PublishersRedis(testing=True) 

146 self.temp_dir = tempfile.mkdtemp() 

147 

148 def tearDown(self) -> None: 

149 shutil.rmtree(self.temp_dir) 

150 

151 def test_load_from_csv(self) -> None: 

152 csv_content = '''"id","name","prefix" 

153"100","Test Publisher 1","10.100" 

154"100","Test Publisher 1","10.101" 

155"200","Test Publisher 2","10.200" 

156''' 

157 csv_path = os.path.join(self.temp_dir, "publishers.csv") 

158 with open(csv_path, "w", encoding="utf-8") as f: 

159 f.write(csv_content) 

160 

161 load_publishers_to_redis(csv_path, self.publishers_redis) 

162 

163 result1 = self.publishers_redis.get_by_member("100") 

164 assert result1 is not None 

165 self.assertEqual(result1["name"], "Test Publisher 1") 

166 self.assertEqual(result1["prefixes"], {"10.100", "10.101"}) 

167 

168 result2 = self.publishers_redis.get_by_prefix("10.200") 

169 assert result2 is not None 

170 self.assertEqual(result2["name"], "Test Publisher 2") 

171 

172 def test_load_nonexistent_file(self) -> None: 

173 load_publishers_to_redis("/nonexistent/file.csv", self.publishers_redis) 

174 self.assertFalse(self.publishers_redis.has_data()) 

175 

176 

177if __name__ == "__main__": 

178 unittest.main()