mirror of
https://github.com/django/daphne.git
synced 2024-09-21 11:08:47 +03:00
118 lines
3.2 KiB
Python
118 lines
3.2 KiB
Python
# coding: utf8
|
|
|
|
from unittest import TestCase
|
|
|
|
from twisted.web.http_headers import Headers
|
|
|
|
from daphne.utils import parse_x_forwarded_for
|
|
|
|
|
|
class TestXForwardedForHttpParsing(TestCase):
|
|
"""
|
|
Tests that the parse_x_forwarded_for util correctly parses twisted Header.
|
|
"""
|
|
|
|
def test_basic(self):
|
|
headers = Headers({
|
|
b"X-Forwarded-For": [b"10.1.2.3"],
|
|
b"X-Forwarded-Port": [b"1234"],
|
|
b"X-Forwarded-Proto": [b"https"]
|
|
})
|
|
result = parse_x_forwarded_for(headers)
|
|
self.assertEqual(result, (["10.1.2.3", 1234], "https"))
|
|
self.assertIsInstance(result[0][0], str)
|
|
self.assertIsInstance(result[1], str)
|
|
|
|
def test_address_only(self):
|
|
headers = Headers({
|
|
b"X-Forwarded-For": [b"10.1.2.3"],
|
|
})
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers),
|
|
(["10.1.2.3", 0], None)
|
|
)
|
|
|
|
def test_v6_address(self):
|
|
headers = Headers({
|
|
b"X-Forwarded-For": [b"1043::a321:0001, 10.0.5.6"],
|
|
})
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers),
|
|
(["1043::a321:0001", 0], None)
|
|
)
|
|
|
|
def test_multiple_proxys(self):
|
|
headers = Headers({
|
|
b"X-Forwarded-For": [b"10.1.2.3, 10.1.2.4"],
|
|
})
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers),
|
|
(["10.1.2.3", 0], None)
|
|
)
|
|
|
|
def test_original(self):
|
|
headers = Headers({})
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers, original_addr=["127.0.0.1", 80]),
|
|
(["127.0.0.1", 80], None)
|
|
)
|
|
|
|
def test_no_original(self):
|
|
headers = Headers({})
|
|
self.assertEqual(parse_x_forwarded_for(headers), (None, None))
|
|
|
|
|
|
class TestXForwardedForWsParsing(TestCase):
|
|
"""
|
|
Tests that the parse_x_forwarded_for util correctly parses dict headers.
|
|
"""
|
|
|
|
def test_basic(self):
|
|
headers = {
|
|
b"X-Forwarded-For": b"10.1.2.3",
|
|
b"X-Forwarded-Port": b"1234",
|
|
b"X-Forwarded-Proto": b"https",
|
|
}
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers),
|
|
(["10.1.2.3", 1234], "https")
|
|
)
|
|
|
|
def test_address_only(self):
|
|
headers = {
|
|
b"X-Forwarded-For": b"10.1.2.3",
|
|
}
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers),
|
|
(["10.1.2.3", 0], None)
|
|
)
|
|
|
|
def test_v6_address(self):
|
|
headers = {
|
|
b"X-Forwarded-For": [b"1043::a321:0001, 10.0.5.6"],
|
|
}
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers),
|
|
(["1043::a321:0001", 0], None)
|
|
)
|
|
|
|
def test_multiple_proxies(self):
|
|
headers = {
|
|
b"X-Forwarded-For": b"10.1.2.3, 10.1.2.4",
|
|
}
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers),
|
|
(["10.1.2.3", 0], None)
|
|
)
|
|
|
|
def test_original(self):
|
|
headers = {}
|
|
self.assertEqual(
|
|
parse_x_forwarded_for(headers, original_addr=["127.0.0.1", 80]),
|
|
(["127.0.0.1", 80], None)
|
|
)
|
|
|
|
def test_no_original(self):
|
|
headers = {}
|
|
self.assertEqual(parse_x_forwarded_for(headers), (None, None))
|