Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 34 additions & 7 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -460,12 +460,17 @@ def test_client_generate_images(httpserver: HTTPServer):

client = Client(httpserver.url_for('/'))

with tempfile.NamedTemporaryFile() as temp:
with tempfile.NamedTemporaryFile(delete=False) as temp:
temp.write(PNG_BYTES)
temp.flush()
response = client.generate('dummy', 'Why is the sky blue?', images=[temp.name])
assert response['model'] == 'dummy'
assert response['response'] == 'Because it is.'

try:
os.remove(temp.name)
except FileNotFoundError:
pass


def test_client_generate_format_json(httpserver: HTTPServer):
Expand Down Expand Up @@ -850,19 +855,29 @@ def test_client_create_blob(httpserver: HTTPServer):

client = Client(httpserver.url_for('/'))

with tempfile.NamedTemporaryFile() as blob:
with tempfile.NamedTemporaryFile(delete=False) as blob:
response = client.create_blob(blob.name)
assert response == 'sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'

try:
os.remove(blob.name)
except FileNotFoundError:
pass


def test_client_create_blob_exists(httpserver: HTTPServer):
httpserver.expect_ordered_request(PrefixPattern('/api/blobs/'), method='POST').respond_with_response(Response(status=200))

client = Client(httpserver.url_for('/'))

with tempfile.NamedTemporaryFile() as blob:
with tempfile.NamedTemporaryFile(delete=False) as blob:
response = client.create_blob(blob.name)
assert response == 'sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'

try:
os.remove(blob.name)
except FileNotFoundError:
pass


def test_client_delete(httpserver: HTTPServer):
Expand Down Expand Up @@ -1055,13 +1070,17 @@ async def test_async_client_generate_images(httpserver: HTTPServer):

client = AsyncClient(httpserver.url_for('/'))

with tempfile.NamedTemporaryFile() as temp:
with tempfile.NamedTemporaryFile(delete=False) as temp:
temp.write(PNG_BYTES)
temp.flush()
response = await client.generate('dummy', 'Why is the sky blue?', images=[temp.name])
assert response['model'] == 'dummy'
assert response['response'] == 'Because it is.'

try:
os.remove(temp.name)
except FileNotFoundError:
pass

async def test_async_client_pull(httpserver: HTTPServer):
httpserver.expect_ordered_request(
Expand Down Expand Up @@ -1227,20 +1246,28 @@ async def test_async_client_create_blob(httpserver: HTTPServer):

client = AsyncClient(httpserver.url_for('/'))

with tempfile.NamedTemporaryFile() as blob:
with tempfile.NamedTemporaryFile(delete=False) as blob:
response = await client.create_blob(blob.name)
assert response == 'sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'

try:
os.remove(blob.name)
except FileNotFoundError:
pass

async def test_async_client_create_blob_exists(httpserver: HTTPServer):
httpserver.expect_ordered_request(PrefixPattern('/api/blobs/'), method='POST').respond_with_response(Response(status=200))

client = AsyncClient(httpserver.url_for('/'))

with tempfile.NamedTemporaryFile() as blob:
with tempfile.NamedTemporaryFile(delete=False) as blob:
response = await client.create_blob(blob.name)
assert response == 'sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'


try:
os.remove(blob.name)
except FileNotFoundError:
pass

async def test_async_client_delete(httpserver: HTTPServer):
httpserver.expect_ordered_request(PrefixPattern('/api/delete'), method='DELETE').respond_with_response(Response(status=200))
Expand Down
15 changes: 13 additions & 2 deletions tests/test_type_serialization.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import tempfile
import os
from base64 import b64encode
from pathlib import Path

Expand Down Expand Up @@ -32,20 +33,30 @@ def test_image_serialization_plain_string():


def test_image_serialization_path():
with tempfile.NamedTemporaryFile() as temp_file:
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_file.write(b'test file content')
temp_file.flush()
img = Image(value=Path(temp_file.name))
assert img.model_dump() == b64encode(b'test file content').decode()

try:
os.remove(temp_file.name)
except FileNotFoundError:
pass


def test_image_serialization_string_path():
with tempfile.NamedTemporaryFile() as temp_file:
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_file.write(b'test file content')
temp_file.flush()
img = Image(value=temp_file.name)
assert img.model_dump() == b64encode(b'test file content').decode()

try:
os.remove(temp_file.name)
except FileNotFoundError:
pass

with pytest.raises(ValueError):
img = Image(value='some_path/that/does/not/exist.png')
img.model_dump()
Expand Down