Replace asynctest with unittest.IsolatedAsyncioTestCase

setup.py

@@ -33,7 +33,7 @@
     packages         = find_packages(),
     package_data     = {'stig': ['settings/default.theme']},
 
-    python_requires  = '>=3.6',
+    python_requires  = '>=3.10',
     install_requires = [
         'urwid>=2.6.12',
         'urwidtrees==1.0.3',
@@ -49,15 +49,18 @@
         'proxy': ['aiohttp-socks'],
     },
     tests_require = [
-        'pytest>=5,<6',
-        'asynctest>=0.11',
+        'pytest',
     ],
 
     entry_points = { 'console_scripts': [ 'stig = stig:run' ] },
 
     classifiers = [
         'License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)',
-        'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3.10',
+        'Programming Language :: Python :: 3.11',
+        'Programming Language :: Python :: 3.12',
+        'Programming Language :: Python :: 3.13',
+        'Programming Language :: Python :: 3.14',
         'Environment :: Console',
         'Operating System :: Unix',
         'Development Status :: 3 - Alpha',

tests/client_test/aiotransmission_test/api_freespace_test.py

@@ -1,14 +1,14 @@
 from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, call
 
-import asynctest
-from asynctest import CoroutineMock, Mock, call
+from testutils import ClockedTestCase
 
 from stig.client.aiotransmission.api_freespace import FreeSpaceAPI
 
 
-class TestFreeSpaceAPI(asynctest.ClockedTestCase):
-    async def setUp(self):
-        self.rpc = SimpleNamespace(free_space=CoroutineMock())
+class TestFreeSpaceAPI(ClockedTestCase):
+    async def asyncSetUp(self):
+        self.rpc = SimpleNamespace(free_space=AsyncMock())
         self.freespace = FreeSpaceAPI((), self.rpc, Mock())
 
     async def test_expected_path_matches(self):

tests/client_test/aiotransmission_test/api_settings_test.py

@@ -1,8 +1,8 @@
+import unittest
 from copy import deepcopy
 from types import SimpleNamespace
 from unittest.mock import Mock, call
 
-import asynctest
 import resources_aiotransmission as rsrc
 
 from stig.client import ClientError
@@ -23,8 +23,8 @@ async def session_set(self, settings):
         self.fake_settings.update(settings)
 
 
-class TestSettingsAPI(asynctest.TestCase):
-    async def setUp(self):
+class TestSettingsAPI(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
         self.rpc = FakeTransmissionRPC()
         srvapi = SimpleNamespace(rpc=self.rpc)
         self.api = SettingsAPI(srvapi)

tests/client_test/aiotransmission_test/api_status_test.py

@@ -1,6 +1,6 @@
+import unittest
 from types import SimpleNamespace
 
-import asynctest
 import resources_aiotransmission as rsrc
 
 from stig.client.aiotransmission import api_status
@@ -45,8 +45,8 @@ async def torrents(self, *args, **kwargs):
         return self.fake_tlist
 
 
-class TestStatusAPI(asynctest.TestCase):
-    async def setUp(self):
+class TestStatusAPI(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
         self.rpc = FakeTransmissionRPC()
         self.torrent = FakeTorrentAPI()
         srvapi = SimpleNamespace(rpc=self.rpc,

tests/client_test/aiotransmission_test/api_torrent_test.py

@@ -1,6 +1,7 @@
 import os.path
+import unittest
+from unittest.mock import patch
 
-import asynctest
 import resources_aiotransmission as rsrc
 
 from stig.client import MAX_TORRENT_FILE_SIZE
@@ -13,16 +14,16 @@
 assert not os.path.exists(rsrc.TORRENTFILE_NOEXIST)
 
 
-class TorrentAPITestCase(asynctest.TestCase):
-    async def setUp(self):
+class TorrentAPITestCase(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
         self.daemon = rsrc.FakeTransmissionDaemon()
         await self.daemon.start()
         self.rpc = TransmissionRPC(self.daemon.host, self.daemon.port)
         self.api = TorrentAPI(self.rpc)
         await self.rpc.connect()
         assert self.rpc.connected is True
 
-    async def tearDown(self):
+    async def asyncTearDown(self):
         await self.rpc.disconnect()
         await self.daemon.stop()
 
@@ -62,8 +63,8 @@ async def test_add_torrent_by_nonexisting_file(self):
         self.assertEqual(response.errors,
                          ('Torrent file is corrupt or doesn\'t exist: %r' % rsrc.TORRENTFILE_NOEXIST,))
 
-    @asynctest.patch('os.path.exists', return_value=True)
-    @asynctest.patch('os.path.getsize', return_value=int(MAX_TORRENT_FILE_SIZE) + 1)
+    @patch('os.path.exists', return_value=True)
+    @patch('os.path.getsize', return_value=int(MAX_TORRENT_FILE_SIZE) + 1)
     async def test_add_torrent_by_giant_file(self, _, __):
         response = await self.api.add('some.torrent')
         self.assertEqual(response.success, False)
@@ -162,8 +163,8 @@ async def test_get_torrents_by_filter(self):
 
 
 class TestManipulatingTorrents(TorrentAPITestCase):
-    async def setUp(self):
-        await super().setUp()
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
         self.mock_method_args = None
         self.mock_method_kwargs = None
         self.daemon.response = rsrc.response_torrents(

tests/client_test/aiotransmission_test/rpc_test.py

@@ -1,15 +1,15 @@
 import asyncio
 
-import asynctest
 import resources_aiotransmission as rsrc
 from aiohttp import web
+from testutils import ClockedTestCase
 
 from stig.client import AuthError, ConnectionError, RPCError, TimeoutError
 from stig.client.aiotransmission.rpc import TransmissionRPC
 
 
-class TestTransmissionRPC(asynctest.ClockedTestCase):
-    async def setUp(self):
+class TestTransmissionRPC(ClockedTestCase):
+    async def asyncSetUp(self):
         self.daemon = rsrc.FakeTransmissionDaemon()
         self.daemon.response = rsrc.SESSION_GET_RESPONSE   # Default response
         await self.daemon.start()
@@ -22,7 +22,7 @@ async def setUp(self):
         self.client.on('disconnected', self.cb_disconnected)
         self.client.on('error', self.cb_error)
 
-    async def tearDown(self):
+    async def asyncTearDown(self):
         await self.client.disconnect()
         await self.daemon.stop()
 

tests/client_test/base_test.py

@@ -1,16 +1,16 @@
 from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, call
 
-import asynctest
-from asynctest import CoroutineMock, Mock, call
+from testutils import ClockedTestCase
 
 from stig.client.base import FreeSpaceAPIBase
 from stig.client.errors import ClientError
 from stig.utils.usertypes import Int
 
 
-class TestFreeSpaceAPI(asynctest.ClockedTestCase):
-    async def setUp(self):
-        self.get_free_space = CoroutineMock()
+class TestFreeSpaceAPI(ClockedTestCase):
+    async def asyncSetUp(self):
+        self.get_free_space = AsyncMock()
 
         class FreeSpaceAPI(FreeSpaceAPIBase):
             get_free_space = self.get_free_space

tests/client_test/poll_test.py

@@ -1,12 +1,12 @@
 import asyncio
 
-import asynctest
+from testutils import ClockedTestCase
 
 from stig.client.errors import AuthError, ConnectionError
 from stig.client.poll import RequestPoller
 
 
-class TestRequestPoller(asynctest.ClockedTestCase):
+class TestRequestPoller(ClockedTestCase):
     def setUp(self):
         self.mock_request_args = None
         self.mock_request_kwargs = None

tests/client_test/trequestpool_test.py

@@ -1,7 +1,7 @@
 import asyncio
 from types import SimpleNamespace
 
-import asynctest
+from testutils import ClockedTestCase
 
 from stig.client.aiotransmission.torrent import Torrent
 from stig.client.filters.torrent import TorrentFilter
@@ -67,8 +67,8 @@ def __add__(self, other):
         return s
 
 
-class TestTorrentRequestPool(asynctest.ClockedTestCase):
-    async def setUp(self):
+class TestTorrentRequestPool(ClockedTestCase):
+    async def asyncSetUp(self):
         self.api = FakeTorrentAPI()
         srvapi = SimpleNamespace(torrent=self.api)
         self.rp = TorrentRequestPool(srvapi)

tests/commands_test/cmdbase_test.py

@@ -1,12 +1,12 @@
 import asyncio
+import unittest
 
-import asynctest
 from resources_cmd import Callback, make_cmdcls
 
 from stig.commands import CmdArgError, _CommandBase
 
 
-class TestCommandBase(asynctest.TestCase):
+class TestCommandBase(unittest.IsolatedAsyncioTestCase):
     def setUp(self):
         def run_sync(self_, A, B):
             assert isinstance(self_, _CommandBase)

tests/commands_test/cmdmanager_test.py

@@ -1,13 +1,14 @@
 import asyncio
+import unittest
+from unittest.mock import MagicMock, call
 
-import asynctest
-from asynctest.mock import call
 from resources_cmd import Callback, make_cmdcls
+from testutils import ClockedTestCase
 
 from stig.commands import CmdError, CommandManager, _CommandBase
 
 
-class TestCommandManagerManagement(asynctest.TestCase):
+class TestCommandManagerManagement(unittest.IsolatedAsyncioTestCase):
     def setUp(self):
         self.cmdmgr = CommandManager()
         self.cmd_foo = make_cmdcls(name='foo', provides=('cli',))
@@ -87,7 +88,7 @@ def test_get_cmdcls(self):
         self.assertIn('interface', str(cm.exception).lower())
 
 
-class TestCommandManagerCallsBase(asynctest.ClockedTestCase):
+class TestCommandManagerCallsBase(ClockedTestCase):
     def setUp(self):
         self.info_handler = Callback()
         self.error_handler = Callback()
@@ -277,7 +278,7 @@ class TestCommandManagerCalls_RunIgnoredCalls(TestCommandManagerCallsBase):
     def setUp(self):
         super().setUp()
 
-        self.mock_gui = asynctest.mock.MagicMock()
+        self.mock_gui = MagicMock()
         argspecs = ({'names': ('A',), 'type': int, 'description': 'First number'},
                     {'names': ('B',), 'type': int, 'description': 'Second number'})
 

tests/commands_test/config_cmds_test.py

@@ -1,7 +1,7 @@
 import os
 import sys
+from unittest.mock import AsyncMock, MagicMock, call, mock_open, patch
 
-from asynctest.mock import CoroutineMock, MagicMock, call, mock_open, patch
 from resources_cmd import CommandTestCase
 
 from stig import objects
@@ -486,8 +486,8 @@ def setUp(self):
                    srvapi=self.srvapi)
 
     async def test_call_syntaxes(self):
-        _set_limits = CoroutineMock()
-        _show_limits = CoroutineMock()
+        _set_limits = AsyncMock()
+        _show_limits = AsyncMock()
         self.patch(RateLimitCmd, _set_limits=_set_limits, _show_limits=_show_limits)
 
         await self.execute(RateLimitCmd, 'up', '1Mb', 'limit-rate-up<1k')

tests/commands_test/resources_cmd.py

@@ -2,11 +2,9 @@
 import io
 import re
 import sys
+import unittest
 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
-
-import asynctest
-from asynctest import CoroutineMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from stig.client.base import TorrentBase
 from stig.client.utils import Response
@@ -139,21 +137,21 @@ class MockSettings(dict):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.reset = MagicMock()
-        self.set = CoroutineMock(side_effect=lambda name, value: self.__setitem__(name, value))
-        self.update = CoroutineMock()
+        self.set = AsyncMock(side_effect=lambda name, value: self.__setitem__(name, value))
+        self.update = AsyncMock()
         self.is_local = lambda name: not name.startswith('srv.')
         self.is_remote = lambda name: name.startswith('srv.')
 
 
-class CommandTestCase(asynctest.TestCase):
+class CommandTestCase(unittest.IsolatedAsyncioTestCase):
     def setUp(self):
         self.srvapi = SimpleNamespace(torrent=MockAPI(),
                                       rpc=MockAPI(),
                                       settings=MockAPI())
         self.cfg = MockSettings()
         self.helpmgr = MockHelpManager()
         self.cmdmgr = MagicMock()
-        self.cmdmgr.run_async = CoroutineMock()
+        self.cmdmgr.run_async = AsyncMock()
 
         self.stdout = sys.stdout = io.StringIO()
         def reset_stdout(): sys.stdout = sys.__stdout__