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/tests/test_inventory_flush.py

@@ -0,0 +1,196 @@
+"""
+Integration test for SqliteInventory.flush()
+
+Runs with both the producer (SqliteInventory) and the consumer (sqlThread)
+active, to verify that inventory items with various payload types survive
+the queue round-trip and land correctly in the database.
+
+Reproduces the bug where BLOB fields (payload, tag) passed as memoryview
+or bytearray cause: "Error binding parameter 3 - probably unsupported type."
+"""
+
+import os
+import sqlite3
+import struct
+import tempfile
+import threading
+import time
+import unittest
+
+from .common import skip_python3
+
+skip_python3()
+
+os.environ['BITMESSAGE_HOME'] = tempfile.mkdtemp()
+
+from pybitmessage.helper_sql import (  # noqa:E402
+    sqlQuery, sql_ready, sqlStoredProcedure)
+from pybitmessage.class_sqlThread import sqlThread  # noqa:E402
+from pybitmessage.storage.sqlite import SqliteInventory  # noqa:E402
+
+
+class TestInventoryFlush(unittest.TestCase):
+    """
+    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):
+        sql_lookup = sqlThread()
+        sql_lookup.daemon = True
+        sql_lookup.start()
+        sql_ready.wait()
+        cls.inventory = SqliteInventory()
+
+    @classmethod
+    def tearDownClass(cls):
+        sqlStoredProcedure('exit')
+        for thread in threading.enumerate():
+            if thread.name == "SQL":
+                thread.join(timeout=10)
+
+    # -- helpers ----------------------------------------------------------
+
+    @staticmethod
+    def _make_hash(seed):
+        """Return a 32-byte hash derived from *seed*."""
+        return (b'\x00' * 31 + bytes([seed & 0xFF]))[-32:]
+
+    def _insert_and_flush(self, obj_hash, obj_type, stream,
+                          payload, expires, tag):
+        """
+        Put one item into the in-memory inventory, call flush(),
+        and return the row that the sqlThread actually wrote.
+        """
+        self.inventory[obj_hash] = (obj_type, stream, payload, expires, tag)
+        self.inventory.flush()
+
+        rows = sqlQuery(
+            'SELECT objecttype, streamnumber, payload, expirestime, tag'
+            ' FROM inventory WHERE hash=?',
+            sqlite3.Binary(obj_hash))
+        return rows
+
+    def _cleanup_hash(self, obj_hash):
+        """Remove a test row so tests stay independent."""
+        from pybitmessage.helper_sql import sqlExecute
+        sqlExecute('DELETE FROM inventory WHERE hash=?',
+                   sqlite3.Binary(obj_hash))
+
+    # -- test cases -------------------------------------------------------
+
+    def test_flush_with_bytes_payload(self):
+        """Baseline: payload and tag are plain bytes — must always work."""
+        h = self._make_hash(1)
+        payload = b'\x80\x01' + os.urandom(64)
+        tag = b'\xff' * 32
+        expires = int(time.time()) + 3600
+
+        rows = self._insert_and_flush(h, 2, 1, payload, expires, tag)
+        try:
+            self.assertEqual(len(rows), 1, "Row not found after flush")
+            self.assertEqual(rows[0][0], 2)
+            self.assertEqual(rows[0][1], 1)
+            self.assertEqual(bytes(rows[0][2]), payload)
+            self.assertEqual(rows[0][3], expires)
+            self.assertEqual(bytes(rows[0][4]), tag)
+        finally:
+            self._cleanup_hash(h)
+
+    def test_flush_with_memoryview_payload(self):
+        """
+        Reproduce the production crash:
+        payload and tag as memoryview objects cause
+        'Error binding parameter 3 - probably unsupported type.'
+        """
+        h = self._make_hash(2)
+        raw_payload = b'\x80\x02' + os.urandom(64)
+        raw_tag = b'\xee' * 32
+        expires = int(time.time()) + 3600
+
+        rows = self._insert_and_flush(
+            h, 2, 1, memoryview(raw_payload), expires, memoryview(raw_tag))
+        try:
+            self.assertEqual(len(rows), 1,
+                             "Row not found — flush likely crashed on "
+                             "memoryview parameters")
+            self.assertEqual(bytes(rows[0][2]), raw_payload)
+            self.assertEqual(bytes(rows[0][4]), raw_tag)
+        finally:
+            self._cleanup_hash(h)
+
+    def test_flush_with_bytearray_payload(self):
+        """bytearray is another bytes-like type that could trip sqlite3."""
+        h = self._make_hash(3)
+        raw_payload = bytearray(b'\x80\x03' + os.urandom(64))
+        raw_tag = bytearray(b'\xdd' * 32)
+        expires = int(time.time()) + 3600
+
+        rows = self._insert_and_flush(
+            h, 2, 1, raw_payload, expires, raw_tag)
+        try:
+            self.assertEqual(len(rows), 1,
+                             "Row not found — flush likely crashed on "
+                             "bytearray parameters")
+            self.assertEqual(bytes(rows[0][2]), bytes(raw_payload))
+            self.assertEqual(bytes(rows[0][4]), bytes(raw_tag))
+        finally:
+            self._cleanup_hash(h)
+
+    def test_flush_with_empty_tag(self):
+        """Empty tag (b'') must not break the INSERT."""
+        h = self._make_hash(4)
+        payload = b'\x80\x04' + os.urandom(64)
+        expires = int(time.time()) + 3600
+
+        rows = self._insert_and_flush(h, 2, 1, payload, expires, b'')
+        try:
+            self.assertEqual(len(rows), 1)
+            self.assertEqual(bytes(rows[0][4]), b'')
+        finally:
+            self._cleanup_hash(h)
+
+    def test_flush_multiple_items(self):
+        """Flush a batch and verify every row arrives."""
+        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
+            # mix types on purpose
+            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()
+
+        try:
+            for i, h in enumerate(hashes):
+                rows = sqlQuery(
+                    'SELECT objecttype FROM inventory WHERE hash=?',
+                    sqlite3.Binary(h))
+                self.assertEqual(
+                    len(rows), 1,
+                    "Item {} missing after batch flush".format(i))
+        finally:
+            for h in hashes:
+                self._cleanup_hash(h)
+
+    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()
+        try:
+            self.assertEqual(len(self.inventory._inventory), 0)
+        finally:
+            self._cleanup_hash(h)