Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Sequences of tuples in treq.testing are a bad idea. #260

Open
wsanchez opened this issue Oct 18, 2019 · 1 comment
Open

Sequences of tuples in treq.testing are a bad idea. #260

wsanchez opened this issue Oct 18, 2019 · 1 comment

Comments

@wsanchez
Copy link
Member

Sequences of tuples in treq.testing are a terrible idea. Some classes with names attributes would be more appropriate.

Here's an attempt at that:

#
# Helpers for mocking treq
# treq.testing uses janky tuples for test data. See:
# https://treq.readthedocs.io/en/release-17.8.0/api.html#treq.testing.RequestSequence
#
TreqExpectedRequest = Tuple[
    bytes,                            # method
    str,                              # url
    Dict[bytes, List[bytes]],         # params
    Dict[bytes, List[bytes]],         # headers
    bytes,                            # data
]
TreqCannedResponse = Tuple[
    int,                                     # code
    Dict[bytes, Union[bytes, List[bytes]]],  # headers
    bytes,                                   # body
]
TreqExpectedRequestsAndResponses = Sequence[
    Tuple[TreqExpectedRequest, TreqCannedResponse]
]



@attrs(frozen=True, auto_attribs=True, kw_only=True)
class ExpectedRequest(Exception):
    """
    Expected request.
    """

    method: str
    url: URL
    headers: Headers
    body: bytes


    def asTreqExpectedRequest(self) -> TreqExpectedRequest:
        """
        Return a corresponding TreqExpectedRequest.
        """
        def params() -> Dict[bytes, List[bytes]]:
            params: Dict[bytes, List[bytes]] = {}
            for key, value in self.url.query:
                values = params.setdefault(key, [])
                values.append(value)
            return params

        return (
            self.method.lower().encode("ascii"),
            self.url.replace(query=()).asText(), params(),
            {k: v for k, v in self.headers.getAllRawHeaders()},
            self.body,
        )



@attrs(frozen=True, auto_attribs=True, kw_only=True)
class CannedResponse(Exception):
    """
    Expected response.
    """

    code: int
    headers: Headers
    body: bytes


    def asTreqCannedResponse(self) -> TreqCannedResponse:
        """
        Return a corresponding TreqCannedResponse.
        """
        return (
            self.code,
            {k: v for k, v in self.headers.getAllRawHeaders()},
            self.body,
        )



@attrs(frozen=True, auto_attribs=True)
class ExpectedRequestsAndResponses(Exception):
    """
    Expected request and response sequence.
    """

    requestsAndResponses: Sequence[
        Tuple[ExpectedRequest, CannedResponse]
    ]

    exceptionClass: Type = AssertionError


    def asTreqExpectedRequestsAndResponses(
        self
    ) -> TreqExpectedRequestsAndResponses:
        return tuple(
            (
                request.asTreqExpectedRequest(),
                response.asTreqCannedResponse(),
            )
            for request, response in self.requestsAndResponses
        )


    def _fail(self, error: Any) -> None:
        raise self.exceptionClass(error)


    @contextmanager
    def testing(self) -> Iterator[None]:
        failures: List[Failure] = []

        requestSequence = RequestSequence(
            self.asTreqExpectedRequestsAndResponses(), failures.append
        )
        stubTreq = StubTreq(StringStubbingResource(requestSequence))

        with patch("txdockerhub.v2._client.httpGET", stubTreq.get):
            with requestSequence.consume(self._fail):
                yield

        if failures:
            self._fail(failures)

The patching of get with stubTreq.get is also janky, since you'd have to patch the other methods, to, so there's some thinking let to do there. (Fixing #244 would help, I suspect.)

This allows one to write a test like the example below, where one can read it and maybe understand what the values are, and you don't have to normalize a method name that should be text to lowercase (?!) bytes, etc.

    def test_ping_noToken(self) -> None:
        """
        Ping when a token is not present does not send an Authorization header.
        """
        client = Client()
        expectedRequestsAndResponses = ExpectedRequestsAndResponses(
            (
                (
                    ExpectedRequest(
                        method="GET",
                        url=client._endpoint.api,
                        headers=Headers({
                            "Connection": ["close"],
                            "Accept-Encoding": ["gzip"],
                            "Host": ["registry-1.docker.io"],
                        }),
                        body=b"",
                    ),
                    CannedResponse(
                        code=UNAUTHORIZED,
                        headers=Headers({
                            "WWW-Authenticate": [
                                'Bearer realm="foo",service="bar"'
                            ],
                        }),
                        body=b"",
                    ),
                ),
            ),
            exceptionClass=self.failureException,
        )
        with expectedRequestsAndResponses.testing():
            self.successResultOf(client.ping())
@twm
Copy link
Contributor

twm commented Oct 19, 2019

Indeed, the tuples are bad. I've found wrapping treq.testing in a "builder" class works really well — the builder can make the test pretty succinct and also check that the types are reasonable (i.e. #191).

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

No branches or pull requests

2 participants