Merge pull request #46 from unmade/switch-test-to-httpbin

Switch test to httpbin

.github/workflows/lint-and-test.yml

@@ -52,10 +52,16 @@ jobs:
       run: |
         tox -e lint
 
-    - name: Test
+    - name: Basic Test Suit
       run: |
         tox -e py
+
+    - name: Test AioHttp Driver
+      run: |
         tox -e py-aiohttp
+
+    - name: Test Requests Driver
+      run: |
         tox -e py-requests
 
     - name: Generate coverage report

src/apiwrappers/drivers/aiohttp.py

@@ -70,6 +70,8 @@ async def fetch(
                 )
             except asyncio.TimeoutError as exc:
                 raise exceptions.Timeout from exc
+            except aiohttp.ClientSSLError as exc:
+                raise ssl.SSLError(str(exc)) from exc
             except aiohttp.ClientConnectionError as exc:
                 raise exceptions.ConnectionFailed from exc
             except aiohttp.ClientError as exc:

tests/test_drivers/conftest.py

@@ -0,0 +1,13 @@
+# pylint: disable=import-outside-toplevel
+
+import pytest
+
+
+@pytest.fixture
+def httpbin_ca_bundle():
+    # original httpbin_ca_bundle fixture only sets REQUESTS_CA_BUNDLE env.
+    # This is not suitable for other HTTP clients, so instead, we return
+    # path to a valid CA bundle.
+    from pytest_httpbin import certs
+
+    return certs.where()

tests/test_drivers/httpbin_client.py

@@ -0,0 +1,74 @@
+from apiwrappers import Method, Request, Url
+from apiwrappers.auth import TokenAuth
+
+
+class HttpBin:
+    def __init__(self, host, driver):
+        self.url = Url(host)
+        self.driver = driver
+
+    def get(self, params=None):
+        """The request's query parameters."""
+        request = Request(Method.GET, self.url("/get"), query_params=params)
+        return self.driver.fetch(request)
+
+    def post(self, data=None, files=None, json=None):
+        """The request's POST parameters."""
+        request = Request(
+            Method.POST, self.url("/post"), data=data, files=files, json=json
+        )
+        return self.driver.fetch(request)
+
+    def headers(self, headers):
+        """Return the incoming request's HTTP headers."""
+        request = Request(Method.GET, self.url("/headers"), headers=headers)
+        return self.driver.fetch(request)
+
+    def response_headers(self, headers):
+        """Returns a set of response headers from the query string."""
+        request = Request(
+            Method.GET, self.url("/response-headers"), query_params=headers
+        )
+        return self.driver.fetch(request)
+
+    def cookies(self, cookies):
+        """Returns cookie data."""
+        request = Request(Method.GET, self.url("/cookies"), cookies=cookies)
+        return self.driver.fetch(request)
+
+    def set_cookie(self, name, value):
+        """Sets a cookie and redirects to cookie list."""
+        url = self.url("/cookies/set/{name}/{value}", name=name, value=value)
+        request = Request(Method.GET, url)
+        return self.driver.fetch(request)
+
+    def delay(self, delay, timeout):
+        """Returns a delayed response (max of 10 seconds)."""
+        request = Request(Method.GET, self.url("/delay/{delay}", delay=delay))
+        return self.driver.fetch(request, timeout=timeout)
+
+    def html(self):
+        """Returns a simple HTML document."""
+        request = Request(Method.GET, self.url("/html"))
+        return self.driver.fetch(request)
+
+    def basic_auth(self, login, password):
+        """Prompts the user for authorization using HTTP Basic Auth."""
+        url = self.url("/basic-auth/{user}/{passwd}", user=login, passwd=password)
+        request = Request(Method.GET, url, auth=(login, password))
+        return self.driver.fetch(request)
+
+    def bearer_auth(self, token):
+        """Prompts the user for authorization using bearer authentication."""
+        request = Request(Method.GET, self.url("/bearer"), auth=TokenAuth(token))
+        return self.driver.fetch(request)
+
+    def complex_auth_flow(self):
+        """Gets a UUID4 and uses it for bearer authentication."""
+
+        def auth_flow():
+            response = yield Request(Method.GET, self.url("/uuid"))
+            return TokenAuth(response.json()["uuid"])()
+
+        request = Request(Method.GET, self.url("/bearer"), auth=auth_flow)
+        return self.driver.fetch(request)

