(self, verb, url, input, headers)
| 319 | return f"Replay data mismatch in {self.__file}" |
| 320 | |
| 321 | def __readNextRequest(self, verb, url, input, headers) -> Request: |
| 322 | fixAuthorizationHeader(headers) |
| 323 | request = Request(self.__protocol, verb, self.__host, self.__port, url, headers, input) |
| 324 | assert request.protocol == self.__file.readline(), self.__replayDataMismatchLine() |
| 325 | assert request.verb == self.__file.readline(), self.__replayDataMismatchLine() |
| 326 | assert request.host == self.__file.readline(), self.__replayDataMismatchLine() |
| 327 | assert str(request.port) == self.__file.readline(), self.__replayDataMismatchLine() |
| 328 | assert self.__splitUrl(request.url) == self.__splitUrl(self.__file.readline()), self.__replayDataMismatchLine() |
| 329 | assert request.request_headers == eval(self.__file.readline()), self.__replayDataMismatchLine() |
| 330 | expectedInput = self.__file.readline() |
| 331 | if isinstance(input, str): |
| 332 | trInput = input.replace("\n", "").replace("\r", "") |
| 333 | if input.startswith("{"): |
| 334 | assert expectedInput.startswith("{"), self.__replayDataMismatchLine() |
| 335 | assert json.loads(trInput) == json.loads(expectedInput), self.__replayDataMismatchLine() |
| 336 | else: |
| 337 | assert trInput == expectedInput, self.__replayDataMismatchLine() |
| 338 | else: |
| 339 | # for non-string input (e.g. upload asset), let it pass. |
| 340 | pass |
| 341 | |
| 342 | return request |
| 343 | |
| 344 | def __splitUrl(self, url): |
| 345 | splitedUrl = url.split("?") |
no test coverage detected