test_slot.py 文件源码

def test__send_keep_alive(slot):
    with patch('threading.Timer') as mock_timer:
        # No matter what we schedule keep alive thread. Sometimes we log an error
        db_error = psycopg2.DatabaseError()
        db_error.message = 'Log ME!'
        assert_stuff_about_keep_alive(slot, mock_timer, db_error, True)

        db_error = psycopg2.DatabaseError()
        db_error.message = 'no COPY in progress\n'
        assert_stuff_about_keep_alive(slot, mock_timer, db_error, False)

        error = Exception()
        error.message = 'no COPY in progress\n'
        assert_stuff_about_keep_alive(slot, mock_timer, error, True)

        # Happy Path
        assert_stuff_about_keep_alive(slot, mock_timer, None, False)