tests/test_drivers/httpbin_client.pyi

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from typing import Awaitable, Generic, MutableMapping, Optional, TypeVar, overload
+
+from apiwrappers import AsyncDriver, Driver, Response, Url
+from apiwrappers.typedefs import JSON, Data, Files, QueryParams, Timeout
+
+T = TypeVar("T", Driver, AsyncDriver)
+
+class HttpBin(Generic[T]):
+    url: Url
+    driver: T
+    def __init__(self, host: str, driver: T): ...
+    @overload
+    def get(self: HttpBin[Driver], params: Optional[QueryParams] = None) -> Response: ...
+    @overload
+    def get(
+        self: HttpBin[AsyncDriver], params: Optional[QueryParams] = None
+    ) -> Awaitable[Response]: ...
+    @overload
+    def post(
+        self: HttpBin[Driver],
+        data: Optional[Data] = None,
+        files: Optional[Files] = None,
+        json: Optional[JSON] = None,
+    ) -> Response: ...
+    @overload
+    def post(
+        self: HttpBin[AsyncDriver],
+        data: Optional[Data] = None,
+        files: Optional[Files] = None,
+        json: Optional[JSON] = None,
+    ) -> Awaitable[Response]: ...
+    @overload
+    def headers(
+        self: HttpBin[Driver], headers: MutableMapping[str, str]
+    ) -> Response: ...
+    @overload
+    def headers(
+        self: HttpBin[AsyncDriver], headers: MutableMapping[str, str]
+    ) -> Awaitable[Response]: ...
+    @overload
+    def response_headers(
+        self: HttpBin[Driver], headers: MutableMapping[str, str]
+    ) -> Response: ...
+    @overload
+    def response_headers(
+        self: HttpBin[AsyncDriver], headers: MutableMapping[str, str]
+    ) -> Awaitable[Response]: ...
+    @overload
+    def cookies(
+        self: HttpBin[Driver], cookies: MutableMapping[str, str]
+    ) -> Response: ...
+    @overload
+    def cookies(
+        self: HttpBin[AsyncDriver], cookies: MutableMapping[str, str]
+    ) -> Awaitable[Response]: ...
+    @overload
+    def set_cookie(self: HttpBin[Driver], name: str, value: str) -> Response: ...
+    @overload
+    def set_cookie(
+        self: HttpBin[AsyncDriver], name: str, value: str
+    ) -> Awaitable[Response]: ...
+    @overload
+    def delay(self: HttpBin[Driver], delay: int, timeout: Timeout) -> Response: ...
+    @overload
+    def delay(
+        self: HttpBin[AsyncDriver], delay: int, timeout: Timeout
+    ) -> Awaitable[Response]: ...
+    @overload
+    def html(self: HttpBin[Driver]) -> Response: ...
+    @overload
+    def html(self: HttpBin[AsyncDriver]) -> Awaitable[Response]: ...
+    @overload
+    def basic_auth(self: HttpBin[Driver], login: str, password: str) -> Response: ...
+    @overload
+    def basic_auth(
+        self: HttpBin[AsyncDriver], login: str, password: str
+    ) -> Awaitable[Response]: ...
+    @overload
+    def bearer_auth(self: HttpBin[Driver], token: str) -> Response: ...
+    @overload
+    def bearer_auth(self: HttpBin[AsyncDriver], token: str) -> Awaitable[Response]: ...
+    @overload
+    def complex_auth_flow(self: HttpBin[Driver]) -> Response: ...
+    @overload
+    def complex_auth_flow(self: HttpBin[AsyncDriver]) -> Awaitable[Response]: ...

tests/test_drivers/middleware.py

@@ -4,11 +4,11 @@
 
 class RequestMiddleware(BaseMiddleware):
     def process_request(self, request: Request) -> Request:
-        request.headers = {"X-Request-Id": "request_middleware"}
+        request.headers = {"Request": "middleware"}
         return super().process_request(request)
 
 
 class ResponseMiddleware(BaseMiddleware):
     def process_response(self, response: Response) -> Response:
-        response.headers.update({"X-Response-Id": "response_middleware"})
+        response.headers.update({"Response": "middleware"})
         return super().process_response(response)