import unittest
import unit
class TestUnitRouting(unit.TestUnitApplicationProto):
def setUpClass():
unit.TestUnit().check_modules('python')
def setUp(self):
super().setUp()
self.conf({
"listeners": {
"*:7080": {
"pass": "routes"
}
},
"routes": [{
"match": { "method": "GET" },
"action": { "pass": "applications/empty" }
}],
"applications": {
"empty": {
"type": "python",
"processes": { "spare": 0 },
"path": self.current_dir + '/python/empty',
"working_directory": self.current_dir + '/python/empty',
"module": "wsgi"
},
"mirror": {
"type": "python",
"processes": { "spare": 0 },
"path": self.current_dir + '/python/mirror',
"working_directory": self.current_dir + '/python/mirror',
"module": "wsgi"
}
}
})
def test_routes_match_method_positive(self):
self.assertEqual(self.get()['status'], 200, 'method positive GET')
self.assertEqual(self.post()['status'], 404, 'method positive POST')
def test_routes_match_method_positive_many(self):
self.assertIn('success', self.conf([{
"match": { "method": ["GET", "POST"] },
"action": { "pass": "applications/empty" }
}], 'routes'), 'method positive many configure')
self.assertEqual(self.get()['status'], 200, 'method positive many GET')
self.assertEqual(self.post()['status'], 200,
'method positive many POST')
self.assertEqual(self.delete()['status'], 404,
'method positive many DELETE')
def test_routes_match_method_negative(self):
self.assertIn('success', self.conf([{
"match": { "method": "!GET" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'method negative configure')
self.assertEqual(self.get()['status'], 404, 'method negative GET')
self.assertEqual(self.post()['status'], 200, 'method negative POST')
def test_routes_match_method_negative_many(self):
self.assertIn('success', self.conf([{
"match": { "method": ["!GET", "!POST"] },
"action": { "pass": "applications/empty" }
}], 'routes'), 'method negative many configure')
self.assertEqual(self.get()['status'], 404, 'method negative many GET')
self.assertEqual(self.post()['status'], 404,
'method negative many POST')
self.assertEqual(self.delete()['status'], 200,
'method negative many DELETE')
def test_routes_match_method_wildcard_left(self):
self.assertIn('success', self.conf([{
"match": { "method": "*ET" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'method wildcard left configure')
self.assertEqual(self.get()['status'], 200, 'method wildcard left GET')
self.assertEqual(self.post()['status'], 404,
'method wildcard left POST')
def test_routes_match_method_wildcard_right(self):
self.assertIn('success', self.conf([{
"match": { "method": "GE*" },
"action": { "pass": "applications/empty"}
}], 'routes'), 'method wildcard right configure')
self.assertEqual(self.get()['status'], 200,
'method wildcard right GET')
self.assertEqual(self.post()['status'], 404,
'method wildcard right POST')
def test_routes_match_method_wildcard_left_right(self):
self.assertIn('success', self.conf([{
"match": { "method": "*GET*" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'method wildcard left right configure')
self.assertEqual(self.get()['status'], 200,
'method wildcard right GET')
self.assertEqual(self.post()['status'], 404,
'method wildcard right POST')
def test_routes_match_method_wildcard(self):
self.assertIn('success', self.conf([{
"match": { "method": "*" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'method wildcard configure')
self.assertEqual(self.get()['status'], 200, 'method wildcard')
def test_routes_match_method_case_insensitive(self):
self.assertIn('success', self.conf([{
"match": { "method": "get" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'method case insensitive configure')
self.assertEqual(self.get()['status'], 200, 'method case insensitive')
def test_routes_absent(self):
self.conf({
"listeners": {
"*:7081": {
"pass": "applications/empty"
}
},
"applications": {
"empty": {
"type": "python",
"processes": { "spare": 0 },
"path": self.current_dir + '/python/empty',
"working_directory": self.current_dir + '/python/empty',
"module": "wsgi"
}
}
})
self.assertEqual(self.get(port=7081)['status'], 200, 'routes absent')
def test_routes_pass_invalid(self):
self.assertIn('error', self.conf({ "pass": "routes/blah" },
'listeners/*:7080'), 'routes invalid')
def test_route_empty(self):
self.assertIn('success', self.conf({
"listeners": {
"*:7080": {
"pass": "routes/main"
}
},
"routes": {"main": []},
"applications": {
"empty": {
"type": "python",
"processes": { "spare": 0 },
"path": self.current_dir + '/python/empty',
"working_directory": self.current_dir + '/python/empty',
"module": "wsgi"
},
"mirror": {
"type": "python",
"processes": { "spare": 0 },
"path": self.current_dir + '/python/mirror',
"working_directory": self.current_dir + '/python/mirror',
"module": "wsgi"
}
}
}), 'route empty configure')
self.assertEqual(self.get()['status'], 404, 'route empty')
def test_routes_route_empty(self):
self.assertIn('success', self.conf({}, 'listeners'),
'routes empty listeners configure')
self.assertIn('success', self.conf({}, 'routes'),
'routes empty configure')
def test_routes_route_match_absent(self):
self.assertIn('success', self.conf([{
"action": { "pass": "applications/empty" }
}], 'routes'), 'route match absent configure')
self.assertEqual(self.get()['status'], 200, 'route match absent')
def test_routes_route_action_absent(self):
self.skip_alerts.append(r'failed to apply new conf')
self.assertIn('error', self.conf([{
"match": { "method": "GET" }
}], 'routes'), 'route pass absent configure')
def test_routes_route_pass_absent(self):
self.skip_alerts.append(r'failed to apply new conf')
self.assertIn('error', self.conf([{
"match": { "method": "GET" },
"action": {}
}], 'routes'), 'route pass absent configure')
def test_routes_rules_two(self):
self.assertIn('success', self.conf([{
"match": { "method": "GET" },
"action": { "pass": "applications/empty" }
},
{
"match": { "method": "POST" },
"action": { "pass": "applications/mirror" }
}], 'routes'), 'rules two configure')
self.assertEqual(self.get()['status'], 200, 'rules two match first')
self.assertEqual(self.post(headers={
'Host': 'localhost',
'Content-Type': 'text/html',
'Connection': 'close'
}, body='X')['status'], 200, 'rules two match second')
def test_routes_two(self):
self.assertIn('success', self.conf({
"listeners": {
"*:7080": {
"pass": "routes/first"
}
},
"routes": {
"first": [{
"match": { "method": "GET" },
"action": { "pass": "routes/second" }
}],
"second": [{
"match": { "host": "localhost" },
"action": { "pass": "applications/empty" }
}],
},
"applications": {
"empty": {
"type": "python",
"processes": { "spare": 0 },
"path": self.current_dir + '/python/empty',
"working_directory": self.current_dir + '/python/empty',
"module": "wsgi"
}
}
}), 'routes two configure')
self.assertEqual(self.get()['status'], 200, 'routes two')
def test_routes_match_host_positive(self):
self.assertIn('success', self.conf([{
"match": { "host": "localhost" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match host positive configure')
self.assertEqual(self.get()['status'], 200,
'match host positive localhost')
self.assertEqual(self.get(headers={'Connection': 'close'})['status'],
404, 'match host positive empty')
self.assertEqual(self.get(headers={
'Host': 'localhost.',
'Connection': 'close'
})['status'], 200, 'match host positive trailing dot')
self.assertEqual(self.get(headers={
'Host': 'www.localhost',
'Connection': 'close'
})['status'], 404, 'match host positive www.localhost')
self.assertEqual(self.get(headers={
'Host': 'localhost1',
'Connection': 'close'
})['status'], 404, 'match host positive localhost1')
self.assertEqual(self.get(headers={
'Host': 'example.com',
'Connection': 'close'
})['status'], 404, 'match host positive example.com')
def test_routes_match_host_ipv4(self):
self.assertIn('success', self.conf([{
"match": { "host": "127.0.0.1" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match host ipv4 configure')
self.assertEqual(self.get(headers={
'Host': '127.0.0.1',
'Connection': 'close'
})['status'], 200, 'match host ipv4')
def test_routes_match_host_ipv6(self):
self.assertIn('success', self.conf([{
"match": { "host": "[::1]" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match host ipv6 configure')
self.assertEqual(self.get(headers={
'Host': '[::1]',
'Connection': 'close'
})['status'], 200, 'match host ipv6')
self.assertEqual(self.get(headers={
'Host': '[::1]:7080',
'Connection': 'close'
})['status'], 200, 'match host ipv6 port')
def test_routes_match_host_positive_many(self):
self.assertIn('success', self.conf([{
"match": { "host": ["localhost", "example.com"] },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match host positive many configure')
self.assertEqual(self.get()['status'], 200,
'match host positive many localhost')
self.assertEqual(self.get(headers={
'Host': 'example.com',
'Connection': 'close'
})['status'], 200, 'match host positive many example.com')
def test_routes_match_host_positive_and_negative(self):
self.assertIn('success', self.conf([{
"match": { "host": ["*example.com", "!www.example.com"] },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match host positive and negative configure')
self.assertEqual(self.get()['status'], 404,
'match host positive and negative localhost')
self.assertEqual(self.get(headers={
'Host': 'example.com',
'Connection': 'close'
})['status'], 200, 'match host positive and negative example.com')
self.assertEqual(self.get(headers={
'Host': 'www.example.com',
'Connection': 'close'
})['status'], 404, 'match host positive and negative www.example.com')
self.assertEqual(self.get(headers={
'Host': '!www.example.com',
'Connection': 'close'
})['status'], 200, 'match host positive and negative !www.example.com')
def test_routes_match_host_positive_and_negative_wildcard(self):
self.assertIn('success', self.conf([{
"match": { "host": ["*example*", "!www.example*"] },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match host positive and negative wildcard configure')
self.assertEqual(self.get(headers={
'Host': 'example.com',
'Connection': 'close'
})['status'], 200,
'match host positive and negative wildcard example.com')
self.assertEqual(self.get(headers={
'Host': 'www.example.com',
'Connection': 'close'
})['status'], 404,
'match host positive and negative wildcard www.example.com')
def test_routes_match_host_case_insensitive(self):
self.assertIn('success', self.conf([{
"match": { "host": "Example.com" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'host case insensitive configure')
self.assertEqual(self.get(headers={
'Host': 'example.com',
'Connection': 'close'
})['status'], 200, 'host case insensitive example.com')
self.assertEqual(self.get(headers={
'Host': 'EXAMPLE.COM',
'Connection': 'close'
})['status'], 200, 'host case insensitive EXAMPLE.COM')
def test_routes_match_host_port(self):
self.assertIn('success', self.conf([{
"match": { "host": "example.com" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match host port configure')
self.assertEqual(self.get(headers={
'Host': 'example.com:7080',
'Connection': 'close'
})['status'], 200, 'match host port')
def test_routes_match_uri_positive(self):
self.assertIn('success', self.conf([{
"match": { "uri": "/" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match uri positive configure')
self.assertEqual(self.get()['status'], 200, 'match uri positive')
self.assertEqual(self.get(url='/blah')['status'], 404,
'match uri positive blah')
self.assertEqual(self.get(url='/#blah')['status'], 200,
'match uri positive #blah')
self.assertEqual(self.get(url='/?var')['status'], 200,
'match uri params')
self.assertEqual(self.get(url='//')['status'], 200,
'match uri adjacent slashes')
self.assertEqual(self.get(url='/blah/../')['status'], 200,
'match uri relative path')
self.assertEqual(self.get(url='/./')['status'], 200,
'match uri relative path')
def test_routes_match_uri_case_sensitive(self):
self.assertIn('success', self.conf([{
"match": { "uri": "/BLAH" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match uri case sensitive configure')
self.assertEqual(self.get(url='/blah')['status'], 404,
'match uri case sensitive blah')
self.assertEqual(self.get(url='/BlaH')['status'], 404,
'match uri case sensitive BlaH')
self.assertEqual(self.get(url='/BLAH')['status'], 200,
'match uri case sensitive BLAH')
def test_routes_match_uri_normalize(self):
self.assertIn('success', self.conf([{
"match": { "uri": "/blah" },
"action": { "pass": "applications/empty" }
}], 'routes'), 'match uri normalize configure')
self.assertEqual(self.get(url='/%62%6c%61%68')['status'], 200,
'match uri normalize')
def test_routes_match_rules(self):
self.assertIn('success', self.conf([{
"match": {
"method": "GET",
"host": "localhost",
"uri": "/"
},
"action": { "pass": "applications/empty" }
}], 'routes'), 'routes match rules configure')
self.assertEqual(self.get()['status'], 200, 'routes match rules')
def test_routes_loop(self):
self.assertIn('success', self.conf([{
"match": { "uri": "/" },
"action": { "pass": "routes" }
}], 'routes'), 'routes loop configure')
self.assertEqual(self.get()['status'], 500, 'routes loop')
if __name__ == '__main__':
TestUnitRouting.main()
