Addon Manager: Refactoring and pylint cleanup
This commit is contained in:
@@ -30,8 +30,6 @@ import time
|
||||
from zipfile import ZipFile
|
||||
import FreeCAD
|
||||
|
||||
from typing import Dict
|
||||
|
||||
from addonmanager_git import GitManager, NoGitFound, GitFailed
|
||||
|
||||
|
||||
@@ -89,6 +87,7 @@ class TestGit(unittest.TestCase):
|
||||
status = self.git.status(checkout_dir).strip()
|
||||
expected_status = "## HEAD (no branch)"
|
||||
self.assertEqual(status, expected_status)
|
||||
|
||||
self.assertEqual(os.getcwd(),self.cwd, "We should be left in the same CWD we started")
|
||||
|
||||
def test_update(self):
|
||||
|
||||
@@ -25,9 +25,10 @@
|
||||
import unittest
|
||||
import FreeCAD
|
||||
|
||||
|
||||
class TestGui(unittest.TestCase):
|
||||
|
||||
MODULE = 'test_gui' # file name without extension
|
||||
MODULE = "test_gui" # file name without extension
|
||||
|
||||
def setUp(self):
|
||||
pass
|
||||
pass
|
||||
|
||||
@@ -22,15 +22,190 @@
|
||||
# * *
|
||||
# ***************************************************************************
|
||||
|
||||
import unittest
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import tempfile
|
||||
import unittest
|
||||
import FreeCAD
|
||||
from addonmanager_git import initialize_git
|
||||
|
||||
from PySide2 import QtCore
|
||||
|
||||
import NetworkManager
|
||||
from Addon import Addon
|
||||
from addonmanager_workers_startup import (
|
||||
CreateAddonListWorker,
|
||||
UpdateChecker,
|
||||
)
|
||||
from addonmanager_workers_installation import InstallWorkbenchWorker
|
||||
|
||||
|
||||
class TestWorkersInstallation(unittest.TestCase):
|
||||
|
||||
MODULE = "test_workers_installation" # file name without extension
|
||||
|
||||
addon_list = (
|
||||
[]
|
||||
) # Cache at the class level so only the first test has to download it
|
||||
|
||||
def setUp(self):
|
||||
pass
|
||||
"""Set up the test"""
|
||||
self.test_dir = os.path.join(
|
||||
FreeCAD.getHomePath(), "Mod", "AddonManager", "AddonManagerTest", "data"
|
||||
)
|
||||
|
||||
self.saved_mod_directory = Addon.mod_directory
|
||||
self.saved_cache_directory = Addon.cache_directory
|
||||
Addon.mod_directory = os.path.join(
|
||||
tempfile.gettempdir(), "FreeCADTesting", "Mod"
|
||||
)
|
||||
Addon.cache_directory = os.path.join(
|
||||
tempfile.gettempdir(), "FreeCADTesting", "Cache"
|
||||
)
|
||||
|
||||
os.makedirs(Addon.mod_directory, mode=0o777, exist_ok=True)
|
||||
os.makedirs(Addon.cache_directory, mode=0o777, exist_ok=True)
|
||||
|
||||
url = "https://api.github.com/zen"
|
||||
NetworkManager.InitializeNetworkManager()
|
||||
result = NetworkManager.AM_NETWORK_MANAGER.blocking_get(url)
|
||||
if result is None:
|
||||
self.skipTest("No active internet connection detected")
|
||||
|
||||
self.macro_counter = 0
|
||||
self.workbench_counter = 0
|
||||
self.prefpack_counter = 0
|
||||
self.addon_from_cache_counter = 0
|
||||
self.macro_from_cache_counter = 0
|
||||
|
||||
self.package_cache = {}
|
||||
self.macro_cache = []
|
||||
|
||||
self.package_cache_filename = os.path.join(
|
||||
Addon.cache_directory, "packages.json"
|
||||
)
|
||||
self.macro_cache_filename = os.path.join(Addon.cache_directory, "macros.json")
|
||||
|
||||
if not TestWorkersInstallation.addon_list:
|
||||
self._create_addon_list()
|
||||
|
||||
# Workbench: use the FreeCAD-Help workbench for testing purposes
|
||||
self.help_addon = None
|
||||
for addon in self.addon_list:
|
||||
if addon.name == "Help":
|
||||
self.help_addon = addon
|
||||
break
|
||||
if not self.help_addon:
|
||||
print("Unable to locate the FreeCAD-Help addon to test with")
|
||||
self.skipTest("No active internet connection detected")
|
||||
|
||||
# Store the user's preference for whether git is enabled or disabled
|
||||
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
|
||||
self.saved_git_disabled_status = pref.GetBool("disableGit", False)
|
||||
|
||||
def tearDown(self):
|
||||
mod_dir = os.path.join(tempfile.gettempdir(), "FreeCADTesting", "Mod")
|
||||
if os.path.exists(mod_dir):
|
||||
self._rmdir(mod_dir)
|
||||
macro_dir = os.path.join(tempfile.gettempdir(), "FreeCADTesting", "Mod")
|
||||
if os.path.exists(macro_dir):
|
||||
self._rmdir(macro_dir)
|
||||
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
|
||||
pref.SetBool("disableGit", self.saved_git_disabled_status)
|
||||
|
||||
def test_workbench_installation(self):
|
||||
addon_location = os.path.join(
|
||||
tempfile.gettempdir(), "FreeCADTesting", "Mod", self.help_addon.name
|
||||
)
|
||||
worker = InstallWorkbenchWorker(self.help_addon, addon_location)
|
||||
worker.run() # Synchronous call, blocks until complete
|
||||
self.assertTrue(os.path.exists(addon_location))
|
||||
self.assertTrue(os.path.exists(os.path.join(addon_location, "package.xml")))
|
||||
|
||||
def test_workbench_installation_git_disabled(self):
|
||||
"""If the testing user has git enabled, also test the addon manager with git disabled"""
|
||||
if self.saved_git_disabled_status:
|
||||
self.skipTest("Git is disabled, this test is redundant")
|
||||
|
||||
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
|
||||
pref.SetBool("disableGit", True)
|
||||
|
||||
self.test_workbench_installation()
|
||||
|
||||
pref.SetBool("disableGit", False)
|
||||
|
||||
def test_workbench_update_checker(self):
|
||||
|
||||
git_manager = initialize_git()
|
||||
|
||||
if not git_manager:
|
||||
return
|
||||
|
||||
# Workbench: use the FreeCAD-Help workbench for testing purposes
|
||||
help_addon = None
|
||||
for addon in self.addon_list:
|
||||
if addon.name == "Help":
|
||||
help_addon = addon
|
||||
break
|
||||
if not help_addon:
|
||||
print("Unable to locate the FreeCAD-Help addon to test with")
|
||||
return
|
||||
|
||||
addon_location = os.path.join(
|
||||
tempfile.gettempdir(), "FreeCADTesting", "Mod", self.help_addon.name
|
||||
)
|
||||
worker = InstallWorkbenchWorker(addon, addon_location)
|
||||
worker.run() # Synchronous call, blocks until complete
|
||||
self.assertEqual(help_addon.status(), Addon.Status.PENDING_RESTART)
|
||||
|
||||
# Back up one revision
|
||||
git_manager.reset(addon_location, ["--hard", "HEAD~1"])
|
||||
|
||||
# At this point the addon should be "out of date", checked out to one revision behind
|
||||
# the most recent.
|
||||
|
||||
worker = UpdateChecker()
|
||||
worker.override_mod_directory(
|
||||
os.path.join(tempfile.gettempdir(), "FreeCADTesting", "Mod")
|
||||
)
|
||||
worker.check_workbench(help_addon) # Synchronous call
|
||||
self.assertEqual(help_addon.status(), Addon.Status.UPDATE_AVAILABLE)
|
||||
|
||||
# Now try to "update" it (which is really done via the install worker)
|
||||
worker = InstallWorkbenchWorker(addon, addon_location)
|
||||
worker.run() # Synchronous call, blocks until complete
|
||||
self.assertEqual(help_addon.status(), Addon.Status.PENDING_RESTART)
|
||||
|
||||
def _rmdir(self, path):
|
||||
try:
|
||||
shutil.rmtree(path, onerror=self._remove_readonly)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
def _remove_readonly(self, func, path, _) -> None:
|
||||
"""Remove a read-only file."""
|
||||
|
||||
os.chmod(path, stat.S_IWRITE)
|
||||
func(path)
|
||||
|
||||
def _create_addon_list(self):
|
||||
"""Create the list of addons"""
|
||||
worker = CreateAddonListWorker()
|
||||
worker.addon_repo.connect(self._addon_added)
|
||||
worker.start()
|
||||
while worker.isRunning():
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
|
||||
def _addon_added(self, addon: Addon):
|
||||
"""Callback for adding an Addon: tracks the list, and counts the various types"""
|
||||
print(f"Addon added: {addon.name}")
|
||||
TestWorkersInstallation.addon_list.append(addon)
|
||||
if addon.contains_workbench():
|
||||
self.workbench_counter += 1
|
||||
if addon.contains_macro():
|
||||
self.macro_counter += 1
|
||||
if addon.contains_preference_pack():
|
||||
self.prefpack_counter += 1
|
||||
|
||||
@@ -27,11 +27,7 @@ import unittest
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
have_git = True
|
||||
try:
|
||||
import git
|
||||
except ImportError:
|
||||
have_git = False
|
||||
from addonmanager_git import initialize_git
|
||||
|
||||
import FreeCAD
|
||||
|
||||
@@ -44,24 +40,31 @@ from addonmanager_workers_startup import (
|
||||
LoadPackagesFromCacheWorker,
|
||||
LoadMacrosFromCacheWorker,
|
||||
CheckSingleUpdateWorker,
|
||||
)
|
||||
)
|
||||
|
||||
from addonmanager_workers_installation import (
|
||||
InstallWorkbenchWorker,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestWorkersStartup(unittest.TestCase):
|
||||
|
||||
MODULE = "test_workers_startup" # file name without extension
|
||||
|
||||
def setUp(self):
|
||||
""" Set up the test """
|
||||
self.test_dir = os.path.join(FreeCAD.getHomePath(), "Mod", "AddonManager", "AddonManagerTest", "data")
|
||||
|
||||
"""Set up the test"""
|
||||
self.test_dir = os.path.join(
|
||||
FreeCAD.getHomePath(), "Mod", "AddonManager", "AddonManagerTest", "data"
|
||||
)
|
||||
|
||||
self.saved_mod_directory = Addon.mod_directory
|
||||
self.saved_cache_directory = Addon.cache_directory
|
||||
Addon.mod_directory = os.path.join(tempfile.gettempdir(),"FreeCADTesting","Mod")
|
||||
Addon.cache_directory = os.path.join(tempfile.gettempdir(),"FreeCADTesting","Cache")
|
||||
Addon.mod_directory = os.path.join(
|
||||
tempfile.gettempdir(), "FreeCADTesting", "Mod"
|
||||
)
|
||||
Addon.cache_directory = os.path.join(
|
||||
tempfile.gettempdir(), "FreeCADTesting", "Cache"
|
||||
)
|
||||
|
||||
os.makedirs(Addon.mod_directory, mode=0o777, exist_ok=True)
|
||||
os.makedirs(Addon.cache_directory, mode=0o777, exist_ok=True)
|
||||
@@ -78,23 +81,29 @@ class TestWorkersStartup(unittest.TestCase):
|
||||
self.prefpack_counter = 0
|
||||
self.addon_from_cache_counter = 0
|
||||
self.macro_from_cache_counter = 0
|
||||
|
||||
# Populated when the addon list is created in the first test
|
||||
|
||||
self.package_cache = {}
|
||||
self.macro_cache = []
|
||||
|
||||
self.package_cache_filename = os.path.join(Addon.cache_directory,"packages.json")
|
||||
self.macro_cache_filename = os.path.join(Addon.cache_directory,"macros.json")
|
||||
self.package_cache_filename = os.path.join(
|
||||
Addon.cache_directory, "packages.json"
|
||||
)
|
||||
self.macro_cache_filename = os.path.join(Addon.cache_directory, "macros.json")
|
||||
|
||||
# Store the user's preference for whether git is enabled or disabled
|
||||
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
|
||||
self.saved_git_disabled_status = pref.GetBool("disableGit", False)
|
||||
|
||||
def tearDown(self):
|
||||
""" Tear down the test """
|
||||
"""Tear down the test"""
|
||||
Addon.mod_directory = self.saved_mod_directory
|
||||
Addon.cache_directory = self.saved_cache_directory
|
||||
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
|
||||
pref.SetBool("disableGit", self.saved_git_disabled_status)
|
||||
|
||||
def test_create_addon_list_worker(self):
|
||||
""" Test whether any addons are added: runs the full query, so this potentially is a SLOW
|
||||
test. Note that this test must be run before any of the other tests, so that the cache gets
|
||||
created. """
|
||||
"""Test whether any addons are added: runs the full query, so this potentially is a SLOW
|
||||
test."""
|
||||
worker = CreateAddonListWorker()
|
||||
worker.addon_repo.connect(self._addon_added)
|
||||
worker.start()
|
||||
@@ -102,21 +111,94 @@ class TestWorkersStartup(unittest.TestCase):
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
|
||||
self.assertGreater(self.macro_counter,0, "No macros returned")
|
||||
self.assertGreater(self.workbench_counter,0, "No workbenches returned")
|
||||
self.assertGreater(self.prefpack_counter,0, "No preference packs returned")
|
||||
self.assertGreater(self.macro_counter, 0, "No macros returned")
|
||||
self.assertGreater(self.workbench_counter, 0, "No workbenches returned")
|
||||
|
||||
# Make sure there are no duplicates:
|
||||
addon_name_set = set()
|
||||
for addon in self.addon_list:
|
||||
addon_name_set.add(addon.name)
|
||||
self.assertEqual(
|
||||
len(addon_name_set), len(self.addon_list), "Duplicate names are not allowed"
|
||||
)
|
||||
|
||||
# Write the cache data
|
||||
if hasattr(self, "package_cache"):
|
||||
with open(self.package_cache_filename,"w",encoding="utf-8") as f:
|
||||
with open(self.package_cache_filename, "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(self.package_cache, indent=" "))
|
||||
if hasattr(self, "macro_cache"):
|
||||
with open(self.macro_cache_filename,"w",encoding="utf-8") as f:
|
||||
with open(self.macro_cache_filename, "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(self.macro_cache, indent=" "))
|
||||
|
||||
def _addon_added(self, addon:Addon):
|
||||
""" Callback for adding an Addon: tracks the list, and counts the various types """
|
||||
print (f"Addon Test: {addon.name}")
|
||||
original_macro_counter = self.macro_counter
|
||||
original_workbench_counter = self.workbench_counter
|
||||
original_addon_list = self.addon_list.copy()
|
||||
self.macro_counter = 0
|
||||
self.workbench_counter = 0
|
||||
self.addon_list.clear()
|
||||
|
||||
# Now try loading the same data from the cache we just created
|
||||
worker = LoadPackagesFromCacheWorker(self.package_cache_filename)
|
||||
worker.override_metadata_cache_path(
|
||||
os.path.join(Addon.cache_directory, "PackageMetadata")
|
||||
)
|
||||
worker.addon_repo.connect(self._addon_added)
|
||||
|
||||
worker.start()
|
||||
while worker.isRunning():
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
|
||||
worker = LoadMacrosFromCacheWorker(self.macro_cache_filename)
|
||||
worker.add_macro_signal.connect(self._addon_added)
|
||||
|
||||
worker.start()
|
||||
while worker.isRunning():
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
|
||||
# Make sure that every addon in the original list is also in the new list
|
||||
fail_counter = 0
|
||||
for original_addon in original_addon_list:
|
||||
found = False
|
||||
for addon in self.addon_list:
|
||||
if addon.name == original_addon.name:
|
||||
found = True
|
||||
break
|
||||
if not found:
|
||||
print(f"Failed to load {addon.name} from cache")
|
||||
fail_counter += 1
|
||||
self.assertEqual(fail_counter, 0)
|
||||
|
||||
# Make sure there are no duplicates:
|
||||
addon_name_set.clear()
|
||||
for addon in self.addon_list:
|
||||
addon_name_set.add(addon.name)
|
||||
|
||||
self.assertEqual(len(addon_name_set), len(self.addon_list))
|
||||
self.assertEqual(len(original_addon_list), len(self.addon_list))
|
||||
|
||||
self.assertEqual(
|
||||
original_macro_counter,
|
||||
self.macro_counter,
|
||||
"Cache loaded a different number of macros",
|
||||
)
|
||||
# We can't check workbench and preference pack counting at this point, because that relies
|
||||
# on the package.xml metadata file, which this test does not download.
|
||||
|
||||
def test_create_addon_list_git_disabled(self):
|
||||
"""If the user has git enabled, also test the addon manager with git disabled"""
|
||||
if self.saved_git_disabled_status:
|
||||
self.skipTest("Git is disabled, this test is redundant")
|
||||
|
||||
pref = FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons")
|
||||
pref.SetBool("disableGit", True)
|
||||
|
||||
self.test_create_addon_list_worker()
|
||||
|
||||
def _addon_added(self, addon: Addon):
|
||||
"""Callback for adding an Addon: tracks the list, and counts the various types"""
|
||||
print(f"Addon added: {addon.name}")
|
||||
self.addon_list.append(addon)
|
||||
if addon.contains_workbench():
|
||||
self.workbench_counter += 1
|
||||
@@ -126,94 +208,7 @@ class TestWorkersStartup(unittest.TestCase):
|
||||
self.prefpack_counter += 1
|
||||
|
||||
# Also record the information for cache purposes
|
||||
self.package_cache[addon.name] = addon.to_cache()
|
||||
|
||||
if addon.macro is not None:
|
||||
if addon.macro is None:
|
||||
self.package_cache[addon.name] = addon.to_cache()
|
||||
else:
|
||||
self.macro_cache.append(addon.macro.to_cache())
|
||||
|
||||
def test_load_packages_from_cache_worker(self):
|
||||
""" Test loading packages from the cache """
|
||||
worker = LoadPackagesFromCacheWorker(self.package_cache_filename)
|
||||
worker.override_metadata_cache_path(os.path.join(Addon.cache_directory,"PackageMetadata"))
|
||||
worker.addon_repo.connect(self._addon_added_from_cache)
|
||||
self.addon_from_cache_counter = 0
|
||||
|
||||
worker.start()
|
||||
while worker.isRunning():
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
|
||||
self.assertGreater(self.addon_from_cache_counter,0, "No addons in the cache")
|
||||
|
||||
def _addon_added_from_cache(self, addon:Addon):
|
||||
""" Callback when addon added from cache """
|
||||
print (f"Addon Cache Test: {addon.name}")
|
||||
self.addon_from_cache_counter += 1
|
||||
|
||||
def test_load_macros_from_cache_worker(self):
|
||||
""" Test loading macros from the cache """
|
||||
worker = LoadMacrosFromCacheWorker(self.macro_cache_filename)
|
||||
worker.add_macro_signal.connect(self._macro_added_from_cache)
|
||||
self.macro_from_cache_counter = 0
|
||||
|
||||
worker.start()
|
||||
while worker.isRunning():
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
|
||||
self.assertGreater(self.macro_from_cache_counter,0, "No macros in the cache")
|
||||
|
||||
def _macro_added_from_cache(self, addon:Addon):
|
||||
""" Callback for adding macros from the cache """
|
||||
print (f"Macro Cache Test: {addon.name}")
|
||||
self.macro_from_cache_counter += 1
|
||||
|
||||
def test_update_checker(self):
|
||||
""" Test the code that checks a single addon for available updates. """
|
||||
|
||||
if not have_git:
|
||||
return
|
||||
|
||||
# Populate the test's Addon List:
|
||||
self.test_create_addon_list_worker()
|
||||
|
||||
# First, install a specific Addon of each kind into a temp location
|
||||
location = os.path.join(tempfile.gettempdir(),"FreeCADTesting")
|
||||
|
||||
self._test_workbench_update_checker(location)
|
||||
|
||||
|
||||
|
||||
# Preference Pack
|
||||
# Macro
|
||||
|
||||
# Arrange for those addons to be out-of-date
|
||||
|
||||
# Check for updates
|
||||
|
||||
def _test_workbench_update_checker(self, location):
|
||||
|
||||
# Workbench: use the FreeCAD-Help workbench for testing purposes
|
||||
help_addon = None
|
||||
for addon in self.addon_list:
|
||||
if addon.name == "Help":
|
||||
help_addon = addon
|
||||
break
|
||||
if not help_addon:
|
||||
print("Unable to locate the FreeCAD-Help addon to test with")
|
||||
return
|
||||
|
||||
addon_location = os.path.join(location, help_addon.name)
|
||||
worker = InstallWorkbenchWorker(addon, addon_location)
|
||||
worker.run() # Synchronous call, blocks until complete
|
||||
gitrepo = git.Git(addon_location)
|
||||
gitrepo.reset("--hard", "HEAD^")
|
||||
print (addon_location)
|
||||
|
||||
# At this point the addon should be "out of date", checked out to one revision behind
|
||||
# the most recent.
|
||||
|
||||
worker = CheckSingleUpdateWorker(help_addon)
|
||||
worker.do_work() # Synchronous call
|
||||
|
||||
self.assertEqual(help_addon.status(), Addon.Status.UPDATE_AVAILABLE)
|
||||
@@ -30,12 +30,15 @@ from PySide2 import QtCore
|
||||
|
||||
import NetworkManager
|
||||
|
||||
|
||||
class TestWorkersUtility(unittest.TestCase):
|
||||
|
||||
MODULE = "test_workers_utility" # file name without extension
|
||||
|
||||
def setUp(self):
|
||||
self.test_dir = os.path.join(FreeCAD.getHomePath(), "Mod", "AddonManager", "AddonManagerTest", "data")
|
||||
self.test_dir = os.path.join(
|
||||
FreeCAD.getHomePath(), "Mod", "AddonManager", "AddonManagerTest", "data"
|
||||
)
|
||||
self.last_result = None
|
||||
|
||||
url = "https://api.github.com/zen"
|
||||
@@ -45,7 +48,7 @@ class TestWorkersUtility(unittest.TestCase):
|
||||
self.skipTest("No active internet connection detected")
|
||||
|
||||
def test_connection_checker_basic(self):
|
||||
""" Tests the connection checking worker's basic operation: does not exit until worker thread completes """
|
||||
"""Tests the connection checking worker's basic operation: does not exit until worker thread completes"""
|
||||
worker = ConnectionChecker()
|
||||
worker.success.connect(self.connection_succeeded)
|
||||
worker.failure.connect(self.connection_failed)
|
||||
@@ -54,8 +57,8 @@ class TestWorkersUtility(unittest.TestCase):
|
||||
while worker.isRunning():
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
self.assertEqual(self.last_result,"SUCCESS")
|
||||
|
||||
self.assertEqual(self.last_result, "SUCCESS")
|
||||
|
||||
def test_connection_checker_thread_interrupt(self):
|
||||
worker = ConnectionChecker()
|
||||
worker.success.connect(self.connection_succeeded)
|
||||
@@ -66,8 +69,10 @@ class TestWorkersUtility(unittest.TestCase):
|
||||
while worker.isRunning():
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
|
||||
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
|
||||
self.assertIsNone(self.last_result, "Requesting interruption of thread failed to interrupt")
|
||||
|
||||
self.assertIsNone(
|
||||
self.last_result, "Requesting interruption of thread failed to interrupt"
|
||||
)
|
||||
|
||||
def connection_succeeded(self):
|
||||
self.last_result = "SUCCESS"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user