Add tests for handling of icon metadata

kittens/transfer/algorithm.c

@@ -933,8 +933,30 @@ parse_ftc(PyObject *self UNUSED, PyObject *args) {
     Py_RETURN_NONE;
 }
 
+static PyObject*
+pyxxh128_hash(PyObject *self UNUSED, PyObject *b) {
+    RAII_PY_BUFFER(data);
+    if (PyObject_GetBuffer(b, &data, PyBUF_SIMPLE) == -1) return NULL;
+    XXH128_canonical_t c;
+    XXH128_canonicalFromHash(&c, XXH3_128bits(data.buf, data.len));
+    return PyBytes_FromStringAndSize((char*)c.digest, sizeof(c.digest));
+}
+
+static PyObject*
+pyxxh128_hash_with_seed(PyObject *self UNUSED, PyObject *args) {
+    RAII_PY_BUFFER(data);
+    unsigned long long seed;
+    if (!PyArg_ParseTuple(args, "y*K", &data, &seed)) return NULL;
+    XXH128_canonical_t c;
+    XXH128_canonicalFromHash(&c, XXH3_128bits_withSeed(data.buf, data.len, seed));
+    return PyBytes_FromStringAndSize((char*)c.digest, sizeof(c.digest));
+}
+
+
 static PyMethodDef module_methods[] = {
     {"parse_ftc", parse_ftc, METH_VARARGS, ""},
+    {"xxh128_hash", pyxxh128_hash, METH_O, ""},
+    {"xxh128_hash_with_seed", pyxxh128_hash_with_seed, METH_VARARGS, ""},
     {NULL, NULL, 0, NULL}        /* Sentinel */
 };
 

kittens/transfer/rsync.pyi

@@ -20,6 +20,9 @@ class Hasher:
     @property
     def name(self) -> str: ...
 
+def xxh128_hash(data: ReadOnlyBuffer) -> bytes: ...
+def xxh128_hash_with_seed(data: ReadOnlyBuffer, seed: int) -> bytes: ...
+
 
 class Patcher:
 

kitty/notifications.py

@@ -28,6 +28,8 @@ class IconDataCache:
         self.base_cache_dir = base_cache_dir
         self.cache_dir = ''
         self.total_size = 0
+        import struct
+        self.seed: int = struct.unpack("!Q", os.urandom(8))[0]
 
     def _ensure_state(self) -> str:
         if not self.cache_dir:
@@ -45,10 +47,14 @@ class IconDataCache:
     def keys(self) -> Iterator[str]:
         yield from self.key_map.keys()
 
+    def hash(self, data: bytes) -> str:
+        from kittens.transfer.rsync import xxh128_hash_with_seed
+        d = xxh128_hash_with_seed(data, self.seed)
+        return d.hex()
+
     def add_icon(self, key: str, data: bytes) -> str:
-        from kittens.transfer.rsync import Hasher
         self._ensure_state()
-        data_hash = Hasher(which='xxh3-128', data=data).hexdigest()
+        data_hash = self.hash(data)
         path = os.path.join(self.cache_dir, data_hash)
         if not os.path.exists(path):
             with open(path, 'wb') as f:
@@ -231,6 +237,8 @@ class NotificationCommand:
         payload_is_encoded = False
         if metadata:
             for part in metadata.split(':'):
+                if not part:
+                    continue
                 k, v = part.split('=', 1)
                 if k == 'p':
                     try:

kitty_tests/notifications.py

@@ -47,7 +47,9 @@ class DesktopIntegration(DesktopIntegration):
         urgency: Urgency = Urgency.Normal,
     ) -> int:
         self.counter += 1
-        self.notifications.append(n(title, body, urgency, self.counter, icon_name, icon_path))
+        ans = n(title, body, urgency, self.counter, icon_name)
+        ans['icon_path'] = os.path.basename(icon_path)
+        self.notifications.append(ans)
         return self.counter
 
 
@@ -76,7 +78,7 @@ class Channel(Channel):
 def do_test(self: 'TestNotifications', tdir: str) -> None:
     di = DesktopIntegration(None)
     ch = Channel()
-    nm = NotificationManager(di, ch, lambda *a, **kw: None)
+    nm = NotificationManager(di, ch, lambda *a, **kw: None, base_cache_dir=tdir)
     di.notification_manager = nm
 
     def reset():
@@ -237,6 +239,31 @@ def do_test(self: 'TestNotifications', tdir: str) -> None:
     del dc
     self.assertFalse(os.path.exists(cache_dir))
 
+    # Test icons
+    def send_with_icon(data='', n='', g=''):
+        m = ''
+        if n:
+            m += f'n={n}:'
+        if g:
+            m += f'g={g}:'
+        h(f'i=9:d=0:{m};title')
+        h(f'i=9:p=icon;{data}')
+
+    dc = nm.icon_data_cache
+    send_with_icon(n='mycon')
+    self.ae(di.notifications, [n(icon_name='mycon')])
+    reset()
+    send_with_icon(g='gid')
+    self.ae(di.notifications, [n()])
+    reset()
+    send_with_icon(g='gid', data='1')
+    self.ae(di.notifications, [n(icon_path=dc.hash(b'1'))])
+    send_with_icon(g='gid', n='moose')
+    self.ae(di.notifications[-1], n(icon_name='moose', icon_path=dc.hash(b'1'), desktop_notification_id=len(di.notifications)))
+    send_with_icon(g='gid2', data='2')
+    self.ae(di.notifications[-1], n(icon_path=dc.hash(b'2'), desktop_notification_id=len(di.notifications)))
+    reset()
+
 
 class TestNotifications(BaseTest):