Tests: routing.

1 files changed, 458 insertions, 0 deletions

diff --git a/test/test_routing.py b/test/test_routing.py
new file mode 100644
index 00000000..07097fc8
--- /dev/null
+++ b/test/test_routing.py
@@ -0,0 +1,458 @@
+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()