Skip to content
This repository was archived by the owner on Oct 30, 2026. It is now read-only.
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
51 changes: 51 additions & 0 deletions test-backends/src/vws_mock_test_backends/http_transport.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
"""HTTP transport for unmatched contract-suite requests."""

from http import HTTPMethod

import requests
from beartype import beartype

from vws_mock_contract_tests.utils.unrouted_requests import (
UnroutedRequests,
UnroutedResponse,
)


@beartype
def http_unrouted_requests(
*, base_vws_url: str, returns_response: bool
) -> UnroutedRequests:
"""Bind unmatched requests to an endpoint and its response policy."""

@beartype
def send_http_request(
*,
method: HTTPMethod,
request_path: str,
headers: dict[str, str],
) -> UnroutedResponse | None:
"""Send an unmatched request to the configured VWS URL."""
try:
response = requests.request(
method=method,
url=base_vws_url + request_path,
headers=headers,
timeout=30,
)
except requests.exceptions.ConnectionError:
return None

return UnroutedResponse(
status_code=response.status_code,
body=response.content,
content_type=response.headers.get("Content-Type"),
content_encoding=response.headers.get("Content-Encoding"),
upstream_service_time=response.headers.get(
"x-envoy-upstream-service-time",
),
)

return UnroutedRequests(
send=send_http_request,
returns_response=returns_response,
)
36 changes: 3 additions & 33 deletions test-backends/src/vws_mock_test_backends/vuforia_backends.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,8 @@
import logging
from collections.abc import Generator
from enum import Enum
from http import HTTPMethod

import pytest
import requests
from beartype import beartype
from pytest_multi_backend import backend_fixture
from vws import VWS
Expand All @@ -28,7 +26,6 @@
)
from vws_mock_contract_tests.utils.unrouted_requests import (
UnroutedRequests,
UnroutedResponse,
)
from vws_mock_test_backends.flask_backend import (
add_cloud_database,
Expand All @@ -37,6 +34,7 @@
seed_flask_recognition_counts,
send_flask_unrouted_request,
)
from vws_mock_test_backends.http_transport import http_unrouted_requests
from vws_mock_test_backends.in_memory_backend import (
enable_in_memory_model_targets,
enable_in_memory_vuforia,
Expand Down Expand Up @@ -480,36 +478,8 @@ def fixture_unrouted_requests(
returns_response=True,
)

@beartype
def send_http_request(
*,
method: HTTPMethod,
request_path: str,
headers: dict[str, str],
) -> UnroutedResponse | None:
"""Send an unmatched request to the configured VWS URL."""
try:
response = requests.request(
method=method,
url=base_vws_url + request_path,
headers=headers,
timeout=30,
)
except requests.exceptions.ConnectionError:
return None

return UnroutedResponse(
status_code=response.status_code,
body=response.content,
content_type=response.headers.get("Content-Type"),
content_encoding=response.headers.get("Content-Encoding"),
upstream_service_time=response.headers.get(
"x-envoy-upstream-service-time",
),
)

return UnroutedRequests(
send=send_http_request,
return http_unrouted_requests(
base_vws_url=base_vws_url,
returns_response=_vws_backend != VuforiaBackend.MOCK,
)

Expand Down
Loading