1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Path hack.
import sys, os
sys.path.insert(0, os.path.abspath('..'))
import sys
import unittest
import select
has_poll = hasattr(select, "poll")
from requests import async
sys.path.append('.')
from test_requests import httpbin, RequestsTestSuite, SERVICES
class RequestsTestSuiteUsingAsyncApi(RequestsTestSuite):
"""Requests async test cases."""
def patched(f):
"""Automatically send request after creation."""
def wrapped(*args, **kwargs):
request = f(*args, **kwargs)
return async.map([request])[0]
return wrapped
# Patched requests.api functions.
global request
request = patched(async.request)
global delete, get, head, options, patch, post, put
delete = patched(async.delete)
get = patched(async.get)
head = patched(async.head)
options = patched(async.options)
patch = patched(async.patch)
post = patched(async.post)
put = patched(async.put)
def test_entry_points(self):
async.request
async.delete
async.get
async.head
async.options
async.patch
async.post
async.put
async.map
async.send
def test_select_poll(self):
"""Test to make sure we don't overwrite the poll"""
self.assertEqual(hasattr(select, "poll"), has_poll)
if __name__ == '__main__':
unittest.main()
|