Reputation: 14614
I have some python code which successfully downloads an image from a URL, using requests, and saves it into /tmp/
. I want to test this does what it should. I'm using responses to test fetching of JSON files, but I'm not sure how to mock the behaviour of fetching a file.
I assume it'd be similar to mocking a standard response, like the below, but I think I'm blanking on how to set the body
to be a file...
@responses.activate
def test_download():
responses.add(responses.GET, 'http://example.org/images/my_image.jpg',
body='', status=200,
content_type='image/jpeg')
#...
UPDATE: Following Ashafix's comment, I'm trying this (python 3):
from io import BytesIO
@responses.activate
def test_download():
with open('tests/images/tester.jpg', 'rb') as img1:
imgIO = BytesIO(img1.read())
responses.add(responses.GET, 'http://example.org/images/my_image.jpg',
body=imgIO, status=200,
content_type='image/jpeg')
imgIO.seek(0)
#...
But when, subsequently, the code I'm testing attempts to do the request I get:
a bytes-like object is required, not '_io.BytesIO'
Feels like it's almost right, but I'm stumped.
UPDATE 2: Trying to follow Steve Jessop's suggestion:
@responses.activate
def test_download():
with open('tests/images/tester.jpg', 'rb') as img1:
responses.add(responses.GET, 'http://example.org/images/my_image.jpg',
body=img1.read(), status=200,
content_type='image/jpeg')
#...
But this time the code being tested raises this:
I/O operation on closed file.
Surely the image should still be open inside the with
block?
UPDATE 3: The code I'm testing is something like this:
r = requests.get(url, stream=True)
if r.status_code == 200:
with open('/tmp/temp.jpg', 'wb') as f:
r.raw.decode_content = True
shutil.copyfileobj(r.raw, f)
It seems to be that the final shutil
line is generating the "I/O operation on closed file." error. I don't understand this enough - the streaming of the file - to know how best to mock this behaviour, to test the downloaded file is saved to /tmp/
.
Upvotes: 9
Views: 11916
Reputation: 768
You might need to pass stream=True
to the responses.add
call. Something like:
@responses.activate
def test_download():
with open("tests/images/tester.jpg", "rb") as img1:
responses.add(
responses.GET,
"http://example.org/images/my_image.jpg",
body=img1.read(),
status=200,
content_type="image/jpeg",
stream=True,
)
Upvotes: 8
Reputation: 14614
First, to summarise my now overly long question... I'm testing some code that's something like:
def download_file(url):
r = requests.get(url, stream=True)
if r.status_code == 200:
filename = os.path.basename(url)
with open('/tmp/%s' % filename, 'wb') as f:
r.raw.decode_content = True
shutil.copyfileobj(r.raw, f)
return filename
It downloads an image and, streaming it, saves it to /tmp/
. I wanted to mock the request so I can test other things.
@responses.activate
def test_downloads_file(self):
url = 'http://example.org/test.jpg'
with open('tests/images/tester.jpg', 'rb') as img:
responses.add(responses.GET, url,
body=img.read(), status=200,
content_type='image/jpg',
adding_headers={'Transfer-Encoding': 'chunked'})
filename = download_file(url)
# assert things here.
Once I had worked out the way to use open()
for this, I was still getting "I/O operation on closed file." from shutil.copyfileobj()
. The thing that's stopped this is to add in the Transfer-Encoding
header, which is present in the headers when I make the real request.
Any suggestions for other, better solutions very welcome!
Upvotes: 1