Fix inventory flushing

- it's a combo of handling a tuple and pickling of raw data, and is
  triggered under rare occasions

Author: PeterSurda

src/storage/sqlite.py

@@ -109,7 +109,12 @@ def flush(self):
                 for objectHash, value in self._inventory.items():
                     sql.execute(
                         'INSERT INTO inventory VALUES (?, ?, ?, ?, ?, ?)',
-                        sqlite3.Binary(objectHash), *value)
+                        sqlite3.Binary(objectHash),
+                        value.type,
+                        value.stream,
+                        sqlite3.Binary(value.payload),
+                        value.expires,
+                        sqlite3.Binary(value.tag))
                 self._inventory.clear()
 
     def clean(self):

src/tests/test_inventory_flush.py

@@ -0,0 +1,135 @@
+"""Tests for SqliteInventory.flush()"""
+# pylint: disable=protected-access,wrong-import-order,wrong-import-position
+# pylint: disable=import-outside-toplevel
+
+import os
+import struct
+import tempfile
+import threading
+import time
+
+from .common import skip_python3
+from .partial import TestPartialRun
+
+skip_python3()
+
+
+class TestInventoryFlush(TestPartialRun):
+    """
+    Integration test: exercises flush() end-to-end with the real sqlThread
+    consumer running, so that type errors in parameter binding surface here
+    rather than silently killing a production thread.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        os.environ['BITMESSAGE_HOME'] = tempfile.gettempdir()
+        super(TestInventoryFlush, cls).setUpClass()
+
+        from class_sqlThread import sqlThread
+        from helper_sql import sql_ready, sqlStoredProcedure
+        from storage.sqlite import SqliteInventory
+
+        cls._sqlStoredProcedure = staticmethod(sqlStoredProcedure)
+
+        sql_lookup = sqlThread()
+        sql_lookup.daemon = True
+        sql_lookup.start()
+        sql_ready.wait()
+        cls.inventory = SqliteInventory()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._sqlStoredProcedure('exit')
+        for thread in threading.enumerate():
+            if thread.name == "SQL":
+                thread.join(timeout=10)
+        super(TestInventoryFlush, cls).tearDownClass()
+
+    # -- helpers ----------------------------------------------------------
+
+    @staticmethod
+    def _make_hash(seed):
+        """Return a 32-byte hash derived from *seed*."""
+        return (b'\x00' * 31 + bytes([seed & 0xFF]))[-32:]
+
+    def _flush_and_check(self, obj_hash):
+        """
+        Flush the inventory to the database, clear the _objects lookup
+        cache so that __contains__ is forced to hit sqlite, then verify
+        the hash is found via the normal inventory API.
+        """
+        self.inventory.flush()
+        self.inventory._objects.clear()
+        self.assertIn(obj_hash, self.inventory)
+
+    # -- test cases -------------------------------------------------------
+
+    def test_flush_with_bytes_payload(self):
+        """Baseline: payload and tag are plain bytes."""
+        h = self._make_hash(1)
+        self.inventory[h] = (
+            2, 1, b'\x80\x01' + os.urandom(64),
+            int(time.time()) + 3600, b'\xff' * 32)
+        self._flush_and_check(h)
+
+    def test_flush_with_memoryview_payload(self):
+        """
+        Reproduce the production crash: payload and tag as memoryview
+        cause 'Error binding parameter 3 - probably unsupported type.'
+        """
+        h = self._make_hash(2)
+        self.inventory[h] = (
+            2, 1, memoryview(b'\x80\x02' + os.urandom(64)),
+            int(time.time()) + 3600, memoryview(b'\xee' * 32))
+        self._flush_and_check(h)
+
+    def test_flush_with_bytearray_payload(self):
+        """bytearray is another bytes-like type that could trip sqlite3."""
+        h = self._make_hash(3)
+        self.inventory[h] = (
+            2, 1, bytearray(b'\x80\x03' + os.urandom(64)),
+            int(time.time()) + 3600, bytearray(b'\xdd' * 32))
+        self._flush_and_check(h)
+
+    def test_flush_with_empty_tag(self):
+        """Empty tag (b'') must not break the INSERT."""
+        h = self._make_hash(4)
+        self.inventory[h] = (
+            2, 1, b'\x80\x04' + os.urandom(64),
+            int(time.time()) + 3600, b'')
+        self._flush_and_check(h)
+
+    # pylint: disable=redefined-variable-type
+    def test_flush_multiple_mixed_types(self):
+        """Flush a batch of items with mixed blob types."""
+        count = 20
+        hashes = [self._make_hash(0x10 + i) for i in range(count)]
+        expires = int(time.time()) + 3600
+
+        for i, h in enumerate(hashes):
+            payload = struct.pack('>I', i) + os.urandom(60)
+            tag = struct.pack('>I', i) + b'\x00' * 28
+            if i % 3 == 0:
+                payload = memoryview(payload)
+                tag = memoryview(tag)
+            elif i % 3 == 1:
+                payload = bytearray(payload)
+                tag = bytearray(tag)
+            self.inventory[h] = (2, 1, payload, expires, tag)
+
+        self.inventory.flush()
+        self.inventory._objects.clear()
+
+        for i, h in enumerate(hashes):
+            self.assertIn(
+                h, self.inventory,
+                "Item {} missing after batch flush".format(i))
+
+    def test_flush_clears_memory_cache(self):
+        """After flush the in-memory _inventory dict must be empty."""
+        h = self._make_hash(0xF0)
+        self.inventory[h] = (
+            2, 1, b'\x00' * 32, int(time.time()) + 3600, b'')
+        self.inventory.flush()
+        self.assertEqual(len(self.inventory._inventory), 0)