authentik.enterprise.providers.google_workspace.clients.test_http 

 1from json import dumps
 2
 3from httplib2 import Response
 4
 5
 6class MockHTTP:
 7
 8    _recorded_requests = []
 9    _responses = {}
10
11    def __init__(
12        self,
13        raise_on_unrecorded=True,
14    ) -> None:
15        self._recorded_requests = []
16        self._responses = {}
17        self.raise_on_unrecorded = raise_on_unrecorded
18
19    def add_response(self, uri: str, body: str | dict = "", meta: dict | None = None, method="GET"):
20        if isinstance(body, dict):
21            body = dumps(body)
22        self._responses[(uri, method.upper())] = (body, meta or {"status": "200"})
23
24    def requests(self):
25        return self._recorded_requests
26
27    def request(
28        self,
29        uri,
30        method="GET",
31        body=None,
32        headers=None,
33        redirections=1,
34        connection_type=None,
35    ):
36        key = (uri, method.upper())
37        self._recorded_requests.append((uri, method, body, headers))
38        if key not in self._responses and self.raise_on_unrecorded:
39            raise AssertionError(key)
40        body, meta = self._responses[key]
41        return Response(meta), body.encode("utf-8")
class MockHTTP: 
 7class MockHTTP:
 8
 9    _recorded_requests = []
10    _responses = {}
11
12    def __init__(
13        self,
14        raise_on_unrecorded=True,
15    ) -> None:
16        self._recorded_requests = []
17        self._responses = {}
18        self.raise_on_unrecorded = raise_on_unrecorded
19
20    def add_response(self, uri: str, body: str | dict = "", meta: dict | None = None, method="GET"):
21        if isinstance(body, dict):
22            body = dumps(body)
23        self._responses[(uri, method.upper())] = (body, meta or {"status": "200"})
24
25    def requests(self):
26        return self._recorded_requests
27
28    def request(
29        self,
30        uri,
31        method="GET",
32        body=None,
33        headers=None,
34        redirections=1,
35        connection_type=None,
36    ):
37        key = (uri, method.upper())
38        self._recorded_requests.append((uri, method, body, headers))
39        if key not in self._responses and self.raise_on_unrecorded:
40            raise AssertionError(key)
41        body, meta = self._responses[key]
42        return Response(meta), body.encode("utf-8")
MockHTTP(raise_on_unrecorded=True) 
12    def __init__(
13        self,
14        raise_on_unrecorded=True,
15    ) -> None:
16        self._recorded_requests = []
17        self._responses = {}
18        self.raise_on_unrecorded = raise_on_unrecorded
raise_on_unrecorded
def add_response( self, uri: str, body: str | dict = '', meta: dict | None = None, method='GET'): 
20    def add_response(self, uri: str, body: str | dict = "", meta: dict | None = None, method="GET"):
21        if isinstance(body, dict):
22            body = dumps(body)
23        self._responses[(uri, method.upper())] = (body, meta or {"status": "200"})
def requests(self): 
25    def requests(self):
26        return self._recorded_requests
def request( self, uri, method='GET', body=None, headers=None, redirections=1, connection_type=None): 
28    def request(
29        self,
30        uri,
31        method="GET",
32        body=None,
33        headers=None,
34        redirections=1,
35        connection_type=None,
36    ):
37        key = (uri, method.upper())
38        self._recorded_requests.append((uri, method, body, headers))
39        if key not in self._responses and self.raise_on_unrecorded:
40            raise AssertionError(key)
41        body, meta = self._responses[key]
42        return Response(meta), body.encode("utf-8")