|
1 | 1 | """Tests for the usage of the mock for ``httpx`` via ``respx``.""" |
2 | 2 |
|
| 3 | +import json |
3 | 4 | import socket |
| 5 | +import uuid |
| 6 | +from http import HTTPMethod, HTTPStatus |
4 | 7 |
|
5 | 8 | import httpx |
6 | 9 | import pytest |
7 | | -from vws_auth_tools import rfc_1123_date |
| 10 | +from vws_auth_tools import authorization_header, rfc_1123_date |
8 | 11 |
|
9 | 12 | from mock_vws import MissingSchemeError, MockVWSForHttpx |
10 | 13 | from mock_vws.database import CloudDatabase, VuMarkDatabase |
| 14 | +from mock_vws.target import VuMarkTarget |
11 | 15 |
|
12 | 16 |
|
13 | 17 | def _request_unmocked_address() -> None: |
@@ -208,6 +212,46 @@ def test_custom_base_vwq_url() -> None: |
208 | 212 | timeout=30, |
209 | 213 | ) |
210 | 214 |
|
| 215 | + @staticmethod |
| 216 | + def test_custom_base_vws_url_with_path_prefix() -> None: |
| 217 | + """A custom base VWS URL with a path prefix intercepts at the |
| 218 | + prefix. |
| 219 | + """ |
| 220 | + with MockVWSForHttpx( |
| 221 | + base_vws_url="https://vuforia.vws.example.com/prefix", |
| 222 | + real_http=False, |
| 223 | + ): |
| 224 | + with pytest.raises(expected_exception=httpx.ConnectError): |
| 225 | + httpx.get( |
| 226 | + url="https://vuforia.vws.example.com/summary", |
| 227 | + timeout=30, |
| 228 | + ) |
| 229 | + |
| 230 | + httpx.get( |
| 231 | + url="https://vuforia.vws.example.com/prefix/summary", |
| 232 | + timeout=30, |
| 233 | + ) |
| 234 | + |
| 235 | + @staticmethod |
| 236 | + def test_custom_base_vwq_url_with_path_prefix() -> None: |
| 237 | + """A custom base VWQ URL with a path prefix intercepts at the |
| 238 | + prefix. |
| 239 | + """ |
| 240 | + with MockVWSForHttpx( |
| 241 | + base_vwq_url="https://vuforia.vwq.example.com/prefix", |
| 242 | + real_http=False, |
| 243 | + ): |
| 244 | + with pytest.raises(expected_exception=httpx.ConnectError): |
| 245 | + httpx.post( |
| 246 | + url="https://vuforia.vwq.example.com/v1/query", |
| 247 | + timeout=30, |
| 248 | + ) |
| 249 | + |
| 250 | + httpx.post( |
| 251 | + url="https://vuforia.vwq.example.com/prefix/v1/query", |
| 252 | + timeout=30, |
| 253 | + ) |
| 254 | + |
211 | 255 | @staticmethod |
212 | 256 | def test_no_scheme() -> None: |
213 | 257 | """An error is raised if a URL is given with no scheme.""" |
@@ -348,3 +392,40 @@ def test_database_summary() -> None: |
348 | 392 | ) |
349 | 393 | # We just verify we get a response (auth will fail but endpoint works) |
350 | 394 | assert response.status_code is not None |
| 395 | + |
| 396 | + @staticmethod |
| 397 | + def test_vumark_bytes_response() -> None: |
| 398 | + """The VuMark endpoint returns bytes content via httpx.""" |
| 399 | + vumark_target = VuMarkTarget(name="test-target") |
| 400 | + database = VuMarkDatabase(vumark_targets={vumark_target}) |
| 401 | + target_id = vumark_target.target_id |
| 402 | + request_path = f"/targets/{target_id}/instances" |
| 403 | + content_type = "application/json" |
| 404 | + content = json.dumps(obj={"instance_id": uuid.uuid4().hex}).encode( |
| 405 | + encoding="utf-8" |
| 406 | + ) |
| 407 | + date = rfc_1123_date() |
| 408 | + auth = authorization_header( |
| 409 | + access_key=database.server_access_key, |
| 410 | + secret_key=database.server_secret_key, |
| 411 | + method=HTTPMethod.POST, |
| 412 | + content=content, |
| 413 | + content_type=content_type, |
| 414 | + date=date, |
| 415 | + request_path=request_path, |
| 416 | + ) |
| 417 | + with MockVWSForHttpx() as mock: |
| 418 | + mock.add_vumark_database(vumark_database=database) |
| 419 | + response = httpx.post( |
| 420 | + url="https://vws.vuforia.com" + request_path, |
| 421 | + headers={ |
| 422 | + "Accept": "image/png", |
| 423 | + "Authorization": auth, |
| 424 | + "Content-Length": str(object=len(content)), |
| 425 | + "Content-Type": content_type, |
| 426 | + "Date": date, |
| 427 | + }, |
| 428 | + content=content, |
| 429 | + timeout=30, |
| 430 | + ) |
| 431 | + assert response.status_code == HTTPStatus.OK |
0 commit comments