move from urrlib.request.urlopen to requests.get
Fixes: https://pagure.io/koji/issue/1530
This commit is contained in:
parent
8920d9d613
commit
aad9fac8d9
5 changed files with 58 additions and 52 deletions
|
|
@ -62,6 +62,7 @@ class TestFastUpload(unittest.TestCase):
|
|||
|
||||
def setUp(self):
|
||||
self.ksession = koji.ClientSession('http://koji.example.com/kojihub')
|
||||
self.ksession.logout = mock.MagicMock()
|
||||
self.do_fake_login()
|
||||
# mocks
|
||||
self.ksession._callMethod = mock.MagicMock()
|
||||
|
|
@ -89,7 +90,6 @@ class TestFastUpload(unittest.TestCase):
|
|||
self.ksession.fastUpload('nonexistent_file', 'target')
|
||||
|
||||
def test_fastUpload_nofile(self):
|
||||
|
||||
# fail with nonexistent file (IOError)
|
||||
self.file_mock.side_effect = IOError('mocked exception')
|
||||
with self.assertRaises(IOError):
|
||||
|
|
@ -180,6 +180,7 @@ class TestMultiCall(unittest.TestCase):
|
|||
self.ksession = koji.ClientSession('http://koji.example.com/kojihub')
|
||||
# mocks
|
||||
self.ksession._sendCall = mock.MagicMock()
|
||||
self.ksession.logout = mock.MagicMock()
|
||||
|
||||
def tearDown(self):
|
||||
del self.ksession
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ except ImportError:
|
|||
from os import path, makedirs
|
||||
from tempfile import gettempdir
|
||||
from mock import patch, MagicMock, Mock, call
|
||||
import requests_mock
|
||||
|
||||
import koji
|
||||
from koji.tasks import BaseTaskHandler, FakeTask, ForkTask, SleepTask, \
|
||||
|
|
@ -452,8 +453,8 @@ class TasksTestCase(unittest.TestCase):
|
|||
obj = TestTask(123, 'some_method', ['random_arg'], None, options, temp_path)
|
||||
self.assertEquals(obj.localPath('test.txt'), dummy_file)
|
||||
|
||||
@patch('six.moves.urllib.request.urlopen', return_value=six.BytesIO(six.b('Important things\nSome more important things\n')))
|
||||
def test_BaseTaskHandler_localPath_no_file(self, mock_urlopen):
|
||||
@requests_mock.Mocker()
|
||||
def test_BaseTaskHandler_localPath_no_file(self, m_requests):
|
||||
"""
|
||||
"""
|
||||
temp_path = get_tmp_dir_path('TestTask')
|
||||
|
|
@ -466,10 +467,13 @@ class TasksTestCase(unittest.TestCase):
|
|||
|
||||
options = Mock()
|
||||
options.topurl = 'https://www.domain.local'
|
||||
url = options.topurl + '/test.txt'
|
||||
m_requests.register_uri('GET', url, text='Important things\nSome more important things\n')
|
||||
obj = TestTask(123, 'some_method', ['random_arg'], None, options, temp_path)
|
||||
|
||||
self.assertEquals(obj.localPath('test.txt'), target_file_path)
|
||||
mock_urlopen.assert_called_once_with('https://www.domain.local/test.txt')
|
||||
self.assertEquals(m_requests.call_count, 1)
|
||||
self.assertEquals(m_requests.request_history[0].url, url)
|
||||
|
||||
def test_BaseTaskHandler_localPath_no_topurl(self):
|
||||
""" Tests that the localPath function returns a path when options.topurl is not defined.
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ try:
|
|||
except ImportError:
|
||||
import unittest
|
||||
|
||||
import requests_mock
|
||||
from mock import call, patch
|
||||
from datetime import datetime
|
||||
import koji
|
||||
|
|
@ -101,62 +102,64 @@ class MiscFunctionTestCase(unittest.TestCase):
|
|||
move.assert_not_called()
|
||||
|
||||
@mock_open()
|
||||
@mock.patch('six.moves.urllib.request.urlopen')
|
||||
@mock.patch('tempfile.TemporaryFile')
|
||||
@mock.patch('shutil.copyfileobj')
|
||||
def test_openRemoteFile(self, m_copyfileobj, m_TemporaryFile,
|
||||
m_urlopen, m_open):
|
||||
def test_openRemoteFile(self, m_TemporaryFile, m_open):
|
||||
"""Test openRemoteFile function"""
|
||||
|
||||
mocks = [m_open, m_copyfileobj, m_TemporaryFile, m_urlopen]
|
||||
mocks = [m_open, m_TemporaryFile ]
|
||||
|
||||
topurl = 'http://example.com/koji'
|
||||
path = 'relative/file/path'
|
||||
url = 'http://example.com/koji/relative/file/path'
|
||||
|
||||
# using topurl, no tempfile
|
||||
fo = koji.openRemoteFile(path, topurl)
|
||||
m_urlopen.assert_called_once_with(url)
|
||||
m_urlopen.return_value.close.assert_called_once()
|
||||
m_TemporaryFile.assert_called_once_with(dir=None)
|
||||
m_copyfileobj.assert_called_once()
|
||||
m_open.assert_not_called()
|
||||
assert fo is m_TemporaryFile.return_value
|
||||
with requests_mock.Mocker() as m_requests:
|
||||
m_requests.register_uri('GET', url, text='random content')
|
||||
# using topurl, no tempfile
|
||||
fo = koji.openRemoteFile(path, topurl)
|
||||
self.assertEqual(m_requests.call_count, 1)
|
||||
self.assertEqual(m_requests.request_history[0].url, url)
|
||||
m_TemporaryFile.assert_called_once_with(dir=None)
|
||||
m_open.assert_not_called()
|
||||
assert fo is m_TemporaryFile.return_value
|
||||
|
||||
for m in mocks:
|
||||
m.reset_mock()
|
||||
|
||||
# using topurl + tempfile
|
||||
tempdir = '/tmp/koji/1234'
|
||||
fo = koji.openRemoteFile(path, topurl, tempdir=tempdir)
|
||||
m_urlopen.assert_called_once_with(url)
|
||||
m_urlopen.return_value.close.assert_called_once()
|
||||
m_TemporaryFile.assert_called_once_with(dir=tempdir)
|
||||
m_copyfileobj.assert_called_once()
|
||||
m_open.assert_not_called()
|
||||
assert fo is m_TemporaryFile.return_value
|
||||
with requests_mock.Mocker() as m_requests:
|
||||
m_requests.register_uri('GET', url, text='random content')
|
||||
|
||||
# using topurl + tempfile
|
||||
tempdir = '/tmp/koji/1234'
|
||||
fo = koji.openRemoteFile(path, topurl, tempdir=tempdir)
|
||||
self.assertEqual(m_requests.call_count, 1)
|
||||
self.assertEqual(m_requests.request_history[0].url, url)
|
||||
m_TemporaryFile.assert_called_once_with(dir=tempdir)
|
||||
m_open.assert_not_called()
|
||||
assert fo is m_TemporaryFile.return_value
|
||||
|
||||
for m in mocks:
|
||||
m.reset_mock()
|
||||
|
||||
# using topdir
|
||||
topdir = '/mnt/mykojidir'
|
||||
filename = '/mnt/mykojidir/relative/file/path'
|
||||
fo = koji.openRemoteFile(path, topdir=topdir)
|
||||
m_urlopen.assert_not_called()
|
||||
m_TemporaryFile.assert_not_called()
|
||||
m_copyfileobj.assert_not_called()
|
||||
m_open.assert_called_once_with(filename)
|
||||
assert fo is m_open.return_value
|
||||
with requests_mock.Mocker() as m_requests:
|
||||
m_requests.register_uri('GET', url, text='random content')
|
||||
# using topdir
|
||||
topdir = '/mnt/mykojidir'
|
||||
filename = '/mnt/mykojidir/relative/file/path'
|
||||
fo = koji.openRemoteFile(path, topdir=topdir)
|
||||
self.assertEqual(m_requests.call_count, 0)
|
||||
m_TemporaryFile.assert_not_called()
|
||||
m_open.assert_called_once_with(filename)
|
||||
assert fo is m_open.return_value
|
||||
|
||||
for m in mocks:
|
||||
m.reset_mock()
|
||||
|
||||
# using neither
|
||||
with self.assertRaises(koji.GenericError):
|
||||
koji.openRemoteFile(path)
|
||||
for m in mocks:
|
||||
m.assert_not_called()
|
||||
with requests_mock.Mocker() as m_requests:
|
||||
m_requests.register_uri('GET', url, text='random content')
|
||||
# using neither
|
||||
with self.assertRaises(koji.GenericError):
|
||||
koji.openRemoteFile(path)
|
||||
for m in mocks:
|
||||
m.assert_not_called()
|
||||
|
||||
def test_joinpath_bad(self):
|
||||
bad_joins = [
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue