From 9132b1106eeed41673e30c10d17be946a470729f Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Jun 01 2017 07:39:25 +0000 Subject: avoid test_compatrequests on py3 --- diff --git a/tests/test_lib/test_compatrequests.py b/tests/test_lib/test_compatrequests.py deleted file mode 100644 index c67f585..0000000 --- a/tests/test_lib/test_compatrequests.py +++ /dev/null @@ -1,267 +0,0 @@ -import httplib -import mock -import unittest -import urlparse - -import koji.compatrequests - - -class TestResponse(unittest.TestCase): - - def setUp(self): - session = mock.MagicMock() - response = mock.MagicMock() - self.response = koji.compatrequests.Response(session, response) - - def tearDown(self): - del self.response - - def test_read(self): - self.response.response.status = 200 - data = [ - "Here's some data", - "Here's some mooore data", - "And look!", - "Here's a nice block of lorem text", - "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do " - "eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut " - "enim ad minim veniam, quis nostrud exercitation ullamco laboris " - "nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor " - "in reprehenderit in voluptate velit esse cillum dolore eu fugiat " - "nulla pariatur. Excepteur sint occaecat cupidatat non proident, " - "sunt in culpa qui officia deserunt mollit anim id est laborum.", - "", #eof - ] - self.response.response.read.side_effect = data - - result = list(self.response.iter_content(blocksize=10240)) - - self.assertEqual(result, data[:-1]) - rcalls = [mock.call(10240) for s in data] - self.response.response.read.assert_has_calls(rcalls) - - self.response.close() - self.response.response.close.assert_called_once() - - def test_error(self): - self.response.response.status = 404 - self.response.response.getheader.return_value = 0 - with self.assertRaises(Exception): - list(self.response.iter_content(8192)) - self.response.response.read.assert_not_called() - - self.response.response.status = 404 - self.response.response.getheader.return_value = 42 - with self.assertRaises(Exception): - list(self.response.iter_content(8192)) - self.response.response.read.assert_called_once() - - self.response.response.status = 404 - self.response.response.reason = 'Not Found' - self.response.response.getheader.return_value = 42 - with self.assertRaises(httplib.HTTPException): - self.response.raise_for_status() - - - -class TestSessionPost(unittest.TestCase): - - def test_simple(self): - session = koji.compatrequests.Session() - url = 'https://www.fakedomain.org/KOJIHUB' - cnx = mock.MagicMock() - session.get_connection = mock.MagicMock() - session.get_connection.return_value = cnx - response = mock.MagicMock() - cnx.getresponse.return_value = response - - ret = session.post(url, data="data", headers={"foo": "bar"}) - cnx.putrequest.assert_called_once_with('POST', '/KOJIHUB') - cnx.putheader.assert_called_once_with('foo', 'bar') - cnx.send.assert_called_once_with("data") - self.assertEqual(ret.response, response) - - def test_less_simple(self): - session = koji.compatrequests.Session() - url = 'https://www.fakedomain.org/KOJIHUB?a=1&b=2' - cnx = mock.MagicMock() - session.get_connection = mock.MagicMock() - session.get_connection.return_value = cnx - response = mock.MagicMock() - cnx.getresponse.return_value = response - - ret = session.post(url, data="data", headers={"foo": "bar"}, - cert="cert", verify="verify", stream=True, timeout=1701) - cnx.putrequest.assert_called_once_with('POST', '/KOJIHUB?a=1&b=2') - cnx.putheader.assert_called_once_with('foo', 'bar') - cnx.send.assert_called_once_with("data") - self.assertEqual(ret.response, response) - - -class TestSessionConnection(unittest.TestCase): - - @mock.patch('httplib.HTTPConnection') - def test_http(self, HTTPConnection): - # no cert, no verify, no timeout - session = koji.compatrequests.Session() - url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - - cnx = session.get_connection(uri, None, None, None) - HTTPConnection.assert_called_once_with('www.fakedomain234234.org', 80) - key = ('http', 'www.fakedomain234234.org', None, None, None) - self.assertEqual(session.connection, (key, cnx)) - - # and close it - session.close() - self.assertEqual(session.connection, None) - cnx.close.assert_called_with() - - # double close should not error - session.close() - - def test_cached(self): - session = koji.compatrequests.Session() - url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - key = ('http', 'www.fakedomain234234.org', None, None, None) - cnx = mock.MagicMock() - session.connection = (key, cnx) - - ret = session.get_connection(uri, None, None, None) - self.assertEqual(ret, cnx) - - def test_badproto(self): - session = koji.compatrequests.Session() - url = 'nosuchproto://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - - with self.assertRaises(IOError): - ret = session.get_connection(uri, None, None, None) - - @mock.patch('httplib.HTTPConnection') - @mock.patch('sys.version_info', new=(2, 7, 12, 'final', 0)) - def test_timeout(self, HTTPConnection): - # no cert, no verify - session = koji.compatrequests.Session() - url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - timeout = 1701 - - cnx = session.get_connection(uri, None, None, 1701) - HTTPConnection.assert_called_once_with('www.fakedomain234234.org', 80, timeout=1701) - key = ('http', 'www.fakedomain234234.org', None, None, 1701) - self.assertEqual(session.connection, (key, cnx)) - - @mock.patch('httplib.HTTPConnection') - @mock.patch('sys.version_info', new=(2, 4, 3, 'final', 0)) - def test_timeout_compat(self, HTTPConnection): - # no cert, no verify - session = koji.compatrequests.Session() - url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - timeout = 1701 - - cnx = session.get_connection(uri, None, None, 1701) - HTTPConnection.assert_called_once_with('www.fakedomain234234.org', 80) - key = ('http', 'www.fakedomain234234.org', None, None, 1701) - self.assertEqual(session.connection, (key, cnx)) - cnx.connect.assert_called_once() - cnx.sock.settimeout.assert_called_with(1701) - - @mock.patch('httplib.HTTPSConnection') - def test_https(self, HTTPSConnection): - # no cert, no verify, no timeout - session = koji.compatrequests.Session() - url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - - cnx = session.get_connection(uri, None, None, None) - HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443) - key = ('https', 'www.fakedomain234234.org', None, None, None) - self.assertEqual(session.connection, (key, cnx)) - - @mock.patch('koji.ssl.SSLCommon.CreateSSLContext') - @mock.patch('koji.ssl.SSLCommon.PlgHTTPSConnection') - def test_cert(self, PlgHTTPSConnection, CreateSSLContext): - # no verify, no timeout - session = koji.compatrequests.Session() - url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - cert = '/path/to/cert/file' - context = mock.MagicMock() - CreateSSLContext.return_value = context - - cnx = session.get_connection(uri, cert, None, None) - PlgHTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, ssl_context=context) - key = ('https', 'www.fakedomain234234.org', cert, None, None) - self.assertEqual(session.connection, (key, cnx)) - - @mock.patch('ssl._create_unverified_context') - @mock.patch('httplib.HTTPSConnection') - @mock.patch('sys.version_info', new=(2, 7, 12, 'final', 0)) - def test_unverified(self, HTTPSConnection, create_unverified_context): - # no cert, verify=False, no timeout - session = koji.compatrequests.Session() - url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - context = mock.MagicMock() - create_unverified_context.return_value = context - - cnx = session.get_connection(uri, None, False, None) - create_unverified_context.assert_called_once() - HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, context=context) - key = ('https', 'www.fakedomain234234.org', None, False, None) - self.assertEqual(session.connection, (key, cnx)) - - @mock.patch('httplib.HTTPSConnection') - @mock.patch('sys.version_info', new=(2, 4, 3, 'final', 0)) - def test_unverified_compat(self, HTTPSConnection): - # no cert, verify=False, no timeout - session = koji.compatrequests.Session() - url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - - cnx = session.get_connection(uri, None, False, None) - HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443) - key = ('https', 'www.fakedomain234234.org', None, False, None) - self.assertEqual(session.connection, (key, cnx)) - - @mock.patch('ssl._create_unverified_context') - @mock.patch('ssl.SSLContext') - @mock.patch('httplib.HTTPSConnection') - @mock.patch('sys.version_info', new=(2, 7, 12, 'final', 0)) - def test_verify(self, HTTPSConnection, SSLContext, create_unverified_context): - # no cert, no timeout - session = koji.compatrequests.Session() - url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - context = mock.MagicMock() - SSLContext.return_value = context - verify = '/path/to/verify/cert' - - cnx = session.get_connection(uri, None, verify, None) - create_unverified_context.assert_not_called() - SSLContext.assert_called_once() - context.load_verify_locations.called_once_with(cafile=verify) - HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, context=context) - key = ('https', 'www.fakedomain234234.org', None, verify, None) - self.assertEqual(session.connection, (key, cnx)) - - @mock.patch('ssl._create_unverified_context') - @mock.patch('ssl.SSLContext') - @mock.patch('httplib.HTTPSConnection') - @mock.patch('sys.version_info', new=(2, 4, 3, 'final', 0)) - def test_verify_compat(self, HTTPSConnection, SSLContext, create_unverified_context): - # no cert, no timeout - session = koji.compatrequests.Session() - url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' - uri = urlparse.urlsplit(url) - verify = '/path/to/verify/cert' - - cnx = session.get_connection(uri, None, verify, None) - create_unverified_context.assert_not_called() - SSLContext.assert_not_called() - HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, cert_file=verify) - key = ('https', 'www.fakedomain234234.org', None, verify, None) - self.assertEqual(session.connection, (key, cnx)) diff --git a/tests/test_lib_compat/test_compatrequests.py b/tests/test_lib_compat/test_compatrequests.py new file mode 100644 index 0000000..c67f585 --- /dev/null +++ b/tests/test_lib_compat/test_compatrequests.py @@ -0,0 +1,267 @@ +import httplib +import mock +import unittest +import urlparse + +import koji.compatrequests + + +class TestResponse(unittest.TestCase): + + def setUp(self): + session = mock.MagicMock() + response = mock.MagicMock() + self.response = koji.compatrequests.Response(session, response) + + def tearDown(self): + del self.response + + def test_read(self): + self.response.response.status = 200 + data = [ + "Here's some data", + "Here's some mooore data", + "And look!", + "Here's a nice block of lorem text", + "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do " + "eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut " + "enim ad minim veniam, quis nostrud exercitation ullamco laboris " + "nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor " + "in reprehenderit in voluptate velit esse cillum dolore eu fugiat " + "nulla pariatur. Excepteur sint occaecat cupidatat non proident, " + "sunt in culpa qui officia deserunt mollit anim id est laborum.", + "", #eof + ] + self.response.response.read.side_effect = data + + result = list(self.response.iter_content(blocksize=10240)) + + self.assertEqual(result, data[:-1]) + rcalls = [mock.call(10240) for s in data] + self.response.response.read.assert_has_calls(rcalls) + + self.response.close() + self.response.response.close.assert_called_once() + + def test_error(self): + self.response.response.status = 404 + self.response.response.getheader.return_value = 0 + with self.assertRaises(Exception): + list(self.response.iter_content(8192)) + self.response.response.read.assert_not_called() + + self.response.response.status = 404 + self.response.response.getheader.return_value = 42 + with self.assertRaises(Exception): + list(self.response.iter_content(8192)) + self.response.response.read.assert_called_once() + + self.response.response.status = 404 + self.response.response.reason = 'Not Found' + self.response.response.getheader.return_value = 42 + with self.assertRaises(httplib.HTTPException): + self.response.raise_for_status() + + + +class TestSessionPost(unittest.TestCase): + + def test_simple(self): + session = koji.compatrequests.Session() + url = 'https://www.fakedomain.org/KOJIHUB' + cnx = mock.MagicMock() + session.get_connection = mock.MagicMock() + session.get_connection.return_value = cnx + response = mock.MagicMock() + cnx.getresponse.return_value = response + + ret = session.post(url, data="data", headers={"foo": "bar"}) + cnx.putrequest.assert_called_once_with('POST', '/KOJIHUB') + cnx.putheader.assert_called_once_with('foo', 'bar') + cnx.send.assert_called_once_with("data") + self.assertEqual(ret.response, response) + + def test_less_simple(self): + session = koji.compatrequests.Session() + url = 'https://www.fakedomain.org/KOJIHUB?a=1&b=2' + cnx = mock.MagicMock() + session.get_connection = mock.MagicMock() + session.get_connection.return_value = cnx + response = mock.MagicMock() + cnx.getresponse.return_value = response + + ret = session.post(url, data="data", headers={"foo": "bar"}, + cert="cert", verify="verify", stream=True, timeout=1701) + cnx.putrequest.assert_called_once_with('POST', '/KOJIHUB?a=1&b=2') + cnx.putheader.assert_called_once_with('foo', 'bar') + cnx.send.assert_called_once_with("data") + self.assertEqual(ret.response, response) + + +class TestSessionConnection(unittest.TestCase): + + @mock.patch('httplib.HTTPConnection') + def test_http(self, HTTPConnection): + # no cert, no verify, no timeout + session = koji.compatrequests.Session() + url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + + cnx = session.get_connection(uri, None, None, None) + HTTPConnection.assert_called_once_with('www.fakedomain234234.org', 80) + key = ('http', 'www.fakedomain234234.org', None, None, None) + self.assertEqual(session.connection, (key, cnx)) + + # and close it + session.close() + self.assertEqual(session.connection, None) + cnx.close.assert_called_with() + + # double close should not error + session.close() + + def test_cached(self): + session = koji.compatrequests.Session() + url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + key = ('http', 'www.fakedomain234234.org', None, None, None) + cnx = mock.MagicMock() + session.connection = (key, cnx) + + ret = session.get_connection(uri, None, None, None) + self.assertEqual(ret, cnx) + + def test_badproto(self): + session = koji.compatrequests.Session() + url = 'nosuchproto://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + + with self.assertRaises(IOError): + ret = session.get_connection(uri, None, None, None) + + @mock.patch('httplib.HTTPConnection') + @mock.patch('sys.version_info', new=(2, 7, 12, 'final', 0)) + def test_timeout(self, HTTPConnection): + # no cert, no verify + session = koji.compatrequests.Session() + url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + timeout = 1701 + + cnx = session.get_connection(uri, None, None, 1701) + HTTPConnection.assert_called_once_with('www.fakedomain234234.org', 80, timeout=1701) + key = ('http', 'www.fakedomain234234.org', None, None, 1701) + self.assertEqual(session.connection, (key, cnx)) + + @mock.patch('httplib.HTTPConnection') + @mock.patch('sys.version_info', new=(2, 4, 3, 'final', 0)) + def test_timeout_compat(self, HTTPConnection): + # no cert, no verify + session = koji.compatrequests.Session() + url = 'http://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + timeout = 1701 + + cnx = session.get_connection(uri, None, None, 1701) + HTTPConnection.assert_called_once_with('www.fakedomain234234.org', 80) + key = ('http', 'www.fakedomain234234.org', None, None, 1701) + self.assertEqual(session.connection, (key, cnx)) + cnx.connect.assert_called_once() + cnx.sock.settimeout.assert_called_with(1701) + + @mock.patch('httplib.HTTPSConnection') + def test_https(self, HTTPSConnection): + # no cert, no verify, no timeout + session = koji.compatrequests.Session() + url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + + cnx = session.get_connection(uri, None, None, None) + HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443) + key = ('https', 'www.fakedomain234234.org', None, None, None) + self.assertEqual(session.connection, (key, cnx)) + + @mock.patch('koji.ssl.SSLCommon.CreateSSLContext') + @mock.patch('koji.ssl.SSLCommon.PlgHTTPSConnection') + def test_cert(self, PlgHTTPSConnection, CreateSSLContext): + # no verify, no timeout + session = koji.compatrequests.Session() + url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + cert = '/path/to/cert/file' + context = mock.MagicMock() + CreateSSLContext.return_value = context + + cnx = session.get_connection(uri, cert, None, None) + PlgHTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, ssl_context=context) + key = ('https', 'www.fakedomain234234.org', cert, None, None) + self.assertEqual(session.connection, (key, cnx)) + + @mock.patch('ssl._create_unverified_context') + @mock.patch('httplib.HTTPSConnection') + @mock.patch('sys.version_info', new=(2, 7, 12, 'final', 0)) + def test_unverified(self, HTTPSConnection, create_unverified_context): + # no cert, verify=False, no timeout + session = koji.compatrequests.Session() + url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + context = mock.MagicMock() + create_unverified_context.return_value = context + + cnx = session.get_connection(uri, None, False, None) + create_unverified_context.assert_called_once() + HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, context=context) + key = ('https', 'www.fakedomain234234.org', None, False, None) + self.assertEqual(session.connection, (key, cnx)) + + @mock.patch('httplib.HTTPSConnection') + @mock.patch('sys.version_info', new=(2, 4, 3, 'final', 0)) + def test_unverified_compat(self, HTTPSConnection): + # no cert, verify=False, no timeout + session = koji.compatrequests.Session() + url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + + cnx = session.get_connection(uri, None, False, None) + HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443) + key = ('https', 'www.fakedomain234234.org', None, False, None) + self.assertEqual(session.connection, (key, cnx)) + + @mock.patch('ssl._create_unverified_context') + @mock.patch('ssl.SSLContext') + @mock.patch('httplib.HTTPSConnection') + @mock.patch('sys.version_info', new=(2, 7, 12, 'final', 0)) + def test_verify(self, HTTPSConnection, SSLContext, create_unverified_context): + # no cert, no timeout + session = koji.compatrequests.Session() + url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + context = mock.MagicMock() + SSLContext.return_value = context + verify = '/path/to/verify/cert' + + cnx = session.get_connection(uri, None, verify, None) + create_unverified_context.assert_not_called() + SSLContext.assert_called_once() + context.load_verify_locations.called_once_with(cafile=verify) + HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, context=context) + key = ('https', 'www.fakedomain234234.org', None, verify, None) + self.assertEqual(session.connection, (key, cnx)) + + @mock.patch('ssl._create_unverified_context') + @mock.patch('ssl.SSLContext') + @mock.patch('httplib.HTTPSConnection') + @mock.patch('sys.version_info', new=(2, 4, 3, 'final', 0)) + def test_verify_compat(self, HTTPSConnection, SSLContext, create_unverified_context): + # no cert, no timeout + session = koji.compatrequests.Session() + url = 'https://www.fakedomain234234.org/KOJIHUB?a=1&b=2' + uri = urlparse.urlsplit(url) + verify = '/path/to/verify/cert' + + cnx = session.get_connection(uri, None, verify, None) + create_unverified_context.assert_not_called() + SSLContext.assert_not_called() + HTTPSConnection.assert_called_once_with('www.fakedomain234234.org', 443, cert_file=verify) + key = ('https', 'www.fakedomain234234.org', None, verify, None) + self.assertEqual(session.connection, (key, cnx))