diff --git a/MANIFEST.in b/MANIFEST.in index 051ea549..ec2fbf0f 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -46,13 +46,14 @@ include pyftpdlib/prefork.py include pyftpdlib/servers.py include pyftpdlib/test/README include pyftpdlib/test/__init__.py +include pyftpdlib/test/conftest.py include pyftpdlib/test/keycert.pem include pyftpdlib/test/test_authorizers.py +include pyftpdlib/test/test_cli.py include pyftpdlib/test/test_filesystems.py include pyftpdlib/test/test_functional.py include pyftpdlib/test/test_functional_ssl.py include pyftpdlib/test/test_ioloop.py -include pyftpdlib/test/test_misc.py include pyftpdlib/test/test_servers.py include pyproject.toml include scripts/ftpbench diff --git a/Makefile b/Makefile index 87030d42..936b0e0a 100644 --- a/Makefile +++ b/Makefile @@ -25,6 +25,7 @@ ifeq ($(PYVER), 2) ipaddress \ mock \ psutil \ + pytest \ pyopenssl \ pysendfile \ setuptools @@ -138,9 +139,9 @@ test-ioloop: ## Run IOLoop tests. ${MAKE} install $(TEST_PREFIX) $(PYTHON) -m pytest $(PYTEST_ARGS) $(ARGS) pyftpdlib/test/test_ioloop.py -test-misc: ## Run miscellaneous tests. +test-cli: ## Run miscellaneous tests. ${MAKE} install - $(TEST_PREFIX) $(PYTHON) -m pytest $(PYTEST_ARGS) $(ARGS) pyftpdlib/test/test_misc.py + $(TEST_PREFIX) $(PYTHON) -m pytest $(PYTEST_ARGS) $(ARGS) pyftpdlib/test/test_cli.py test-lastfailed: ## Run previously failed tests ${MAKE} install diff --git a/pyftpdlib/test/__init__.py b/pyftpdlib/test/__init__.py index fea76fc1..4010000f 100644 --- a/pyftpdlib/test/__init__.py +++ b/pyftpdlib/test/__init__.py @@ -88,23 +88,6 @@ class PyftpdlibTestCase(unittest.TestCase): """All test classes inherit from this one.""" - def setUp(self): - self._test_ctx = {} - self._test_ctx["threads"] = set(threading.enumerate()) - - def tearDown(self): - if not hasattr(self, "_test_ctx"): - raise AssertionError( - "super().setUp() was not called for this test class" - ) - threads = set(threading.enumerate()) - if len(threads) > len(self._test_ctx["threads"]): - extra = threads - self._test_ctx["threads"] - raise AssertionError( - "%s orphaned thread(s) were left behind: %r" - % (len(extra), extra) - ) - def __str__(self): # Print a full path representation of the single unit tests # being run. diff --git a/pyftpdlib/test/conftest.py b/pyftpdlib/test/conftest.py new file mode 100644 index 00000000..332f3145 --- /dev/null +++ b/pyftpdlib/test/conftest.py @@ -0,0 +1,67 @@ +# Copyright (C) 2007 Giampaolo Rodola' . +# Use of this source code is governed by MIT license that can be +# found in the LICENSE file. + +""" +pytest config file (file name has special meaning), executed before +running tests. + +In here we tell pytest to execute setup/teardown functions before/after +each unit-test. We do so to make sure no orphaned resources are left +behind. + +In unittest terms, this is equivalent to implicitly defining setUp(), +tearDown(), setUpClass(), tearDownClass() methods for each test class. +""" + +import threading +import warnings + +import pytest + + +def collect_resources(): + # Note: files and sockets are already collected by pytest, so no + # need to use psutil for it. + res = {} + res["threads"] = set(threading.enumerate()) + return res + + +def setup(origin): + ctx = collect_resources() + ctx["_origin"] = origin + return ctx + + +def warn(msg): + warnings.warn(msg, ResourceWarning, stacklevel=3) + + +def assert_closed_resources(setup_ctx, request): + if request.session.testsfailed: + return # no need to warn if test already failed + + before = setup_ctx.copy() + after = collect_resources() + for key, value in before.items(): + if key.startswith("_"): + continue + msg = "%r left some unclosed %r resources behind: " % ( + setup_ctx['_origin'], + key, + ) + extra = after[key] - before[key] + if extra: + if isinstance(value, set): + msg += repr(extra) + warn(msg) + elif extra > 0: # unused, here just in case we extend it later + msg += "before=%r, after=%r" % (before[key], after[key]) + warn(msg) + + +@pytest.fixture(autouse=True, scope="function") +def for_each_test_method(request): + ctx = setup(request.node.nodeid) + request.addfinalizer(lambda: assert_closed_resources(ctx, request)) diff --git a/pyftpdlib/test/test_authorizers.py b/pyftpdlib/test/test_authorizers.py index d50bc9d2..23506c06 100644 --- a/pyftpdlib/test/test_authorizers.py +++ b/pyftpdlib/test/test_authorizers.py @@ -6,7 +6,6 @@ import random import string import sys -import unittest import warnings import pytest @@ -484,9 +483,9 @@ def test_override_user_errors(self): # ===================================================================== -@unittest.skipUnless(POSIX, "UNIX only") -@unittest.skipUnless( - UnixAuthorizer is not None, "UnixAuthorizer class not available" +@pytest.mark.skipif(not POSIX, reason="UNIX only") +@pytest.mark.skipif( + UnixAuthorizer is None, reason="UnixAuthorizer class not available" ) class TestUnixAuthorizer(_SharedAuthorizerTests, PyftpdlibTestCase): """Unix authorizer specific tests.""" @@ -614,7 +613,7 @@ def test_not_root(self): # ===================================================================== -@unittest.skipUnless(WINDOWS, "Windows only") +@pytest.mark.skipif(not WINDOWS, reason="Windows only") class TestWindowsAuthorizer(_SharedAuthorizerTests, PyftpdlibTestCase): """Windows authorizer specific tests.""" diff --git a/pyftpdlib/test/test_misc.py b/pyftpdlib/test/test_cli.py similarity index 93% rename from pyftpdlib/test/test_misc.py rename to pyftpdlib/test/test_cli.py index 6aed2c2c..54f0cde7 100644 --- a/pyftpdlib/test/test_misc.py +++ b/pyftpdlib/test/test_cli.py @@ -21,6 +21,7 @@ from pyftpdlib.authorizers import DummyAuthorizer from pyftpdlib.servers import FTPServer from pyftpdlib.test import PyftpdlibTestCase +from pyftpdlib.test import mock class TestCommandLineParser(PyftpdlibTestCase): @@ -36,7 +37,6 @@ class DummyFTPServer(FTPServer): def serve_forever(self, *args, **kwargs): self.close_all() - return if PY3: import io @@ -45,10 +45,12 @@ def serve_forever(self, *args, **kwargs): else: self.devnull = BytesIO() self.original_ftpserver_class = FTPServer + self.clog = mock.patch("pyftpdlib.__main__.config_logging") + self.clog.start() pyftpdlib.__main__.FTPServer = DummyFTPServer def tearDown(self): - self.devnull.close() + self.clog.stop() pyftpdlib.servers.FTPServer = self.original_ftpserver_class super().tearDown() @@ -77,6 +79,7 @@ def test_write_opt(self): main(["-w", "-p", "0"]) with warnings.catch_warnings(): + warnings.filterwarnings("ignore") ftpd = main(["-w", "-p", "0"]) perms = ftpd.handler.authorizer.get_perms("anonymous") assert ( @@ -85,8 +88,9 @@ def test_write_opt(self): ) # unexpected argument - with pytest.raises(SystemExit): - main(["-w", "foo", "-p", "0"]) + with warnings.catch_warnings(): + with pytest.raises(SystemExit): + main(["-w", "foo", "-p", "0"]) def test_directory_opt(self): dirname = self.get_testfn() diff --git a/pyftpdlib/test/test_filesystems.py b/pyftpdlib/test/test_filesystems.py index f9cc71cd..609d9df4 100644 --- a/pyftpdlib/test/test_filesystems.py +++ b/pyftpdlib/test/test_filesystems.py @@ -4,7 +4,8 @@ import os import tempfile -import unittest + +import pytest from pyftpdlib._compat import getcwdu from pyftpdlib._compat import u @@ -194,7 +195,7 @@ def test_validpath_external_symlink(self): safe_rmpath(testfn) -@unittest.skipUnless(POSIX, "UNIX only") +@pytest.mark.skipif(not POSIX, reason="UNIX only") class TestUnixFilesystem(PyftpdlibTestCase): def test_case(self): diff --git a/pyftpdlib/test/test_functional.py b/pyftpdlib/test/test_functional.py index 81bfa0d5..b24f0e90 100644 --- a/pyftpdlib/test/test_functional.py +++ b/pyftpdlib/test/test_functional.py @@ -22,7 +22,6 @@ except ImportError: from io import BytesIO -import unittest import pytest @@ -281,27 +280,27 @@ def test_type(self): self.client.sendcmd('type i') self.client.sendcmd('type l7') self.client.sendcmd('type l8') - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unsupported type"): self.client.sendcmd('type ?!?') def test_stru(self): self.client.sendcmd('stru f') self.client.sendcmd('stru F') - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unimplemented"): self.client.sendcmd('stru p') - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unimplemented"): self.client.sendcmd('stru r') - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unrecognized"): self.client.sendcmd('stru ?!?') def test_mode(self): self.client.sendcmd('mode s') self.client.sendcmd('mode S') - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unimplemented"): self.client.sendcmd('mode b') - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unimplemented"): self.client.sendcmd('mode c') - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unrecognized"): self.client.sendcmd('mode ?!?') def test_noop(self): @@ -320,7 +319,7 @@ def test_help(self): self.client.sendcmd('help') cmd = random.choice(list(FTPHandler.proto_cmds.keys())) self.client.sendcmd('help %s' % cmd) - with pytest.raises(ftplib.error_perm): + with pytest.raises(ftplib.error_perm, match="Unrecognized"): self.client.sendcmd('help ?!?') def test_site(self): @@ -1043,8 +1042,10 @@ def test_stor_empty_file(self): assert not f.read() -@unittest.skipUnless(POSIX, "POSIX only") -@unittest.skipIf(not PY3 and sendfile is None, "pysendfile not installed") +@pytest.mark.skipif(not POSIX, reason="POSIX only") +@pytest.mark.skipif( + not PY3 and sendfile is None, reason="pysendfile not installed" +) class TestFtpStoreDataNoSendfile(TestFtpStoreData): """Test STOR, STOU, APPE, REST, TYPE not using sendfile().""" @@ -1184,8 +1185,10 @@ def test_retr_empty_file(self): assert self.dummyfile.read() == b"" -@unittest.skipUnless(POSIX, "POSIX only") -@unittest.skipIf(not PY3 and sendfile is None, "pysendfile not installed") +@pytest.mark.skipif(not POSIX, reason="POSIX only") +@pytest.mark.skipif( + not PY3 and sendfile is None, reason="pysendfile not installed" +) class TestFtpRetrieveDataNoSendfile(TestFtpRetrieveData): """Test RETR, REST, TYPE by not using sendfile().""" @@ -1434,8 +1437,10 @@ def test_abor_during_transfer(self): # with a 226 assert self.client.voidresp()[:3] == '226' - @unittest.skipUnless(hasattr(socket, 'MSG_OOB'), "MSG_OOB not available") - @unittest.skipIf(OSX, "does not work on OSX") + @pytest.mark.skipif( + not hasattr(socket, 'MSG_OOB'), reason="MSG_OOB not available" + ) + @pytest.mark.skipif(OSX, reason="does not work on OSX") def test_oob_abor(self): # Send ABOR by following the RFC-959 directives of sending # Telnet IP/Synch sequence as OOB data. @@ -1736,33 +1741,45 @@ def test_max_connections(self): c2 = self.client_class() c3 = self.client_class() try: + # on control connection c1.connect(self.server.host, self.server.port) c2.connect(self.server.host, self.server.port) - with pytest.raises(ftplib.error_temp): + with pytest.raises( + ftplib.error_temp, match="Too many connections" + ): c3.connect( self.server.host, self.server.port, ) + # with passive data channel established c2.quit() c1.login(USER, PASSWD) c1.makepasv() - with pytest.raises(ftplib.error_temp): + with pytest.raises( + ftplib.error_temp, match="Too many connections" + ): c2.connect( self.server.host, self.server.port, ) + # with passive data socket waiting for connection c1.login(USER, PASSWD) c1.sendcmd('pasv') - with pytest.raises(ftplib.error_temp): + with pytest.raises( + ftplib.error_temp, match="Too many connections" + ): + c2.close() c2.connect( self.server.host, self.server.port, ) + # with active data channel established c1.login(USER, PASSWD) with contextlib.closing(c1.makeport()): + c2.close() with pytest.raises(ftplib.error_temp): c2.connect( self.server.host, @@ -1772,7 +1789,10 @@ def test_max_connections(self): for c in (c1, c2, c3): try: c.quit() - except (socket.error, EOFError): # already disconnected + except (socket.error, EOFError, ftplib.Error): + # already disconnected + pass + finally: c.close() @disable_log_warning @@ -2064,9 +2084,10 @@ def test_on_incomplete_file_received(self): break # If a data transfer is in progress server is supposed to send # a 426 reply followed by a 226 reply. - with pytest.raises(ftplib.error_temp): - self.client.getresp() # 426 - assert self.client.getresp()[:3] == "226" + resp = self.client.getmultiline() + assert resp == "426 Transfer aborted via ABOR." + resp = self.client.getmultiline() + assert resp.startswith("226") self.read_file( 'on_connect,on_login:%s,on_incomplete_file_received:%s,' % (USER, self.testfn2) @@ -2221,7 +2242,7 @@ def test_epsv_all(self): ) -@unittest.skipUnless(SUPPORTS_IPV4, "IPv4 not supported") +@pytest.mark.skipif(not SUPPORTS_IPV4, reason="IPv4 not supported") class TestIPv4Environment(_TestNetworkProtocols, PyftpdlibTestCase): """Test PASV, EPSV, PORT and EPRT commands. @@ -2271,7 +2292,7 @@ def test_pasv_v4(self): s.connect((host, port)) -@unittest.skipUnless(SUPPORTS_IPV6, "IPv6 not supported") +@pytest.mark.skipif(not SUPPORTS_IPV6, reason="IPv6 not supported") class TestIPv6Environment(_TestNetworkProtocols, PyftpdlibTestCase): """Test PASV, EPSV, PORT and EPRT commands. @@ -2303,7 +2324,9 @@ def test_eprt_v6(self): assert 'foreign address' in resp -@unittest.skipUnless(SUPPORTS_HYBRID_IPV6, "IPv4/6 dual stack not supported") +@pytest.mark.skipif( + not SUPPORTS_HYBRID_IPV6, reason="IPv4/6 dual stack not supported" +) class TestIPv6MixedEnvironment(PyftpdlibTestCase): """By running the server by specifying "::" as IP address the server is supposed to listen on all interfaces, supporting both @@ -2432,7 +2455,7 @@ def test_port_race_condition(self): s, _ = sock.accept() s.close() - @unittest.skipUnless(POSIX, "POSIX only") + @pytest.mark.skipif(not POSIX, reason="POSIX only") def test_quick_connect(self): # Clients that connected and disconnected quickly could cause # the server to crash, due to a failure to catch errors in the @@ -2526,7 +2549,7 @@ def test_ioloop_fileno(self): # # TODO: disabled as on certain platforms (OSX and Windows) # # produces failures with python3. Will have to get back to # # this and fix it. -# @unittest.skipIf(OSX or WINDOWS, "fails on OSX or Windows") +# @pytest.mark.skipif(OSX or WINDOWS, reason="fails on OSX or Windows") # class TestUnicodePathNames(PyftpdlibTestCase): # """Test FTP commands and responses by using path names with non # ASCII characters. @@ -2709,11 +2732,8 @@ class ThreadedFTPTests(PyftpdlibTestCase): def setUp(self): super().setUp() - self.server = self.server_class() - self.server.start() - self.client = self.client_class(timeout=GLOBAL_TIMEOUT) - self.client.connect(self.server.host, self.server.port) - self.client.login(USER, PASSWD) + self.client = None + self.server = None self.tempfile = self.get_testfn() self.tempdir = self.get_testfn() touch(self.tempfile) @@ -2722,40 +2742,20 @@ def setUp(self): self.dummy_sendfile = BytesIO() def tearDown(self): - close_client(self.client) - self.server.stop() + if self.client: + close_client(self.client) + if self.server: + self.server.stop() + self.server.handler = FTPHandler + self.server.handler.abstracted_fs = AbstractedFS self.dummy_recvfile.close() self.dummy_sendfile.close() super().tearDown() - @retry_on_failure() - def test_unforeseen_mdtm_event(self): - # Emulate a case where the file last modification time is prior - # to year 1900. This most likely will never happen unless - # someone specifically force the last modification time of a - # file in some way. - # To do so we temporarily override os.path.getmtime so that it - # returns a negative value referring to a year prior to 1900. - # It causes time.localtime/gmtime to raise a ValueError exception - # which is supposed to be handled by server. - - # On python 3 it seems that the trick of replacing the original - # method with the lambda doesn't work. - if not PY3: - _getmtime = AbstractedFS.getmtime - try: - AbstractedFS.getmtime = lambda x, y: -9000000000 - with pytest.raises( - ftplib.error_perm, - match="550 Can't determine file's last modification time", - ): - self.client.sendcmd( - 'mdtm ' + self.tempfile, - ) - # make sure client hasn't been disconnected - self.client.sendcmd('noop') - finally: - AbstractedFS.getmtime = _getmtime + def connect_client(self): + self.client = self.client_class(timeout=GLOBAL_TIMEOUT) + self.client.connect(self.server.host, self.server.port) + self.client.login(USER, PASSWD) @retry_on_failure() def test_stou_max_tries(self): @@ -2769,72 +2769,59 @@ def mkstemp(self, *args, **kwargs): errno.EEXIST, "No usable temporary file name found" ) - with self.server.lock: - self.server.handler.abstracted_fs = TestFS - try: - self.client.quit() - self.client.connect(self.server.host, self.server.port) - self.client.login(USER, PASSWD) - with pytest.raises(ftplib.error_temp): - self.client.sendcmd('stou') - finally: - with self.server.lock: - self.server.handler.abstracted_fs = AbstractedFS + self.server = self.server_class() + self.server.handler.abstracted_fs = TestFS + self.server.start() + self.connect_client() + with pytest.raises(ftplib.error_temp): + self.client.sendcmd('stou') @retry_on_failure() def test_idle_timeout(self): # Test control channel timeout. The client which does not send # any command within the time specified in FTPHandler.timeout is # supposed to be kicked off. - with self.server.lock: - self.server.handler.timeout = 0.1 - - try: - self.client.quit() - self.client.connect() - self.client.login(USER, PASSWD) - # fail if no msg is received within 1 second - self.client.sock.settimeout(1) - data = self.client.sock.recv(BUFSIZE) - assert data == b"421 Control connection timed out.\r\n" - # ensure client has been kicked off - with pytest.raises((socket.error, EOFError)): - self.client.sendcmd('noop') - finally: - with self.server.lock: - self.server.handler.timeout = 0.1 + self.server = self.server_class() + self.server.handler.timeout = 0.1 + self.server.start() + self.connect_client() - @unittest.skipUnless( - hasattr(socket, 'TCP_NODELAY'), 'TCP_NODELAY not available' - ) - @retry_on_failure() - def test_tcp_no_delay(self): - s = get_server_handler().socket - assert s.getsockopt(socket.SOL_TCP, socket.TCP_NODELAY) self.client.quit() - with self.server.lock: - self.server.handler.tcp_no_delay = False - self.client.connect(self.server.host, self.server.port) - self.client.sendcmd('noop') - s = get_server_handler().socket - assert not s.getsockopt(socket.SOL_TCP, socket.TCP_NODELAY) + self.client.connect() + self.client.login(USER, PASSWD) + # fail if no msg is received within 1 second + self.client.sock.settimeout(1) + data = self.client.sock.recv(BUFSIZE) + assert data == b"421 Control connection timed out.\r\n" + # ensure client has been kicked off + with pytest.raises((socket.error, EOFError)): + self.client.sendcmd('noop') @retry_on_failure() def test_permit_foreign_address_false(self): + self.server = self.server_class() + self.server.handler.permit_foreign_addresses = False + self.server.start() + self.connect_client() handler = get_server_handler() - with self.server.lock: - handler.permit_foreign_addresses = False - handler.remote_ip = '9.9.9.9' + handler.remote_ip = '9.9.9.9' + # sync + self.client.sendcmd("noop") with pytest.raises(ftplib.error_perm) as cm: - self.client.makeport() + port = self.client.sock.getsockname()[1] + host = self.client.sock.getsockname()[0] + resp = self.client.sendport(host, port) assert 'foreign address' in str(cm.value) @retry_on_failure() def test_permit_foreign_address_true(self): + self.server = self.server_class() + self.server.handler.permit_foreign_addresses = True + self.server.start() + self.connect_client() handler = get_server_handler() - with self.server.lock: - handler.permit_foreign_addresses = True - handler.remote_ip = '9.9.9.9' + handler.remote_ip = '9.9.9.9' + self.client.sendcmd("noop") s = self.client.makeport() s.close() @@ -2851,7 +2838,7 @@ def test_permit_privileged_ports(self): except socket.error: # not registered port; go on try: - sock = socket.socket(self.client.af, socket.SOCK_STREAM) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.addCleanup(sock.close) sock.settimeout(GLOBAL_TIMEOUT) sock.bind((HOST, port)) @@ -2872,27 +2859,42 @@ def test_permit_privileged_ports(self): # no usable privileged port was found sock = None - with self.server.lock: - self.server.handler.permit_privileged_ports = False - with pytest.raises(ftplib.error_perm): - self.client.sendport(HOST, port) + # permit_privileged_ports = False + self.server = self.server_class() + self.server.handler.permit_privileged_ports = False + self.server.start() + self.connect_client() + with pytest.raises(ftplib.error_perm, match="privileged port"): + self.client.sendport(HOST, 1023) + + # permit_privileged_ports = True if sock: + self.tearDown() + + self.server = self.server_class() + self.server.handler.permit_privileged_ports = True + self.server.start() + self.connect_client() port = sock.getsockname()[1] - with self.server.lock: - self.server.handler.permit_privileged_ports = True sock.listen(5) sock.settimeout(GLOBAL_TIMEOUT) self.client.sendport(HOST, port) s, _ = sock.accept() s.close() + sock.close() - @unittest.skipUnless(POSIX, "POSIX only") - @unittest.skipIf(not PY3 and sendfile is None, "pysendfile not installed") + @pytest.mark.skipif(not POSIX, reason="POSIX only") + @pytest.mark.skipif( + not PY3 and sendfile is None, reason="pysendfile not installed" + ) @retry_on_failure() def test_sendfile_fails(self): # Makes sure that if sendfile() fails and no bytes were # transmitted yet the server falls back on using plain # send() + self.server = self.server_class() + self.server.start() + self.connect_client() data = b'abcde12345' * 100000 self.dummy_sendfile.write(data) self.dummy_sendfile.seek(0) diff --git a/pyftpdlib/test/test_functional_ssl.py b/pyftpdlib/test/test_functional_ssl.py index e5c6c57c..fe12585f 100644 --- a/pyftpdlib/test/test_functional_ssl.py +++ b/pyftpdlib/test/test_functional_ssl.py @@ -7,7 +7,6 @@ import os import socket import ssl -import unittest import OpenSSL # requires "pip install pyopenssl" import pytest @@ -95,11 +94,11 @@ class TestFtpFsOperationsTLSMixin(TLSTestMixin, TestFtpFsOperations): class TestFtpStoreDataTLSMixin(TLSTestMixin, TestFtpStoreData): - @unittest.skipIf(1, "fails with SSL") + @pytest.mark.skip(reason="fails with SSL") def test_stou(self): pass - @unittest.skipIf(WINDOWS, "unreliable on Windows + SSL") + @pytest.mark.skipif(WINDOWS, reason="unreliable on Windows + SSL") def test_stor_ascii_2(self): pass @@ -113,7 +112,7 @@ def test_stor_ascii_2(self): class TestFtpRetrieveDataTLSMixin(TLSTestMixin, TestFtpRetrieveData): - @unittest.skipIf(WINDOWS, "may fail on windows") + @pytest.mark.skipif(WINDOWS, reason="may fail on windows") def test_restore_on_retr(self): super().test_restore_on_retr() @@ -127,21 +126,21 @@ class TestFtpListingCmdsTLSMixin(TLSTestMixin, TestFtpListingCmds): # File "/opt/python/2.7.9/lib/python2.7/ssl.py", line 771, in unwrap # s = self._sslobj.shutdown() # error: [Errno 0] Error - @unittest.skipIf(CI_TESTING, "may fail on CI") + @pytest.mark.skipif(CI_TESTING, reason="may fail on CI") def test_nlst(self): super().test_nlst() class TestFtpAbortTLSMixin(TLSTestMixin, TestFtpAbort): - @unittest.skipIf(1, "fails with SSL") + @pytest.mark.skip(reason="fails with SSL") def test_oob_abor(self): pass class TestTimeoutsTLSMixin(TLSTestMixin, TestTimeouts): - @unittest.skipIf(1, "fails with SSL") + @pytest.mark.skip(reason="fails with SSL") def test_data_timeout_not_reached(self): pass diff --git a/pyftpdlib/test/test_ioloop.py b/pyftpdlib/test/test_ioloop.py index 70b7b5af..c533ec52 100644 --- a/pyftpdlib/test/test_ioloop.py +++ b/pyftpdlib/test/test_ioloop.py @@ -7,7 +7,6 @@ import select import socket import time -import unittest import pytest @@ -188,8 +187,9 @@ def test_select_eintr(self): # =================================================================== -@unittest.skipUnless( - hasattr(pyftpdlib.ioloop, 'Poll'), "poll() not available on this platform" +@pytest.mark.skipif( + not hasattr(pyftpdlib.ioloop, 'Poll'), + reason="poll() not available on this platform", ) class PollIOLoopTestCase(PyftpdlibTestCase, BaseIOLoopTestCase): ioloop_class = getattr(pyftpdlib.ioloop, "Poll", None) @@ -264,9 +264,9 @@ def test_enoent_on_modify(self): # =================================================================== -@unittest.skipUnless( - hasattr(pyftpdlib.ioloop, 'Epoll'), - "epoll() not available on this platform (Linux only)", +@pytest.mark.skipif( + not hasattr(pyftpdlib.ioloop, 'Epoll'), + reason="epoll() not available on this platform (Linux only)", ) class EpollIOLoopTestCase(PollIOLoopTestCase): ioloop_class = getattr(pyftpdlib.ioloop, "Epoll", None) @@ -278,9 +278,9 @@ class EpollIOLoopTestCase(PollIOLoopTestCase): # =================================================================== -@unittest.skipUnless( - hasattr(pyftpdlib.ioloop, 'DevPoll'), - "/dev/poll not available on this platform (Solaris only)", +@pytest.mark.skipif( + not hasattr(pyftpdlib.ioloop, 'DevPoll'), + reason="/dev/poll not available on this platform (Solaris only)", ) class DevPollIOLoopTestCase(PyftpdlibTestCase, BaseIOLoopTestCase): ioloop_class = getattr(pyftpdlib.ioloop, "DevPoll", None) @@ -291,9 +291,9 @@ class DevPollIOLoopTestCase(PyftpdlibTestCase, BaseIOLoopTestCase): # =================================================================== -@unittest.skipUnless( - hasattr(pyftpdlib.ioloop, 'Kqueue'), - "/dev/poll not available on this platform (BSD only)", +@pytest.mark.skipif( + not hasattr(pyftpdlib.ioloop, 'Kqueue'), + reason="/dev/poll not available on this platform (BSD only)", ) class KqueueIOLoopTestCase(PyftpdlibTestCase, BaseIOLoopTestCase): ioloop_class = getattr(pyftpdlib.ioloop, "Kqueue", None) diff --git a/pyftpdlib/test/test_servers.py b/pyftpdlib/test/test_servers.py index 76582856..a3e66546 100644 --- a/pyftpdlib/test/test_servers.py +++ b/pyftpdlib/test/test_servers.py @@ -5,7 +5,8 @@ import contextlib import ftplib import socket -import unittest + +import pytest from pyftpdlib import handlers from pyftpdlib import servers @@ -52,7 +53,7 @@ def tearDown(self): self.server.stop() super().tearDown() - @unittest.skipIf(WINDOWS, "POSIX only") + @pytest.mark.skipif(WINDOWS, reason="POSIX only") def test_sock_instead_of_addr(self): # pass a socket object instead of an address tuple to FTPServer # constructor @@ -164,7 +165,7 @@ class MProcFTPTestMixin: else: - @unittest.skipIf(True, "multiprocessing module not installed") + @pytest.mark.skip(reason="multiprocessing module not installed") class MProcFTPTestMixin: pass diff --git a/scripts/internal/winmake.py b/scripts/internal/winmake.py index 01b1f201..6d40904e 100755 --- a/scripts/internal/winmake.py +++ b/scripts/internal/winmake.py @@ -403,9 +403,9 @@ def test_ioloop(): sh("%s pyftpdlib\\test\\test_ioloop.py" % PYTHON) -def test_misc(): +def test_cli(): build() - sh("%s pyftpdlib\\test\\test_misc.py" % PYTHON) + sh("%s pyftpdlib\\test\\test_cli.py" % PYTHON) def test_servers():