test_io.py 文件源码

def test_ticket_char_by_char(self):
        # Tests the streaming code for the ticket response.
        ticket_url = "http://ticket.com"
        ticket = {"htsget": {"urls": []}, "padding": "X" * 10}
        returned_response = MockedTicketResponse(
                json.dumps(ticket).encode(), char_by_char=True)
        with mock.patch("requests.get", return_value=returned_response) as mocked_get:
            with tempfile.NamedTemporaryFile("wb+") as f:
                htsget.get(ticket_url, f)
                f.seek(0)
                self.assertEqual(f.read(), b"")
            # Because we have no URLs in the returned ticked, it should be called
            # only once.
            self.assertEqual(mocked_get.call_count, 1)
            # Note that we only get the arguments for the last call using this method.
            args, kwargs = mocked_get.call_args
            self.assertEqual(args[0], ticket_url)
            headers = {}
            self.assertEqual(kwargs["headers"], headers)
            self.assertEqual(kwargs["stream"], True)