diff options
Diffstat (limited to '')
| -rw-r--r-- | test/test_client_ip.py | 27 | ||||
| -rw-r--r-- | test/test_routing.py | 14 |
2 files changed, 37 insertions, 4 deletions
diff --git a/test/test_client_ip.py b/test/test_client_ip.py index 24436351..50aa6afc 100644 --- a/test/test_client_ip.py +++ b/test/test_client_ip.py @@ -1,4 +1,5 @@ from unit.applications.lang.python import TestApplicationPython +from unit.option import option class TestClientIP(TestApplicationPython): @@ -15,15 +16,27 @@ class TestClientIP(TestApplicationPython): "client_ip": options, "pass": "applications/client_ip", }, + "unix:" + + option.temp_dir + + "/sock": { + "client_ip": options, + "pass": "applications/client_ip", + }, }, 'listeners', ), 'listeners configure' def get_xff(self, xff, sock_type='ipv4'): - port = 7081 if sock_type == 'ipv4' else 7082 + address = { + 'ipv4': ('127.0.0.1', 7081), + 'ipv6': ('::1', 7082), + 'unix': (option.temp_dir + '/sock', None), + } + (addr, port) = address[sock_type] return self.get( sock_type=sock_type, + addr=addr, port=port, headers={'Connection': 'close', 'X-Forwarded-For': xff}, )['body'] @@ -85,6 +98,18 @@ class TestClientIP(TestApplicationPython): ]: assert self.get_xff(ip, 'ipv6') == ip, 'replace' + def test_client_ip_unix(self, temp_dir): + self.client_ip({'header': 'X-Forwarded-For', 'source': 'unix'}) + + assert self.get_xff('1.1.1.1') == '127.0.0.1', 'bad source ipv4' + assert self.get_xff('1.1.1.1', 'ipv6') == '::1', 'bad source ipv6' + + for ip in [ + '1.1.1.1', + '::11.22.33.44', + ]: + assert self.get_xff(ip, 'unix') == ip, 'replace' + def test_client_ip_recursive(self): self.client_ip( { diff --git a/test/test_routing.py b/test/test_routing.py index fda429a4..3649b37c 100644 --- a/test/test_routing.py +++ b/test/test_routing.py @@ -1723,18 +1723,26 @@ class TestRouting(TestApplicationPython): addr = temp_dir + '/sock' assert 'success' in self.conf( - {"unix:" + addr: {"pass": "routes"}}, 'listeners' + { + "127.0.0.1:7081": {"pass": "routes"}, + "unix:" + addr: {"pass": "routes"}, + }, + 'listeners', ), 'source listeners configure' self.route_match({"source": "!0.0.0.0/0"}) assert ( self.get(sock_type='unix', addr=addr)['status'] == 200 - ), 'unix ipv4' + ), 'unix ipv4 neg' self.route_match({"source": "!::/0"}) assert ( self.get(sock_type='unix', addr=addr)['status'] == 200 - ), 'unix ipv6' + ), 'unix ipv6 neg' + + self.route_match({"source": "unix"}) + assert self.get(port=7081)['status'] == 404, 'unix ipv4' + assert self.get(sock_type='unix', addr=addr)['status'] == 200, 'unix' def test_routes_match_source(self): self.route_match({"source": "::"}) |