aboutsummaryrefslogtreecommitdiffstats
path: root/sources/pyside6/tests/QtRemoteObjects
diff options
context:
space:
mode:
Diffstat (limited to 'sources/pyside6/tests/QtRemoteObjects')
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/CMakeLists.txt12
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/cpp_interop/CMakeLists.txt25
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop.cpp127
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop_test.py189
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/dynamic_types_test.py97
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/integration_test.py369
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/repfile_test.py65
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/simple.rep7
-rw-r--r--sources/pyside6/tests/QtRemoteObjects/test_shared.py126
9 files changed, 1016 insertions, 1 deletions
diff --git a/sources/pyside6/tests/QtRemoteObjects/CMakeLists.txt b/sources/pyside6/tests/QtRemoteObjects/CMakeLists.txt
index 2f7cb08b9..ace1a00fa 100644
--- a/sources/pyside6/tests/QtRemoteObjects/CMakeLists.txt
+++ b/sources/pyside6/tests/QtRemoteObjects/CMakeLists.txt
@@ -1 +1,11 @@
-# Please add some tests, here
+# Copyright (C) 2025 Ford Motor Company
+# SPDX-License-Identifier: BSD-3-Clause
+
+# FIXME: TypeError: Failed to generate default value. Error: name 'int' is not defined. Problematic code: int(2)
+if(NOT APPLE)
+PYSIDE_TEST(repfile_test.py)
+PYSIDE_TEST(dynamic_types_test.py)
+PYSIDE_TEST(integration_test.py)
+
+add_subdirectory(cpp_interop)
+endif()
diff --git a/sources/pyside6/tests/QtRemoteObjects/cpp_interop/CMakeLists.txt b/sources/pyside6/tests/QtRemoteObjects/cpp_interop/CMakeLists.txt
new file mode 100644
index 000000000..407a8f874
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/cpp_interop/CMakeLists.txt
@@ -0,0 +1,25 @@
+# Copyright (C) 2024 The Qt Company Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+
+set(CMAKE_INCLUDE_CURRENT_DIR ON)
+
+find_package(Qt6 REQUIRED COMPONENTS Core RemoteObjects)
+
+add_executable(cpp_interop ${MOC_SOURCES} cpp_interop.cpp)
+set_target_properties(cpp_interop PROPERTIES AUTOMOC ON)
+
+target_link_libraries(cpp_interop PUBLIC
+ Qt6::Core
+ Qt6::RemoteObjects
+)
+
+# Add a custom target to build the C++ program
+add_custom_target(build_cpp_interop
+ COMMAND ${CMAKE_COMMAND} --build ${CMAKE_BINARY_DIR} --target cpp_interop
+ WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
+)
+
+# Exclude windows (see cpp_interop.cpp)
+if(NOT WIN32)
+ PYSIDE_TEST(cpp_interop_test.py)
+endif() \ No newline at end of file
diff --git a/sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop.cpp b/sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop.cpp
new file mode 100644
index 000000000..6aeef91dd
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop.cpp
@@ -0,0 +1,127 @@
+// Copyright (C) 2025 Ford Motor Company
+// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
+
+#include <QtCore/qcoreapplication.h>
+#include <QtCore/qsocketnotifier.h>
+#include <QtCore/qtimer.h>
+
+#include <QtRemoteObjects/qremoteobjectreplica.h>
+#include <QtRemoteObjects/qremoteobjectnode.h>
+
+#ifdef Q_OS_WIN
+# include <QtCore/qt_windows.h>
+# include <QtCore/qwineventnotifier.h>
+#endif // Q_OS_WIN
+
+#include <iostream>
+
+using namespace Qt::StringLiterals;
+
+class CommandReader : public QObject
+{
+ Q_OBJECT
+public:
+ explicit CommandReader(QObject *parent = nullptr) : QObject(parent)
+ {
+#ifndef Q_OS_WIN
+ auto *notifier = new QSocketNotifier(fileno(stdin), QSocketNotifier::Read, this);
+ connect(notifier, &QSocketNotifier::activated, this, &CommandReader::handleInput);
+#else
+ // FIXME: Does not work, signals triggers too often, the app is stuck in getline()
+ auto notifier = new QWinEventNotifier(GetStdHandle(STD_INPUT_HANDLE), this);
+ connect(notifier, &QWinEventNotifier::activated, this, &CommandReader::handleInput);
+#endif
+ }
+
+signals:
+ void started();
+
+private slots:
+ void handleInput()
+ {
+ std::string line;
+ if (!std::getline(std::cin, line))
+ return;
+
+ if (line == "quit") {
+ std::cerr << "harness: Received quit. Stopping harness event loop.\n";
+ QCoreApplication::quit();
+ } else if (line == "start") {
+ std::cerr << "harness: Received start. Initializing harness nodes.\n";
+ emit started();
+ } else {
+ std::cerr << "harness: Unknown command \"" << line << "\"\n";
+ }
+ }
+};
+
+class Runner : public QObject
+{
+ Q_OBJECT
+public:
+ Runner(const QUrl &url, const QString &repName, QObject *parent = nullptr)
+ : QObject(parent)
+ , m_url(url)
+ , m_repName(repName)
+ {
+ m_host.setObjectName("cpp_host");
+ if (!m_host.setHostUrl(QUrl("tcp://127.0.0.1:0"_L1))) {
+ qWarning() << "harness: setHostUrl failed: " << m_host.lastError() << m_host.hostUrl();
+ std::cerr << "harness: Fatal harness error.\n";
+ QCoreApplication::exit(-2);
+ }
+
+ m_node.setObjectName("cpp_node");
+ std::cerr << "harness: Host url:" << m_host.hostUrl().toEncoded().constData() << '\n';
+ }
+
+public slots:
+ void onStart()
+ {
+ m_node.connectToNode(m_url);
+ m_replica.reset(m_node.acquireDynamic(m_repName));
+ if (!m_replica->waitForSource(1000)) {
+ std::cerr << "harness: Failed to acquire replica.\n";
+ QCoreApplication::exit(-1);
+ }
+
+ m_host.enableRemoting(m_replica.get());
+ }
+
+private:
+ QUrl m_url;
+ QString m_repName;
+ QRemoteObjectHost m_host;
+ QRemoteObjectNode m_node;
+ std::unique_ptr<QRemoteObjectDynamicReplica> m_replica;
+};
+
+int main(int argc, char *argv[])
+{
+ QCoreApplication a(argc, argv);
+ if (argc < 3) {
+ std::cerr << "Usage: " << argv[0] << " <url> <name of type>\n";
+ return -1;
+ }
+ QUrl url = QUrl::fromUserInput(QString::fromUtf8(argv[1]));
+ QString repName = QString::fromUtf8(argv[2]);
+
+ if (!url.isValid()) {
+ std::cerr << "Invalid URL: " << argv[1] << '\n';
+ return -1;
+ }
+
+ CommandReader reader;
+ Runner runner(url, repName);
+
+
+ QRemoteObjectNode node;
+ node.setObjectName("cpp_node");
+ std::unique_ptr<QRemoteObjectDynamicReplica> replica;
+
+ QObject::connect(&reader, &CommandReader::started, &runner, &Runner::onStart);
+
+ return QCoreApplication::exec();
+}
+
+#include "cpp_interop.moc"
diff --git a/sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop_test.py b/sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop_test.py
new file mode 100644
index 000000000..d9ab60c23
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/cpp_interop/cpp_interop_test.py
@@ -0,0 +1,189 @@
+#!/usr/bin/python
+# Copyright (C) 2025 Ford Motor Company
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+from __future__ import annotations
+
+'''Verify Python <--> C++ interop'''
+
+import os
+import sys
+import textwrap
+
+import unittest
+
+from pathlib import Path
+sys.path.append(os.fspath(Path(__file__).resolve().parents[2])) # For init_paths
+from init_paths import init_test_paths
+init_test_paths(False)
+
+from PySide6.QtCore import QUrl, QProcess, QObject, Signal
+from PySide6.QtRemoteObjects import (QRemoteObjectHost, QRemoteObjectNode, QRemoteObjectReplica,
+ RepFile)
+from PySide6.QtTest import QSignalSpy, QTest
+
+sys.path.append(os.fspath(Path(__file__).resolve().parents[1])) # For wrap_tests_for_cleanup
+from test_shared import wrap_tests_for_cleanup
+from helper.usesqapplication import UsesQApplication
+
+
+"""
+The previous tests all verify Remote Objects integration, but only
+using Python for both Source and Replica. We need to make sure there
+aren't any surprises in the interplay between Python and C++.
+
+This implements an initial test harness with a C++ app that is
+started by the Python unittest. We leverage the fact that Remote
+Objects can
+1) Allow remoting any QObject as a Source with enableRemoting
+2) Acquire Dynamic Replicas, where the definition needed for the
+ Replica is sent from the source.
+
+With these, we can create a working C++ app that doesn't need to be
+compiled with any information about the types being used. We have
+a host node in Python that shares a class derived from a RepFile
+Source type. The address of this node is passed to the C++ app via
+QProcess, and there a C++ node connects to that address to acquire
+(dynamically) a replica of the desired object.
+
+The C++ code also creates a host node and sends the address/port
+back to Python via the QProcess interface. Once the Python code
+receives the C++ side address and port, it connects a node to that
+URL and acquires the RepFile based type from Python.
+
+Python C++
+Host -----> Node (Dynamic acquire)
+ |
+ | Once initialized, the dynamic replica is
+ | shared (enable_remoting) from the C++ Host
+ |
+Node <----- Host
+"""
+
+
+def msg_cannot_start(process, executable):
+ return ('Cannot start "' + executable + '" in "'
+ + os.fspath(Path.cwd()) + '": ' + process.errorString())
+
+
+def stop_process(process):
+ result = process.waitForFinished(2000)
+ if not result:
+ process.kill()
+ result = process.waitForFinished(2000)
+ return result
+
+
+class Controller(QObject):
+ ready = Signal()
+
+ def __init__(self, utest: unittest.TestCase):
+ super().__init__()
+ # Store utest so we can make assertions
+ self.utest = utest
+
+ # Set up nodes
+ self.host = QRemoteObjectHost()
+ self.host.setObjectName("py_host")
+ self.host.setHostUrl(QUrl("tcp://127.0.0.1:0"))
+ self.cpp_url = None
+ self.node = QRemoteObjectNode()
+ self.node.setObjectName("py_node")
+ self._executable = "cpp_interop.exe" if os.name == "nt" else "./cpp_interop"
+
+ def start(self):
+ # Start the C++ application
+ self.process = QProcess()
+ self.process.readyReadStandardOutput.connect(self.process_harness_output)
+ self.process.readyReadStandardError.connect(self.process_harness_output)
+ urls = self.host.hostUrl().toDisplayString()
+ print(f'Starting C++ application "{self._executable}" "{urls}"', file=sys.stderr)
+ self.process.start(self._executable, [self.host.hostUrl().toDisplayString(), "Simple"])
+ self.utest.assertTrue(self.process.waitForStarted(2000),
+ msg_cannot_start(self.process, self._executable))
+
+ # Wait for the C++ application to output the host url
+ spy = QSignalSpy(self.ready)
+ self.utest.assertTrue(spy.wait(1000))
+ self.utest.assertTrue(self.cpp_url.isValid())
+
+ self.utest.assertTrue(self.node.connectToNode(self.cpp_url))
+ return True
+
+ def stop(self):
+ if self.process.state() == QProcess.ProcessState.Running:
+ print(f'Stopping C++ application "{self._executable}" {self.process.processId()}',
+ file=sys.stderr)
+ self.process.write("quit\n".encode())
+ self.process.closeWriteChannel()
+ self.utest.assertTrue(stop_process(self.process))
+ self.utest.assertEqual(self.process.exitStatus(), QProcess.ExitStatus.NormalExit)
+
+ def add_source(self, Source, Replica):
+ """
+ Source and Replica are types.
+
+ Replica is from the rep file
+ Source is a class derived from the rep file's Source type
+ """
+ self.process.write("start\n".encode())
+ source = Source()
+ self.host.enableRemoting(source)
+ replica = self.node.acquire(Replica)
+ self.utest.assertTrue(replica.waitForSource(5000))
+ self.utest.assertEqual(replica.state(), QRemoteObjectReplica.State.Valid)
+ return source, replica
+
+ def process_harness_output(self):
+ '''Process stderr from the C++ application'''
+ output = self.process.readAllStandardError().trimmed()
+ lines = output.data().decode().split("\n")
+ HOST_LINE = "harness: Host url:"
+ for line in lines:
+ print(line, file=sys.stderr)
+ if line.startswith(HOST_LINE):
+ urls = line[len(HOST_LINE):].strip()
+ print(f'url="{urls}"', file=sys.stderr)
+ self.cpp_url = QUrl(urls)
+ self.ready.emit()
+
+
+class HarnessTest(UsesQApplication):
+ def setUp(self):
+ super().setUp()
+ self.rep = RepFile(self.__class__.contents)
+ self.controller = Controller(self)
+ self.assertTrue(self.controller.start())
+
+ def tearDown(self):
+ self.controller.stop()
+ self.app.processEvents()
+ super().tearDown()
+ QTest.qWait(100) # Wait for 100 msec
+
+
+@wrap_tests_for_cleanup(extra=['rep'])
+class TestBasics(HarnessTest):
+ contents = textwrap.dedent("""\
+ class Simple
+ {
+ PROP(int i = 2);
+ PROP(float f = -1. READWRITE);
+ }
+ """)
+
+ def compare_properties(self, instance, values):
+ '''Compare properties of instance with values'''
+ self.assertEqual(instance.i, values[0])
+ self.assertAlmostEqual(instance.f, values[1], places=5)
+
+ def testInitialization(self):
+ '''Test constructing RepFile from a path string'''
+ class Source(self.rep.source["Simple"]):
+ pass
+ source, replica = self.controller.add_source(Source, self.rep.replica["Simple"])
+ self.compare_properties(source, [2, -1])
+ self.compare_properties(replica, [2, -1])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/sources/pyside6/tests/QtRemoteObjects/dynamic_types_test.py b/sources/pyside6/tests/QtRemoteObjects/dynamic_types_test.py
new file mode 100644
index 000000000..5fb828cd2
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/dynamic_types_test.py
@@ -0,0 +1,97 @@
+#!/usr/bin/python
+# Copyright (C) 2025 Ford Motor Company
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+from __future__ import annotations
+
+'''Test cases for dynamic source/replica types'''
+
+import os
+import sys
+import unittest
+from pathlib import Path
+sys.path.append(os.fspath(Path(__file__).resolve().parents[1]))
+from init_paths import init_test_paths
+init_test_paths(False)
+
+from PySide6.QtRemoteObjects import RepFile
+
+from test_shared import wrap_tests_for_cleanup
+
+
+contents = """
+class Simple
+{
+ PROP(int i = 2);
+ PROP(float f = -1. READWRITE);
+ SIGNAL(random(int i));
+ SLOT(void reset());
+};
+"""
+
+
+@wrap_tests_for_cleanup(extra=['rep_file'])
+class QDynamicReplicas(unittest.TestCase):
+ '''Test case for dynamic Replicas'''
+
+ def setUp(self):
+ '''Set up test environment'''
+ self.rep_file = RepFile(contents)
+
+ def testDynamicReplica(self):
+ '''Verify that a valid Replica is created'''
+ Replica = self.rep_file.replica["Simple"]
+ self.assertIsNotNone(Replica)
+ replica = Replica()
+ self.assertIsNotNone(replica)
+ self.assertIsNotNone(replica.metaObject())
+ meta = replica.metaObject()
+ self.assertEqual(meta.className(), "Simple")
+ self.assertEqual(meta.superClass().className(), "QRemoteObjectReplica")
+ i = meta.indexOfProperty("i")
+ self.assertNotEqual(i, -1)
+ self.assertEqual(replica.propAsVariant(0), int(2))
+ self.assertEqual(replica.propAsVariant(1), float(-1.0))
+ self.assertEqual(replica.i, int(2))
+ self.assertEqual(replica.f, float(-1.0))
+
+
+@wrap_tests_for_cleanup(extra=['rep_file'])
+class QDynamicSources(unittest.TestCase):
+ '''Test case for dynamic Sources'''
+
+ def setUp(self):
+ '''Set up test environment'''
+ self.rep_file = RepFile(contents)
+ self.test_val = 0
+
+ def on_changed(self, val):
+ self.test_val = val
+
+ def testDynamicSource(self):
+ '''Verify that a valid Source is created'''
+ Source = self.rep_file.source["Simple"]
+ self.assertIsNotNone(Source)
+ source = Source()
+ self.assertIsNotNone(source)
+ self.assertIsNotNone(source.metaObject())
+ meta = source.metaObject()
+ self.assertEqual(meta.className(), "SimpleSource")
+ self.assertEqual(meta.superClass().className(), "QObject")
+ i = meta.indexOfProperty("i")
+ self.assertNotEqual(i, -1)
+ self.assertIsNotNone(source.__dict__.get('__PROPERTIES__'))
+ self.assertEqual(source.i, int(2))
+ self.assertEqual(source.f, float(-1.0))
+ source.iChanged.connect(self.on_changed)
+ source.fChanged.connect(self.on_changed)
+ source.i = 7
+ self.assertEqual(source.i, int(7))
+ self.assertEqual(self.test_val, int(7))
+ source.i = 3
+ self.assertEqual(self.test_val, int(3))
+ source.f = 3.14
+ self.assertAlmostEqual(self.test_val, float(3.14), places=5)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/sources/pyside6/tests/QtRemoteObjects/integration_test.py b/sources/pyside6/tests/QtRemoteObjects/integration_test.py
new file mode 100644
index 000000000..69b4930da
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/integration_test.py
@@ -0,0 +1,369 @@
+#!/usr/bin/python
+# Copyright (C) 2025 Ford Motor Company
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+from __future__ import annotations
+
+'''Test cases for basic Source/Replica communication'''
+
+import os
+import sys
+import textwrap
+import enum
+import gc
+
+import unittest
+
+from pathlib import Path
+sys.path.append(os.fspath(Path(__file__).resolve().parents[1]))
+from init_paths import init_test_paths
+init_test_paths(False)
+
+from PySide6.QtCore import QUrl, qWarning
+from PySide6.QtRemoteObjects import (QRemoteObjectHost, QRemoteObjectNode, QRemoteObjectReplica,
+ QRemoteObjectPendingCall, RepFile, getCapsuleCount)
+from PySide6.QtTest import QSignalSpy, QTest
+
+from test_shared import wrap_tests_for_cleanup
+from helper.usesqapplication import UsesQApplication
+
+contents = """
+class Simple
+{
+ PROP(int i = 2);
+ PROP(float f = -1. READWRITE);
+ SIGNAL(random(int i));
+ SLOT(void reset());
+ SLOT(int add(int i));
+};
+"""
+
+
+class QBasicTest(UsesQApplication):
+ '''Test case for basic source/replica communication'''
+ def setUp(self):
+ # Separate output to make debugging easier
+ qWarning(f"\nSet up {self.__class__.__qualname__}")
+ super().setUp()
+ '''Set up test environment'''
+ if hasattr(self.__class__, "contents"):
+ qWarning(f"Using class contents >{self.__class__.contents}<")
+ self.rep = RepFile(self.__class__.contents)
+ else:
+ self.rep = RepFile(contents)
+ self.host = QRemoteObjectHost(QUrl("tcp://127.0.0.1:0"))
+ self.host.setObjectName("host")
+ self.node = QRemoteObjectNode()
+ self.node.setObjectName("node")
+ self.node.connectToNode(self.host.hostUrl()) # pick up the url with the assigned port
+
+ def compare_properties(self, instance, values):
+ '''Compare properties of instance with values'''
+ self.assertEqual(instance.i, values[0])
+ self.assertAlmostEqual(instance.f, values[1], places=5)
+
+ def default_setup(self):
+ '''Set up default test environment'''
+ replica = self.node.acquire(self.rep.replica["Simple"])
+ # Make sure the replica is initialized with default values
+ self.compare_properties(replica, [2, -1])
+ self.assertEqual(replica.isInitialized(), False)
+ source = self.rep.source["Simple"]()
+ # Make sure the source is initialized with default values
+ self.compare_properties(source, [2, -1])
+ return replica, source
+
+ def tearDown(self):
+ self.assertEqual(getCapsuleCount(), 0)
+ self.app.processEvents()
+ super().tearDown()
+ # Separate output to make debugging easier
+ qWarning(f"Tore down {self.__class__.__qualname__}\n")
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class ReplicaInitialization(QBasicTest):
+ def test_ReplicaInitialization(self):
+ replica, source = self.default_setup()
+ source.i = -1
+ source.f = 3.14
+ self.compare_properties(source, [-1, 3.14])
+ init_spy = QSignalSpy(replica.initialized)
+ self.host.enableRemoting(source)
+ self.assertEqual(replica.waitForSource(1000), True)
+ self.assertEqual(replica.state(), QRemoteObjectReplica.State.Valid)
+ # Make sure the replica values are updated to the source values
+ self.compare_properties(replica, [-1, 3.14])
+ self.assertEqual(init_spy.count(), 1)
+ self.assertEqual(replica.isInitialized(), True)
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class SourcePropertyChange(QBasicTest):
+ def test_SourcePropertyChange(self):
+ replica, source = self.default_setup()
+ self.host.enableRemoting(source)
+ self.assertEqual(replica.waitForSource(1000), True)
+ # Make sure the replica values are unchanged since the source had the same values
+ self.compare_properties(replica, [2, -1])
+ source_spy = QSignalSpy(source.iChanged)
+ replica_spy = QSignalSpy(replica.iChanged)
+ source.i = 42
+ self.assertEqual(source_spy.count(), 1)
+ # Make sure the source value is updated
+ self.compare_properties(source, [42, source.f])
+ self.assertTrue(replica_spy.wait(1000))
+ self.assertEqual(replica_spy.count(), 1)
+ # Make sure the replica value is updated
+ self.compare_properties(replica, [42, replica.f])
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class ReplicaPropertyChange(QBasicTest):
+ def test_ReplicaPropertyChange(self):
+ replica, source = self.default_setup()
+ self.host.enableRemoting(source)
+ self.assertEqual(replica.waitForSource(1000), True)
+ # Make sure push methods are working
+ source_spy = QSignalSpy(source.iChanged)
+ replica_spy = QSignalSpy(replica.iChanged)
+ replica.pushI(11)
+ # # Let eventloop run to update the source and verify the values
+ self.assertTrue(source_spy.wait(1000))
+ self.assertEqual(source_spy.count(), 1)
+ self.compare_properties(source, [11, source.f])
+ # Let eventloop run to update the replica and verify the values
+ self.assertTrue(replica_spy.wait(1000))
+ self.assertEqual(replica_spy.count(), 1)
+ self.compare_properties(replica, [11, replica.f])
+
+ # Test setter on replica
+ source_spy = QSignalSpy(source.fChanged)
+ replica_spy = QSignalSpy(replica.fChanged)
+ replica.f = 4.2
+ # Make sure the replica values are ** NOT CHANGED ** since the eventloop hasn't run
+ self.compare_properties(replica, [11, -1])
+ # Let eventloop run to update the source and verify the values
+ self.assertTrue(source_spy.wait(1000))
+ self.assertEqual(source_spy.count(), 1)
+ self.compare_properties(source, [source.i, 4.2])
+ # Let eventloop run to update the replica and verify the values
+ self.assertTrue(replica_spy.wait(1000))
+ self.assertEqual(replica_spy.count(), 1)
+ self.compare_properties(replica, [replica.i, 4.2])
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class DerivedReplicaPropertyChange(QBasicTest):
+ def test_DerivedReplicaPropertyChange(self):
+ # Don't use default_setup(), instead create a derived replica
+ Replica = self.rep.replica["Simple"]
+
+ class DerivedReplica(Replica):
+ pass
+
+ replica = self.node.acquire(DerivedReplica)
+ # Make sure the replica is initialized with default values
+ self.compare_properties(replica, [2, -1])
+ self.assertEqual(replica.isInitialized(), False)
+ source = self.rep.source["Simple"]()
+ self.host.enableRemoting(source)
+ self.assertEqual(replica.waitForSource(1000), True)
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class ReplicaSlotNotImplementedChange(QBasicTest):
+ def test_ReplicaSlotNotImplementedChange(self):
+ replica, source = self.default_setup()
+ self.host.enableRemoting(source)
+ self.assertEqual(replica.waitForSource(1000), True)
+ # Ideally this would fail as the slot is not implemented on the source
+ res = replica.reset()
+ self.assertEqual(type(res), type(None))
+ QTest.qWait(100) # Wait for 100 ms for async i/o. There isn't a signal to wait on
+ res = replica.add(5)
+ self.assertEqual(type(res), QRemoteObjectPendingCall)
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class ReplicaSlotImplementedChange(QBasicTest):
+ def test_ReplicaSlotImplementedChange(self):
+ replica = self.node.acquire(self.rep.replica["Simple"])
+ replica.setObjectName("replica")
+
+ class Source(self.rep.source["Simple"]):
+ def __init__(self):
+ super().__init__()
+ self.i = 6
+ self.f = 3.14
+
+ def reset(self):
+ self.i = 0
+ self.f = 0
+
+ def add(self, i):
+ return self.i + i
+
+ source = Source()
+ source.setObjectName("source")
+ self.host.enableRemoting(source)
+ self.assertEqual(replica.waitForSource(1000), True)
+ self.compare_properties(source, [6, 3.14])
+ self.compare_properties(replica, [6, 3.14])
+ replica_spy = QSignalSpy(replica.iChanged)
+ res = replica.reset()
+ self.assertEqual(type(res), type(None))
+ self.assertEqual(replica_spy.wait(1000), True)
+ self.compare_properties(source, [0, 0])
+ self.compare_properties(replica, [0, 0])
+ res = replica.add(5)
+ self.assertEqual(type(res), QRemoteObjectPendingCall)
+ self.assertEqual(res.waitForFinished(1000), True)
+ self.assertEqual(res.returnValue(), 5)
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class RefCountTest(QBasicTest):
+ contents = textwrap.dedent("""\
+ POD MyPOD{
+ ENUM class Position : unsigned short {position1=1, position2=2, position3=4}
+ Position pos,
+ QString name
+ }
+ class Simple
+ {
+ ENUM Position {Left, Right, Top, Bottom}
+ PROP(MyPOD myPod);
+ PROP(Position pos);
+ }
+ """)
+
+ def test_RefCount(self):
+ # Once the rep file is loaded, we should be tracking 4 converter capsules
+ # - 1 for the POD itself
+ # - 1 for the enum in the POD
+ # - 1 for the enum in the Source
+ # - 1 for the enum in the Replica
+ # We should be tracking 3 qobject capsules (POD, Replica, Source)
+ # Note: Source and Replica are distinct types, so Source::EPosition and
+ # Replica::EPosition are distinct as well.
+ # Note 2: The name of the enum ("Position") can be reused for different
+ # types in different classes as shown above.
+ self.assertEqual(getCapsuleCount(), 7)
+ MyPod = self.rep.pod["MyPOD"]
+ self.assertTrue(isinstance(MyPod, type))
+ self.assertTrue(issubclass(MyPod, tuple))
+ MyEnum = MyPod.get_enum("Position")
+ self.assertTrue(isinstance(MyEnum, type))
+ self.assertTrue(issubclass(MyEnum, enum.Enum))
+ e = MyEnum(4) # noqa: F841
+ Source = self.rep.source["Simple"]
+ source = Source() # noqa: F841
+ source = None # noqa: F841
+ Source = None
+ Replica = self.rep.replica["Simple"]
+ replica = self.node.acquire(Replica) # noqa: F841
+ replica = None # noqa: F841
+ Replica = None
+ MyEnum = None
+ MyPod = None
+ self.rep = None
+ e = None # noqa: F841
+ gc.collect()
+ # The enum and POD capsules will only be deleted (garbage collected) if
+ # the types storing them (RepFile, Replica and Source) are garbage
+ # collected first.
+ self.assertEqual(getCapsuleCount(), 0)
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class EnumTest(QBasicTest):
+ contents = textwrap.dedent("""\
+ POD MyPOD{
+ ENUM class Position : unsigned short {position1=1, position2=2, position3=4}
+ Position pos,
+ QString name
+ }
+ class Simple
+ {
+ ENUM Position {Left, Right, Top, Bottom}
+ PROP(MyPOD myPod);
+ PROP(Position pos);
+ }
+ """)
+
+ def test_Enum(self):
+ MyPod = self.rep.pod["MyPOD"]
+ self.assertTrue(isinstance(MyPod, type))
+ self.assertTrue(issubclass(MyPod, tuple))
+ PodEnum = MyPod.get_enum("Position")
+ self.assertTrue(isinstance(PodEnum, type))
+ self.assertTrue(issubclass(PodEnum, enum.Enum))
+ t = (PodEnum(4), "test")
+ myPod = MyPod(*t)
+ with self.assertRaises(ValueError):
+ myPod = MyPod(PodEnum(0), "thing") # 0 isn't a valid enum value
+ myPod = MyPod(PodEnum(2), "thing")
+ self.assertEqual(myPod.pos, PodEnum.position2)
+ replica = self.node.acquire(self.rep.replica["Simple"])
+ replica.setObjectName("replica")
+ source = self.rep.source["Simple"]()
+ source.setObjectName("source")
+ source.myPod = (PodEnum.position2, "Hello")
+ SourceEnum = source.get_enum("Position")
+ self.assertTrue(isinstance(SourceEnum, type))
+ self.assertTrue(issubclass(SourceEnum, enum.Enum))
+ source.pos = SourceEnum.Top
+ self.assertEqual(source.myPod, (PodEnum.position2, "Hello"))
+ self.assertNotEqual(source.pos, 2)
+ self.host.enableRemoting(source)
+ self.assertEqual(replica.waitForSource(1000), True)
+ self.assertEqual(replica.myPod, (PodEnum.position2, "Hello"))
+ ReplicaEnum = replica.get_enum("Position")
+ # Test invalid comparisons
+ self.assertNotEqual(replica.pos, 2)
+ self.assertNotEqual(replica.pos, SourceEnum.Top)
+ self.assertNotEqual(replica.myPod, (SourceEnum(2), "Hello"))
+ self.assertNotEqual(replica.myPod, (ReplicaEnum(2), "Hello"))
+ self.assertNotEqual(replica.myPod, (2, "Hello"))
+ # Test valid comparisons to Replica enum
+ self.assertEqual(replica.pos, ReplicaEnum.Top)
+ self.assertEqual(replica.myPod, (PodEnum(2), "Hello"))
+
+
+@wrap_tests_for_cleanup(extra=['rep', 'host', 'node'])
+class PodTest(QBasicTest):
+ contents = textwrap.dedent("""\
+ POD MyPod(int i, QString s)
+
+ class Simple
+ {
+ PROP(MyPod pod);
+ }
+ """)
+
+ def test_Pod(self):
+ MyPod = self.rep.pod["MyPod"]
+ self.assertTrue(isinstance(MyPod, type))
+ self.assertTrue(issubclass(MyPod, tuple))
+ source = self.rep.source["Simple"]()
+ t = (42, "Hello")
+ pod = MyPod(*t)
+ source.pod = t
+ self.assertEqual(source.pod, t)
+ self.assertEqual(source.pod, pod)
+ source.pod = pod
+ self.assertEqual(source.pod, t)
+ self.assertEqual(source.pod, pod)
+ with self.assertRaises(ValueError):
+ source.pod = (11, "World", "!")
+ with self.assertRaises(TypeError):
+ source.pod = MyPod("Hello", "World")
+ self.assertEqual(source.pod, pod)
+ self.assertTrue(isinstance(pod, MyPod))
+ self.assertEqual(pod.i, 42)
+ self.assertEqual(pod.s, "Hello")
+ self.assertTrue(isinstance(source.pod, MyPod))
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/sources/pyside6/tests/QtRemoteObjects/repfile_test.py b/sources/pyside6/tests/QtRemoteObjects/repfile_test.py
new file mode 100644
index 000000000..b73c84f3a
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/repfile_test.py
@@ -0,0 +1,65 @@
+#!/usr/bin/python
+# Copyright (C) 2025 Ford Motor Company
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+from __future__ import annotations
+
+'''Test cases for RepFile'''
+
+import os
+import sys
+import unittest
+from pathlib import Path
+sys.path.append(os.fspath(Path(__file__).resolve().parents[1]))
+from init_paths import init_test_paths
+init_test_paths(False)
+from PySide6.QtRemoteObjects import RepFile
+
+from test_shared import wrap_tests_for_cleanup
+
+contents = """
+class Simple
+{
+ PROP(int i = 2);
+ PROP(float f = -1. READWRITE);
+ SIGNAL(random(int i));
+ SLOT(void reset());
+};
+"""
+
+
+@wrap_tests_for_cleanup()
+class QRepFileConstructor(unittest.TestCase):
+ '''Test case for RepFile constructors'''
+ expected = "RepFile(Classes: [Simple], PODs: [])"
+
+ def setUp(self):
+ '''Set up test environment'''
+ self.cwd = Path(__file__).parent
+ self.path = self.cwd / "simple.rep"
+
+ def testRepFileFromPath(self):
+ '''Test constructing RepFile from a path'''
+ with open(self.path, 'r') as f:
+ rep_file = RepFile(f.read())
+ self.assertEqual(str(rep_file), self.expected)
+
+ def testRepFileFromString(self):
+ '''Test constructing RepFile from a string'''
+ rep_file = RepFile(contents)
+ self.assertEqual(str(rep_file), self.expected)
+
+ def testRepFileInvalidString(self):
+ '''Test constructing RepFile from a string'''
+ with self.assertRaises(RuntimeError) as result:
+ RepFile("\n\n}\n\n")
+ self.assertEqual(str(result.exception),
+ "Error parsing input, line 3: error: Unknown token encountered")
+
+ def testRepFileNoArguments(self):
+ '''Test constructing RepFile with no arguments'''
+ with self.assertRaises(TypeError):
+ RepFile()
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/sources/pyside6/tests/QtRemoteObjects/simple.rep b/sources/pyside6/tests/QtRemoteObjects/simple.rep
new file mode 100644
index 000000000..7e801a8c6
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/simple.rep
@@ -0,0 +1,7 @@
+class Simple
+{
+ PROP(int i = 2);
+ PROP(float f = -1. READWRITE);
+ SIGNAL(random(int i));
+ SLOT(void reset());
+};
diff --git a/sources/pyside6/tests/QtRemoteObjects/test_shared.py b/sources/pyside6/tests/QtRemoteObjects/test_shared.py
new file mode 100644
index 000000000..5b176ce9d
--- /dev/null
+++ b/sources/pyside6/tests/QtRemoteObjects/test_shared.py
@@ -0,0 +1,126 @@
+# Copyright (C) 2025 Ford Motor Company
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+from __future__ import annotations
+
+import gc
+import sys
+from functools import wraps
+
+
+def _cleanup_local_variables(self, extra, debug):
+ """
+ Function to clean up local variables after a unit test.
+
+ This method will set any local variables defined in the test run to None. It also
+ sets variables of self to None, if they are provided in the extra list.
+
+ The self argument is passed by the decorator, so we can access the instance variables.
+ """
+ local_vars = self._locals
+ if debug:
+ print(f" Cleaning up locals: {local_vars.keys()} and member of self: {extra}",
+ file=sys.stderr)
+ exclude_vars = {'__builtins__', 'self', 'args', 'kwargs'}
+ for var in list(local_vars.keys()):
+ if var not in exclude_vars:
+ local_vars[var] = None
+ if debug:
+ print(f" Set {var} to None", file=sys.stderr)
+ # Remove variables added to 'self' during our test
+ for var in list(vars(self).keys()):
+ if var in extra:
+ setattr(self, var, None)
+ if debug:
+ print(f" Set self.{var} to None", file=sys.stderr)
+ gc.collect()
+
+
+# This leverages the tip from # https://stackoverflow.com/a/9187022/169296
+# for capturing local variables using sys.setprofile and a tracer function
+def wrap_tests_for_cleanup(extra: str | list[str] = None, debug: bool = False):
+ """
+ Method that returns a decorator for setting variables used in a test to
+ None, thus allowing the garbage collection to clean up properly and ensure
+ destruction behavior is correct. Using a method to return the decorator
+ allows us to pass extra arguments to the decorator, in this case for extra
+ data members on `self` to set to None or whether to output additional debug
+ logging.
+
+ It simply returns the class decorator to be used.
+ """
+ def decorator(cls):
+ """
+ This is a class decorator that finds and wraps all test methods in a
+ class.
+
+ The provided extra is used to define a set() of variables that are set
+ to None on `self` after the test method has run. This is useful for
+ making sure the local and self variables can be garbage collected.
+ """
+ _extra = set()
+ if extra:
+ if isinstance(extra, str):
+ _extra.add(extra)
+ else:
+ _extra.update(extra)
+ for name, attr in cls.__dict__.items():
+ if name.startswith("test") and callable(attr):
+ """
+ Only wrap methods that start with 'test' and are callable.
+ """
+ def make_wrapper(method):
+ """
+ This is the actual wrapper that will be used to wrap the
+ test methods. It will set a tracer function to capture the
+ local variables and then calls our cleanup function to set
+ the variables to None.
+ """
+ @wraps(method)
+ def wrapper(self, *args, **kwargs):
+ if debug:
+ print(f"wrap_tests_for_cleanup - calling {method.__name__}",
+ file=sys.stderr)
+
+ def tracer(frame, event, arg):
+ if event == 'return':
+ self._locals = frame.f_locals.copy()
+
+ # tracer is activated on next call, return or exception
+ sys.setprofile(tracer)
+ try:
+ # trace the function call
+ return method(self, *args, **kwargs)
+ finally:
+ # disable tracer and replace with old one
+ sys.setprofile(None)
+ # call our cleanup function
+ _cleanup_local_variables(self, _extra, debug)
+ if debug:
+ print(f"wrap_tests_for_cleanup - done calling {method.__name__}",
+ file=sys.stderr)
+ return wrapper
+ setattr(cls, name, make_wrapper(attr))
+ return cls
+ return decorator
+
+
+if __name__ == "__main__":
+ # Set up example test class
+ @wrap_tests_for_cleanup(extra="name", debug=True)
+ class test:
+ def __init__(self):
+ self.name = "test"
+
+ def testStuff(self):
+ value = 42
+ raise ValueError("Test")
+ temp = 11 # noqa: F841
+ return value
+
+ t = test()
+ try:
+ t.testStuff()
+ except ValueError:
+ pass
+ # Should print that `value` and `self.name` are set to None, even with the
+ # exception being raised.