From 3b65cc6a7700b44792e35900725688c01086a572 Mon Sep 17 00:00:00 2001 From: Ben Finney Date: Fri, 7 Aug 2015 13:08:08 +1000 Subject: [PATCH] =?utf8?q?Add=20more=20unit=20tests=20for=20=E2=80=98dput.?= =?utf8?q?methods.http.upload=E2=80=99.?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Ben Finney --- test/test_methods.py | 269 ++++++++++++++++++++++++++++++++++----------------- 1 file changed, 179 insertions(+), 90 deletions(-) diff --git a/test/test_methods.py b/test/test_methods.py index e20aba9..47f123b 100644 --- a/test/test_methods.py +++ b/test/test_methods.py @@ -16,6 +16,7 @@ from __future__ import (absolute_import, unicode_literals) import sys import os import os.path +import io import stat import pkgutil import importlib @@ -24,13 +25,18 @@ import tempfile import textwrap import doctest import getpass -import socket -import urllib2 import ftplib -import httplib + +if sys.version_info >= (3, 3): + import urllib.parse as urlparse +elif sys.version_info >= (3, 0): + raise RuntimeError("Python 3 earlier than 3.3 is not supported.") +else: + import urlparse import testtools import testscenarios +import httpretty __package__ = str("test") __import__(__package__) @@ -45,7 +51,6 @@ import dput.methods.http from .helper import ( mock, - StringIO, FakeSystemExit, patch_system_interfaces, EXIT_STATUS_SUCCESS, EXIT_STATUS_FAILURE, @@ -873,11 +878,11 @@ class http_upload_TestCase(upload_TestCase): """ Set up test fixtures. """ super(http_upload_TestCase, self).setUp() - self.set_connection_double() - self.patch_httplib_connection_classes() + httpretty.enable() + self.addCleanup(httpretty.disable) - self.set_authhandler_double() - self.patch_urllib_authhandler_classes() + self.set_response_header_fields() + self.patch_put_requests() patch_getpass_getpass(self) self.fake_password = self.getUniqueString() @@ -901,81 +906,29 @@ class http_upload_TestCase(upload_TestCase): if self.function_to_test is dput.methods.http.upload: self.test_args['protocol'] = self.protocol - def make_http_response(self, status, reason, fields): - """ Make an `HTTPResponse` for this test case. """ - status_line = "{version} {status:d} {reason}".format( - version=self.protocol_version, - status=status, - reason=reason) - header_lines = [ - "{name}: {value}".format(name=field_name, value=field_value) - for (field_name, field_value) - in self.response_header_fields.items()] - - response_content_lines = [status_line] - response_content_lines += header_lines - response_content = "\n".join(response_content_lines) - - fake_socket_file = StringIO(response_content) - fake_socket = mock.MagicMock(spec=socket.SocketType) - fake_socket.makefile.return_value = fake_socket_file - - response = httplib.HTTPResponse(sock=fake_socket) - - return response - - def set_connection_double(self): - """ Set the HTTP connection double. """ - self.connection_double = mock.MagicMock( - name="HTTP connection", spec=httplib.HTTPConnection) + def make_upload_uri(self, file_name): + """ Make the URI for a file for upload. """ + uri = urlparse.urlunsplit([ + self.protocol, self.test_args['fqdn'], + os.path.join(os.path.sep, self.incoming_path, file_name), + None, None]) + return uri + def set_response_header_fields(self): + """ Set the header fields for the HTTP response. """ if not hasattr(self, 'response_header_fields'): self.response_header_fields = {} - response = self.make_http_response( - self.status_code, self.status_reason, - self.response_header_fields) - response.begin() - self.connection_double.getresponse.return_value = response - - def patch_httplib_connection_classes(self): - """ Patch `httplib` connection classes for this test case. """ - for name in ["HTTPConnection", "HTTPSConnection"]: - patcher = mock.patch.object(httplib, name) - class_double = patcher.start() - self.addCleanup(patcher.stop) - - class_double.return_value = self.connection_double - - def set_authhandler_double(self): - """ Set the HTTP auth handler double. """ - double = mock.MagicMock( - name="HTTP auth handler", spec=urllib2.HTTPBasicAuthHandler) - - if not hasattr(self, 'auth_response_header_fields'): - self.auth_response_header_fields = {} - - def fake_http_error_401(req, fp, code, msg, headers): - req.headers = self.auth_response_header_fields - - response = self.make_http_response( - self.auth_response_status_code, - self.auth_response_status_reason, - self.auth_response_header_fields) - response.begin() - - double.http_error_401.side_effect = fake_http_error_401 - - self.authhandler_double = double - - def patch_urllib_authhandler_classes(self): - """ Patch `urllib` auth handler classes for this test case. """ - for name in ["HTTPBasicAuthHandler", "HTTPDigestAuthHandler"]: - patcher = mock.patch.object(urllib2, name) - class_double = patcher.start() - self.addCleanup(patcher.stop) - - class_double.return_value = self.authhandler_double + def patch_put_requests(self): + """ Patch the HTTP PUT requests. """ + self.path_by_request_uri = {} + for path in self.paths_to_upload: + upload_uri = self.make_upload_uri(os.path.basename(path)) + self.path_by_request_uri[upload_uri] = path + response_body = "" + httpretty.register_uri( + httpretty.PUT, upload_uri, + status=self.status_code, body=response_body) class http_upload_SuccessTestCase(http_upload_TestCase): @@ -987,8 +940,8 @@ class http_upload_SuccessTestCase(http_upload_TestCase): 'status_reason': "Okay", }), ('chatter', { - 'status_code': 257, - 'status_reason': "Lorem Ipsum", + 'status_code': 203, + 'status_reason': "Non-Authoritative Information", }), ] @@ -999,6 +952,20 @@ class http_upload_SuccessTestCase(http_upload_TestCase): }), ] + size_scenarios = [ + ('size-empty', { + 'fake_file': io.BytesIO(), + }), + ('size-1k', { + 'fake_file': io.BytesIO( + b"Lorem ipsum, dolor sit amet.___\n" * 32), + }), + ('size-100k', { + 'fake_file': io.BytesIO( + b"Lorem ipsum, dolor sit amet.___\n" * 3200), + }), + ] + incoming_scenarios = list(upload_TestCase.incoming_scenarios) for (scenario_name, scenario) in incoming_scenarios: scenario['expected_url_path_prefix'] = os.path.join( @@ -1007,22 +974,93 @@ class http_upload_SuccessTestCase(http_upload_TestCase): scenarios = testscenarios.multiply_scenarios( upload_TestCase.files_scenarios, + size_scenarios, upload_TestCase.incoming_scenarios, http_upload_TestCase.protocol_scenarios, http_upload_TestCase.login_scenarios, response_scenarios, auth_scenarios) - def test_starts_put_request_with_expected_args(self): - """ Should start PUT request with requested args. """ + def test_emits_debug_message_for_upload(self): + """ Should emit debug message for upload. """ + self.test_args['debug'] = True self.function_to_test(**self.test_args) - expected_calls = [] for path in self.paths_to_upload: - expected_path = os.path.join( - self.expected_url_path_prefix, os.path.basename(path)) - expected_calls.append(mock.call( - "PUT", expected_path, skip_accept_encoding=True)) - self.connection_double.putrequest.assert_has_calls( - expected_calls, any_order=True) + expected_uri = self.make_upload_uri(os.path.basename(path)) + expected_output = textwrap.dedent("""\ + D: HTTP-PUT to URL: {uri} + """).format(uri=expected_uri) + self.expectThat( + sys.stdout.getvalue(), + testtools.matchers.Contains(expected_output)) + + def test_request_has_expected_fields(self): + """ Should send request with expected fields in header. """ + if not self.paths_to_upload: + self.skipTest("No files to upload") + self.function_to_test(**self.test_args) + registry = FileDouble.get_registry_for_testcase(self) + path = self.paths_to_upload[-1] + double = registry[path] + request = httpretty.last_request() + expected_fields = { + 'User-Agent': "dput", + 'Connection': "close", + 'Content-Length': "{size:d}".format( + size=len(double.fake_file.getvalue())), + } + for (name, value) in expected_fields.items(): + self.expectThat( + request.headers.get(name), + testtools.matchers.Equals(value)) + + +class http_upload_ProgressTestCase(http_upload_TestCase): + """ Test cases for `methods.http.upload` function, with progress meter. """ + + files_scenarios = list( + (scenario_name, scenario) for (scenario_name, scenario) + in upload_TestCase.files_scenarios + if scenario['paths_to_upload']) + + response_scenarios = [ + ('okay', { + 'status_code': 200, + 'status_reason': "Okay", + }), + ] + + progress_scenarios = [ + ('progress-type-1', { + 'progress_type': 1, + }), + ('progress-type-2', { + 'progress_type': 2, + }), + ] + + scenarios = testscenarios.multiply_scenarios( + files_scenarios, + progress_scenarios, + upload_TestCase.incoming_scenarios, + http_upload_TestCase.protocol_scenarios, + http_upload_TestCase.login_scenarios, + http_upload_SuccessTestCase.auth_scenarios, + response_scenarios) + + def test_filewithprogress_has_expected_attributes(self): + """ Should have expected attributes on the `FileWithProgress`. """ + expected_attributes_by_path = ( + make_expected_filewithprogress_attributes_by_path( + self, {'ptype': self.progress_type})) + self.function_to_test(**self.test_args) + path = self.paths_to_upload[-1] + expected_attributes = expected_attributes_by_path[path] + fake_file_attributes = { + name: getattr(self.fake_filewithprogress, name) + for name in expected_attributes} + self.expectThat( + expected_attributes, + testtools.matchers.Equals(fake_file_attributes)) class http_upload_UnknownProtocolTestCase(http_upload_TestCase): @@ -1071,6 +1109,57 @@ class http_upload_UnknownProtocolTestCase(http_upload_TestCase): sys.exit.assert_called_with(self.expected_exit_status) +class http_upload_FileStatFailureTestCase(http_upload_TestCase): + """ Test cases for `methods.http.upload` function, `os.stat` failure. """ + + files_scenarios = list( + (scenario_name, scenario) for (scenario_name, scenario) + in upload_TestCase.files_scenarios + if scenario['paths_to_upload']) + + os_stat_scenarios = [ + ('os-stat-notfound', { + 'os_stat_scenario_name': "notfound_error", + 'expected_exit_status': EXIT_STATUS_FAILURE, + }), + ('os-stat-denied', { + 'os_stat_scenario_name': "denied_error", + 'expected_exit_status': EXIT_STATUS_FAILURE, + }), + ] + + response_scenarios = list( + (scenario_name, scenario) for (scenario_name, scenario) + in http_upload_SuccessTestCase.response_scenarios + if scenario['status_code'] == 200) + + scenarios = testscenarios.multiply_scenarios( + files_scenarios, + os_stat_scenarios, + upload_TestCase.incoming_scenarios, + http_upload_TestCase.protocol_scenarios, + http_upload_TestCase.login_scenarios, + response_scenarios, + http_upload_SuccessTestCase.auth_scenarios) + + def test_emits_error_message(self): + """ Should emit error message when `os.stat` failure. """ + try: + self.function_to_test(**self.test_args) + except FakeSystemExit: + pass + expected_output = textwrap.dedent("""\ + Determining size of file '{path}' failed + """).format(path=self.paths_to_upload[0]) + self.assertIn(expected_output, sys.stderr.getvalue()) + + def test_calls_sys_exit_with_expected_exit_status(self): + """ Should call `sys.exit` with expected exit status. """ + with testtools.ExpectedException(FakeSystemExit): + self.function_to_test(**self.test_args) + sys.exit.assert_called_with(self.expected_exit_status) + + class http_upload_ResponseErrorTestCase(http_upload_TestCase): """ Error test cases for `methods.http.upload` function. """ -- 2.11.4.GIT