Merge pull request #8881 from accumulator/qobject_test

add a simple test framework for testing QObjects and their signal/slot mechanism

.cirrus.yml

@@ -36,6 +36,8 @@ task:
   install_script:
     - apt-get update
     - apt-get -y install libsecp256k1-dev
+    # qml test reqs:
+    - apt-get -y install libgl1 libegl1 libxkbcommon0 libdbus-1-3
     - pip install -r $ELECTRUM_REQUIREMENTS_CI
   tox_script:
     - export PYTHONASYNCIODEBUG

electrum/gui/qml/qeqr.py

@@ -10,7 +10,13 @@ from PIL import ImageQt
 from PyQt6.QtCore import pyqtProperty, pyqtSignal, pyqtSlot, QObject, QRect
 from PyQt6.QtGui import QImage, QColor
 from PyQt6.QtQuick import QQuickImageProvider
-from PyQt6.QtMultimedia import QVideoSink
+try:
+    from PyQt6.QtMultimedia import QVideoSink
+except ImportError:
+    # stub QVideoSink when not found, as it's not essential on android
+    # and requires many dependencies when unit testing.
+    # Note: missing QtMultimedia will lead to errors when using QR scanner on desktop
+    from PyQt6.QtCore import QObject as QVideoSink
 
 from electrum.logging import get_logger
 from electrum.qrreader import get_qr_reader

electrum/tests/qt_util.py

