Function testRecordAndReplay

tests/Connection.py:79–116

(replaying_connection_class, protocol, response_body, expected_recording)

)
@responses.activate
def testRecordAndReplay(replaying_connection_class, protocol, response_body, expected_recording):
    file = StringIO()
    host = "api.github.com"
    verb = "GET"
    url = "/user"
    headers = {"Authorization": "Basic p4ssw0rd", "User-Agent": "PyGithub/Python"}

    response = Mock()
    response.status = 200
    response.getheaders.return_value = {}
    response.read.return_value = response_body

    connection = Mock()
    connection.getresponse.return_value = response

    # write mock response to buffer
    rdf = Framework.ReplayDataFile("string", file)
    RecordingMockConnection.setOpenFile(lambda mode: rdf)
    recording_connection = RecordingMockConnection(protocol, host, None, lambda *args, **kwds: connection)
    recording_connection.request(verb, url, None, headers)
    recording_connection.getresponse()
    recording_connection.close()

    # validate contents of buffer
    file_value_lines = file.getvalue().split("\n")
    expected_recording_lines = (protocol + expected_recording).split("\n")
    assert file_value_lines[:5] == expected_recording_lines[:5]
    assert eval(file_value_lines[5]) == eval(expected_recording_lines[5])
    # dict literal, so keys not in guaranteed order
    assert file_value_lines[6:] == expected_recording_lines[6:]

    # rewind buffer and attempt to replay response from it
    file.seek(0)
    rdf = Framework.ReplayDataFile("string", file)
    replaying_connection_class.setOpenFile(lambda mode: rdf)
    replaying_connection = replaying_connection_class(host=host, port=None)
    replaying_connection.request(verb, url, None, headers)
    replaying_connection.getresponse()