Skip to content
Merged
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
50 changes: 50 additions & 0 deletions tests/test_query.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,13 @@
"""Tests for the ``CloudRecoService`` querying functionality."""

import datetime
import io
import uuid
from typing import BinaryIO

import pytest
import requests
from freezegun import freeze_time
from mock_vws import MockVWS
from mock_vws.database import VuforiaDatabase

Expand Down Expand Up @@ -42,6 +46,52 @@ def test_match(
assert matching_target.target_id == target_id


class TestDefaultRequestTimeout:
"""Tests for the default request timeout."""

@staticmethod
@pytest.mark.parametrize(
argnames=("response_delay_seconds", "expect_timeout"),
argvalues=[(29, False), (31, True)],
)
def test_default_timeout(
image: io.BytesIO | BinaryIO,
*,
response_delay_seconds: int,
expect_timeout: bool,
) -> None:
"""At 29 seconds there is no error; at 31 seconds there is a
timeout.
"""
with (
freeze_time() as frozen_datetime,
MockVWS(
response_delay_seconds=response_delay_seconds,
sleep_fn=lambda seconds: (
frozen_datetime.tick(
delta=datetime.timedelta(seconds=seconds),
),
None,
)[1],
) as mock,
):
database = VuforiaDatabase()
mock.add_database(database=database)
cloud_reco_client = CloudRecoService(
client_access_key=database.client_access_key,
client_secret_key=database.client_secret_key,
)

if expect_timeout:
with pytest.raises(
expected_exception=requests.exceptions.Timeout,
):
cloud_reco_client.query(image=image)
else:
matches = cloud_reco_client.query(image=image)
assert not matches


class TestCustomBaseVWQURL:
"""Tests for using a custom base VWQ URL."""

Expand Down
58 changes: 58 additions & 0 deletions tests/test_vws.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from typing import BinaryIO

import pytest
import requests
from freezegun import freeze_time
from mock_vws import MockVWS
from mock_vws.database import VuforiaDatabase
Expand Down Expand Up @@ -92,6 +93,63 @@ def test_add_two_targets(
)


class TestDefaultRequestTimeout:
"""Tests for the default request timeout."""

@staticmethod
@pytest.mark.parametrize(
argnames=("response_delay_seconds", "expect_timeout"),
argvalues=[(29, False), (31, True)],
)
def test_default_timeout(
image: io.BytesIO | BinaryIO,
*,
response_delay_seconds: int,
expect_timeout: bool,
) -> None:
"""At 29 seconds there is no error; at 31 seconds there is a
timeout.
"""
with (
freeze_time() as frozen_datetime,
MockVWS(
response_delay_seconds=response_delay_seconds,
sleep_fn=lambda seconds: (
frozen_datetime.tick(
delta=datetime.timedelta(seconds=seconds),
),
None,
)[1],
) as mock,
):
database = VuforiaDatabase()
mock.add_database(database=database)
vws_client = VWS(
server_access_key=database.server_access_key,
server_secret_key=database.server_secret_key,
)

if expect_timeout:
with pytest.raises(
expected_exception=requests.exceptions.Timeout,
):
vws_client.add_target(
name="x",
width=1,
image=image,
active_flag=True,
application_metadata=None,
)
else:
vws_client.add_target(
name="x",
width=1,
image=image,
active_flag=True,
application_metadata=None,
)


class TestCustomBaseVWSURL:
"""Tests for using a custom base VWS URL."""

Expand Down