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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Tests for requests_unixsocket"""
import logging
import pytest
import requests
import requests_unixsocket
from requests_unixsocket.testutils import UnixSocketServerThread
logger = logging.getLogger(__name__)
def test_unix_domain_adapter_ok():
with UnixSocketServerThread() as usock_thread:
session = requests_unixsocket.Session('http+unix://')
urlencoded_usock = requests.compat.quote_plus(usock_thread.usock)
url = 'http+unix://%s/path/to/page' % urlencoded_usock
for method in ['get', 'post', 'head', 'patch', 'put', 'delete',
'options']:
logger.debug('Calling session.%s(%r) ...', method, url)
r = getattr(session, method)(url)
logger.debug(
'Received response: %r with text: %r and headers: %r',
r, r.text, r.headers)
assert r.status_code == 200
assert r.headers['server'] == 'waitress'
assert r.headers['X-Transport'] == 'unix domain socket'
assert r.headers['X-Requested-Path'] == '/path/to/page'
assert r.headers['X-Socket-Path'] == usock_thread.usock
assert isinstance(r.connection, requests_unixsocket.UnixAdapter)
assert r.url == url
if method == 'head':
assert r.text == ''
else:
assert r.text == 'Hello world!'
def test_unix_domain_adapter_url_with_query_params():
with UnixSocketServerThread() as usock_thread:
session = requests_unixsocket.Session('http+unix://')
urlencoded_usock = requests.compat.quote_plus(usock_thread.usock)
url = ('http+unix://%s'
'/containers/nginx/logs?timestamp=true' % urlencoded_usock)
for method in ['get', 'post', 'head', 'patch', 'put', 'delete',
'options']:
logger.debug('Calling session.%s(%r) ...', method, url)
r = getattr(session, method)(url)
logger.debug(
'Received response: %r with text: %r and headers: %r',
r, r.text, r.headers)
assert r.status_code == 200
assert r.headers['server'] == 'waitress'
assert r.headers['X-Transport'] == 'unix domain socket'
assert r.headers['X-Requested-Path'] == '/containers/nginx/logs'
assert r.headers['X-Requested-Query-String'] == 'timestamp=true'
assert r.headers['X-Socket-Path'] == usock_thread.usock
assert isinstance(r.connection, requests_unixsocket.UnixAdapter)
assert r.url == url
if method == 'head':
assert r.text == ''
else:
assert r.text == 'Hello world!'
def test_unix_domain_adapter_connection_error():
session = requests_unixsocket.Session('http+unix://')
for method in ['get', 'post', 'head', 'patch', 'put', 'delete', 'options']:
with pytest.raises(requests.ConnectionError):
getattr(session, method)(
'http+unix://socket_does_not_exist/path/to/page')
def test_unix_domain_adapter_connection_proxies_error():
session = requests_unixsocket.Session('http+unix://')
for method in ['get', 'post', 'head', 'patch', 'put', 'delete', 'options']:
with pytest.raises(ValueError) as excinfo:
getattr(session, method)(
'http+unix://socket_does_not_exist/path/to/page',
proxies={"http+unix": "http://10.10.1.10:1080"})
assert ('UnixAdapter does not support specifying proxies'
in str(excinfo.value))
def test_unix_domain_adapter_monkeypatch():
with UnixSocketServerThread() as usock_thread:
with requests_unixsocket.monkeypatch('http+unix://'):
urlencoded_usock = requests.compat.quote_plus(usock_thread.usock)
url = 'http+unix://%s/path/to/page' % urlencoded_usock
for method in ['get', 'post', 'head', 'patch', 'put', 'delete',
'options']:
logger.debug('Calling session.%s(%r) ...', method, url)
r = getattr(requests, method)(url)
logger.debug(
'Received response: %r with text: %r and headers: %r',
r, r.text, r.headers)
assert r.status_code == 200
assert r.headers['server'] == 'waitress'
assert r.headers['X-Transport'] == 'unix domain socket'
assert r.headers['X-Requested-Path'] == '/path/to/page'
assert r.headers['X-Socket-Path'] == usock_thread.usock
assert isinstance(r.connection,
requests_unixsocket.UnixAdapter)
assert r.url == url
if method == 'head':
assert r.text == ''
else:
assert r.text == 'Hello world!'
for method in ['get', 'post', 'head', 'patch', 'put', 'delete', 'options']:
with pytest.raises(requests.exceptions.InvalidSchema):
getattr(requests, method)(url)
|
