Coverage for ivatar/test_generator_equivalence.py: 100%

1""" 

2Byte-identity and cache-behavior regression tests for the accelerated 

3pagan / robohash generator implementations. 

4 

5These tests compare the public ivatar APIs against the *pristine* upstream 

6implementations (pagan.generator.generate + Robohash.assemble with the 

7original functions restored), so a future change that alters rendered 

8pixels fails here instead of silently shipping visual drift. 

9""" 

10 

11import hashlib 

12import io 

13import threading 

14from unittest.mock import patch 

15 

16from django.core.cache import cache 

17from django.test import TestCase 

18 

19import pagan.generator as pristine_pagan_gen 

20 

21from ivatar.pagan_optimized import create_optimized_pagan 

22import ivatar.pagan_optimized as pagan_mod 

23import ivatar.robohash as robo_mod 

24from ivatar.robohash import create_robohash, clear_robohash_cache 

25 

26# Digests chosen to exercise different robohash sets (set1 300px parts, 

27# set3/4/5 1015/1024px parts) and both pagan aspect branches. 

28DIGESTS = [hashlib.md5(f"equiv-{i}".encode()).hexdigest() for i in range(12)] 

29 

30 

31def _clear_local_caches(): 

32 clear_robohash_cache() 

33 robo_mod._robohash_png_cache.clear() 

34 robo_mod.OptimizedRobohash._assembly_cache.clear() 

35 pagan_mod.clear_pagan_cache() 

36 pagan_mod._pagan_png_cache.clear() 

37 

38 

39def _pristine_robohash(digest: str, size: int = 80) -> bytes: 

40 """Render with upstream Robohash.assemble + _get_list_of_files restored.""" 

41 from robohash.robohash import Robohash 

42 

43 patched_assemble = Robohash.assemble 

44 patched_listfiles = Robohash._get_list_of_files 

45 Robohash.assemble = robo_mod._orig_assemble_ref 

46 Robohash._get_list_of_files = robo_mod._orig_listfiles_ref 

47 try: 

48 robo = Robohash(digest) 

49 robo.assemble(roboset="any", sizex=size, sizey=size) 

50 buf = io.BytesIO() 

51 robo.img.save(buf, "png") 

52 return buf.getvalue() 

53 finally: 

54 Robohash.assemble = patched_assemble 

55 Robohash._get_list_of_files = patched_listfiles 

56 

57 

58def _pristine_pagan(digest: str, size: int = 80) -> bytes: 

59 from PIL import Image 

60 

61 img128 = pristine_pagan_gen.generate(digest, 0) 

62 img = img128.resize((size, size), Image.LANCZOS) 

63 buf = io.BytesIO() 

64 img.save(buf, "PNG") 

65 return buf.getvalue() 

66 

67 

68class ByteIdentityEquivalence(TestCase): 

69 """Patched pipelines must render byte-identical to pristine upstream.""" 

70 

71 def test_robohash_matches_pristine_upstream(self): 

72 for digest in DIGESTS: 

73 _clear_local_caches() 

74 patched = create_robohash(digest, 80, "any").getvalue() 

75 _clear_local_caches() 

76 self.assertEqual(patched, _pristine_robohash(digest), digest) 

77 

78 def test_pagan_matches_pristine_generator(self): 

79 for digest in DIGESTS: 

80 _clear_local_caches() 

81 patched = create_optimized_pagan(digest, 80).getvalue() 

82 _clear_local_caches() 

83 self.assertEqual(patched, _pristine_pagan(digest), digest) 

84 

85 def test_robohash_matches_pristine_other_sizes(self): 

86 for digest in DIGESTS[:5]: 

87 for size in (28, 128, 256): 

88 _clear_local_caches() 

89 patched = create_robohash(digest, size, "any").getvalue() 

90 _clear_local_caches() 

91 self.assertEqual( 

92 patched, _pristine_robohash(digest, size), (digest, size) 

93 ) 

94 

95 

96class StructureCacheSelection(TestCase): 

97 """Structure cache must not freeze per-digest file selection.""" 

98 

99 def test_selection_varies_with_digest(self): 

100 # chosen parts come from digest bits: two digests must not always 

101 # produce the same robot image 

102 imgs = set() 

103 for digest in DIGESTS[:8]: 

104 _clear_local_caches() 

105 imgs.add(create_robohash(digest, 80, "any").getvalue()) 

106 self.assertGreater(len(imgs), 1, "all digests rendered the same robot") 

107 

108 

109class SharedRedisCacheBehavior(TestCase): 

110 """pagan/robohash payloads get a cross-worker (default alias) cache.""" 

111 

112 def setUp(self): 

113 cache.clear() 

114 _clear_local_caches() 

115 

116 def test_robohash_populates_shared_cache(self): 

117 digest = DIGESTS[0] 

118 create_robohash(digest, 80, "any") 

119 self.assertIsNotNone( 

120 cache.get(robo_mod._redis_default_avatar_key(digest, 80, "any")) 

121 ) 

122 

123 def test_pagan_populates_shared_cache(self): 

124 digest = DIGESTS[0] 

125 create_optimized_pagan(digest, 80) 

126 self.assertIsNotNone(cache.get(pagan_mod._redis_pagan_key(digest, 80))) 

127 

128 def test_shared_hit_served_without_generation(self): 

129 digest = DIGESTS[0] 

130 payload = create_robohash(digest, 80, "any").getvalue() 

131 _clear_local_caches() 

132 with patch.object( 

133 robo_mod.OptimizedRobohash, 

134 "assemble_optimized", 

135 side_effect=AssertionError("must not regenerate on shared hit"), 

136 ): 

137 served = create_robohash(digest, 80, "any").getvalue() 

138 self.assertEqual(served, payload) 

139 

140 def test_shared_hit_pagan_without_generation(self): 

141 digest = DIGESTS[1] 

142 create_optimized_pagan(digest, 80) 

143 _clear_local_caches() 

144 with patch.object( 

145 pagan_mod.OptimizedPagan, 

146 "generate_optimized", 

147 side_effect=AssertionError("must not regenerate"), 

148 ) as mock: 

149 create_optimized_pagan(digest, 80) 

150 mock.assert_not_called() 

151 

152 def test_cache_failure_falls_back_to_generation(self): 

153 digest = DIGESTS[2] 

154 with patch.object( 

155 cache, 

156 "get", 

157 side_effect=ConnectionError("Redis down"), 

158 ): 

159 # get fails -> generate normally 

160 out = create_robohash(digest, 80, "any").getvalue() 

161 self.assertTrue(out) 

162 

163 def test_threadsafe_concurrent_generation(self): 

164 """Concurrent first-time requests for the same digest all succeed.""" 

165 digest = DIGESTS[3] 

166 results: list = [] 

167 errors: list = [] 

168 

169 def worker(): 

170 try: 

171 results.append(create_robohash(digest, 80, "any").getvalue()) 

172 except Exception as exc: # pragma: no cover 

173 errors.append(exc) 

174 

175 threads = [threading.Thread(target=worker) for _ in range(8)] 

176 for t in threads: 

177 t.start() 

178 for t in threads: 

179 t.join() 

180 self.assertFalse(errors) 

181 self.assertEqual(len(set(results)), 1)