@@ -0,0 +1,93 @@
+import threading
+import traceback
+import unittest
+from functools import wraps, partial
+from unittest import SkipTest
+
+from PyQt6.QtCore import QCoreApplication, QTimer, QMetaObject, Qt, pyqtSlot, QObject
+
+
+class TestQCoreApplication(QCoreApplication):
+    @pyqtSlot()
+    def doInvoke(self):
+        getattr(self._instance, self._method)()
+
+
+class QEventReceiver(QObject):
+    def __init__(self, *signals):
+        super().__init__()
+        self.received = []
+        self.signals = []
+        for signal in signals:
+            self.signals.append(signal)
+            signal.connect(partial(self.doReceive, signal))
+
+    # intentionally no pyqtSlot decorator, to catch all
+    def doReceive(self, signal, *args):
+        self.received.append((signal, args))
+
+    def receivedForSignal(self, signal):
+        return list(filter(lambda x: x[0] == signal, self.received))
+
+    def clear(self):
+        self.received.clear()
+
+
+class QETestCase(unittest.IsolatedAsyncioTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.app = None
+        self._e = None
+        self._event = threading.Event()
+
+        def start_qt_task():
+            self.app = TestQCoreApplication([])
+
+            self.timer = QTimer(self.app)
+            self.timer.setSingleShot(False)
+            self.timer.setInterval(500)  # msec
+            self.timer.timeout.connect(lambda: None)  # periodically enter python scope
+
+            self.app.exec()
+            self.app = None
+
+        self.qt_thread = threading.Thread(target=start_qt_task)
+        self.qt_thread.start()
+
+    def tearDown(self):
+        self.app.exit()
+        if self.qt_thread.is_alive():
+            self.qt_thread.join()
+
+
+def qt_test(func):
+    @wraps(func)
+    def decorator(self, *args):
+        if threading.current_thread().name == 'MainThread':
+            self.app._instance = self
+            self.app._method = func.__name__
+            QMetaObject.invokeMethod(self.app, 'doInvoke', Qt.ConnectionType.QueuedConnection)
+            self._event.wait(15)
+            if self._e:
+                print("".join(traceback.format_exception(self._e)))
+                # deallocate stored exception from qt thread otherwise we SEGV garbage collector
+                # instead, re-create using the exception message, special casing AssertionError and SkipTest
+                e = None
+                if isinstance(self._e, AssertionError):
+                    e = AssertionError(str(self._e))
+                elif isinstance(self._e, SkipTest):
+                    e = SkipTest(str(self._e))
+                else:
+                    e = Exception(str(self._e))
+                self._e = None
+                raise e
+            return
+        try:
+            func(self, *args)
+        except Exception as e:
+            self._e = e
+        finally:
+            self._event.set()
+            self._event.clear()
+    return decorator

electrum/tests/test_qml_types.py

@@ -0,0 +1,140 @@
+import shutil
+import tempfile
+
+from electrum import SimpleConfig
+from electrum.gui.qml.qetypes import QEAmount
+from electrum.invoices import Invoice, LN_EXPIRY_NEVER
+from electrum.tests.qt_util import QETestCase, QEventReceiver, qt_test
+from electrum.transaction import PartialTxOutput
+
+
+class WalletMock:
+    def __init__(self, electrum_path):
+        self.config = SimpleConfig({
+            'electrum_path': electrum_path,
+            'decimal_point': 5
+        })
+        self.contacts = None
+
+
+class TestTypes(QETestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.electrum_path = tempfile.mkdtemp()
+        self.wallet = WalletMock(self.electrum_path)
+
+    def tearDown(self):
+        super().tearDown()
+        shutil.rmtree(self.electrum_path)
+
+    @qt_test
+    def test_qeamount(self):
+        a = QEAmount()
+        self.assertTrue(a.isEmpty)
+        a_er = QEventReceiver(a.valueChanged)
+        a.satsInt = 1
+        self.assertTrue(bool(a_er.received))
+        self.assertFalse(a.isEmpty)
+        self.assertEqual('1', a.satsStr)
+
+        a_er.clear()
+        a.clear()
+        self.assertTrue(a.isEmpty)
+        self.assertTrue(bool(a_er.received))
+        self.assertEqual('0', a.satsStr)
+
+        a.clear()
+        a_er.clear()
+        a.isMax = True
+        self.assertTrue(a.isMax)
+        self.assertFalse(a.isEmpty)
+        self.assertTrue(bool(a_er.received))
+        self.assertEqual('0', a.satsStr)
+
+        a.clear()
+        a_er.clear()
+        a.msatsInt = 1
+        self.assertTrue(bool(a_er.received))
+        self.assertFalse(a.isEmpty)
+        self.assertEqual('1', a.msatsStr)
+
+    @qt_test
+    def test_qeamount_copy(self):
+        a = QEAmount()
+        b = QEAmount()
+        b.satsInt = 1
+        c = QEAmount()
+        c.msatsInt = 1
+        d = QEAmount()
+        d.isMax = True
+
+        t = QEAmount()
+        t_er = QEventReceiver(t.valueChanged)
+
+        t.copyFrom(a)
+        self.assertTrue(t.isEmpty)
+        self.assertEqual(0, len(t_er.received))
+
+        t.clear()
+        t_er.clear()
+        t.copyFrom(b)
+        self.assertFalse(t.isEmpty)
+        self.assertEqual(t.satsInt, 1)
+        self.assertEqual(1, len(t_er.received))
+
+        t.clear()
+        t_er.clear()
+        t.copyFrom(c)
+        self.assertFalse(t.isEmpty)
+        self.assertEqual(t.msatsInt, 1)
+        self.assertEqual(1, len(t_er.received))
+
+        t.clear()
+        t_er.clear()
+        t.copyFrom(d)
+        self.assertFalse(t.isEmpty)
+        self.assertTrue(t.isMax)
+        self.assertEqual(1, len(t_er.received))
+
+    @qt_test
+    def test_qeamount_frominvoice(self):
+        amount_sat = 10_000
+        outputs = [PartialTxOutput.from_address_and_value('bc1qj3zx2zc4rpv3npzmznxhdxzn0wm7pzqp8p2293', amount_sat)]
+        invoice = Invoice(
+            amount_msat=amount_sat * 1000,
+            message="mymsg",
+            time=1692716965,
+            exp=LN_EXPIRY_NEVER,
+            outputs=outputs,
+            bip70=None,
+            height=0,
+            lightning_invoice=None,
+        )
+        a = QEAmount(from_invoice=invoice)
+        self.assertEqual(10_000, a.satsInt)
+        self.assertEqual(10_000_000, a.msatsInt)
+        self.assertFalse(a.isMax)
+
+        outputs = [PartialTxOutput.from_address_and_value('bc1qj3zx2zc4rpv3npzmznxhdxzn0wm7pzqp8p2293', '!')]
+        invoice = Invoice(
+            amount_msat='!',
+            message="mymsg",
+            time=1692716965,
+            exp=LN_EXPIRY_NEVER,
+            outputs=outputs,
+            bip70=None,
+            height=0,
+            lightning_invoice=None,
+        )
+        a = QEAmount(from_invoice=invoice)
+        self.assertTrue(a.isMax)
+        self.assertEqual(0, a.satsInt)
+        self.assertEqual(0, a.msatsInt)
+
+        bolt11 = 'lnbc20m1pvjluezpp5qqqsyqcyq5rqwzqfqqqsyqcyq5rqwzqfqqqsyqcyq5rqwzqfqypqhp58yjmdan79s6qqdhdzgynm4zwqd5d7xmw5fk98klysy043l2ahrqsfpp3qjmp7lwpagxun9pygexvgpjdc4jdj85fr9yq20q82gphp2nflc7jtzrcazrra7wwgzxqc8u7754cdlpfrmccae92qgzqvzq2ps8pqqqqqqpqqqqq9qqqvpeuqafqxu92d8lr6fvg0r5gv0heeeqgcrqlnm6jhphu9y00rrhy4grqszsvpcgpy9qqqqqqgqqqqq7qqzqj9n4evl6mr5aj9f58zp6fyjzup6ywn3x6sk8akg5v4tgn2q8g4fhx05wf6juaxu9760yp46454gpg5mtzgerlzezqcqvjnhjh8z3g2qqdhhwkj'
+        invoice = Invoice.from_bech32(bolt11)
+        a = QEAmount(from_invoice=invoice)
+        self.assertEqual(2_000_000, a.satsInt)
+        self.assertEqual(2_000_000_000, a.msatsInt)
+        self.assertFalse(a.isMax)

setup.py

@@ -45,6 +45,7 @@ extras_require = {
     'gui': ['pyqt5'],
     'crypto': ['cryptography>=2.6'],
     'tests': ['pycryptodomex>=3.7', 'cryptography>=2.6', 'pyaes>=0.1a1'],
+    'qml_gui': ['pyqt6', 'Pillow==8.4.0']
 }
 # 'full' extra that tries to grab everything an enduser would need (except for libsecp256k1...)
 extras_require['full'] = [pkg for sublist in

tox.ini

@@ -14,3 +14,4 @@ commands=
     coverage report
 extras=
     tests
+    qml_gui