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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
#!/usr/bin/env python
"""Tests for requests_unixsocket"""
import logging
import pytest
import requests
from requests_unixsocket.testutils import UnixSocketServerThread
import requests_unixsocket
logger = logging.getLogger(__name__)
def test_use_UnixAdapter_directly():
"""Test using UnixAdapter directly, because
https://github.com/httpie/httpie-unixsocket does this
"""
adapter = requests_unixsocket.UnixAdapter()
prepared_request = requests.Request(
method="GET",
url="http+unix://%2Fvar%2Frun%2Fdocker.sock/info",
).prepare()
url = adapter.request_url(request=prepared_request, proxies=None)
assert url == "/info"
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.lower() == url.lower()
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.lower() == url.lower()
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.lower() == url.lower()
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)
|
