xref: /unit/test/test_routing.py (revision 2073:bc6ad31ce286)

# -*- coding: utf-8 -*-
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option


class TestRouting(TestApplicationPython):
    prerequisites = {'modules': {'python': 'any'}}

    def setup_method(self):
        assert 'success' in self.conf(
            {
                "listeners": {"*:7080": {"pass": "routes"}},
                "routes": [
                    {
                        "match": {"method": "GET"},
                        "action": {"return": 200},
                    }
                ],
                "applications": {},
            }
        ), 'routing configure'

    def route(self, route):
        return self.conf([route], 'routes')

    def route_match(self, match):
        assert 'success' in self.route(
            {"match": match, "action": {"return": 200}}
        ), 'route match configure'

    def route_match_invalid(self, match):
        assert 'error' in self.route(
            {"match": match, "action": {"return": 200}}
        ), 'route match configure invalid'

    def host(self, host, status):
        assert (
            self.get(headers={'Host': host, 'Connection': 'close'})['status']
            == status
        ), 'match host'

    def cookie(self, cookie, status):
        assert (
            self.get(
                headers={
                    'Host': 'localhost',
                    'Cookie': cookie,
                    'Connection': 'close',
                },
            )['status']
            == status
        ), 'match cookie'

    def test_routes_match_method_positive(self):
        assert self.get()['status'] == 200, 'GET'
        assert self.post()['status'] == 404, 'POST'

    def test_routes_match_method_positive_many(self):
        self.route_match({"method": ["GET", "POST"]})

        assert self.get()['status'] == 200, 'GET'
        assert self.post()['status'] == 200, 'POST'
        assert self.delete()['status'] == 404, 'DELETE'

    def test_routes_match_method_negative(self):
        self.route_match({"method": "!GET"})

        assert self.get()['status'] == 404, 'GET'
        assert self.post()['status'] == 200, 'POST'

    def test_routes_match_method_negative_many(self):
        self.route_match({"method": ["!GET", "!POST"]})

        assert self.get()['status'] == 404, 'GET'
        assert self.post()['status'] == 404, 'POST'
        assert self.delete()['status'] == 200, 'DELETE'

    def test_routes_match_method_wildcard_left(self):
        self.route_match({"method": "*ET"})

        assert self.get()['status'] == 200, 'GET'
        assert self.post()['status'] == 404, 'POST'

    def test_routes_match_method_wildcard_right(self):
        self.route_match({"method": "GE*"})

        assert self.get()['status'] == 200, 'GET'
        assert self.post()['status'] == 404, 'POST'

    def test_routes_match_method_wildcard_left_right(self):
        self.route_match({"method": "*GET*"})

        assert self.get()['status'] == 200, 'GET'
        assert self.post()['status'] == 404, 'POST'

    def test_routes_match_method_wildcard(self):
        self.route_match({"method": "*"})

        assert self.get()['status'] == 200, 'GET'

    def test_routes_match_invalid(self):
        self.route_match_invalid({"method": "**"})

    def test_routes_match_valid(self):
        self.route_match({"method": "blah*"})
        self.route_match({"host": "*blah*blah"})
        self.route_match({"host": "blah*blah*blah"})
        self.route_match({"host": "blah*blah*"})

    def test_routes_match_empty_exact(self):
        self.route_match({"uri": ""})
        assert self.get()['status'] == 404

        self.route_match({"uri": "/"})
        assert self.get()['status'] == 200
        assert self.get(url='/blah')['status'] == 404

    def test_routes_match_negative(self):
        self.route_match({"uri": "!"})
        assert self.get()['status'] == 200

        self.route_match({"uri": "!*"})
        assert self.get()['status'] == 404

        self.route_match({"uri": "!/"})
        assert self.get()['status'] == 404
        assert self.get(url='/blah')['status'] == 200

        self.route_match({"uri": "!*blah"})
        assert self.get()['status'] == 200
        assert self.get(url='/bla')['status'] == 200
        assert self.get(url='/blah')['status'] == 404
        assert self.get(url='/blah1')['status'] == 200

        self.route_match({"uri": "!/blah*1*"})
        assert self.get()['status'] == 200
        assert self.get(url='/blah')['status'] == 200
        assert self.get(url='/blah1')['status'] == 404
        assert self.get(url='/blah12')['status'] == 404
        assert self.get(url='/blah2')['status'] == 200

    def test_routes_match_wildcard_middle(self):
        self.route_match({"host": "ex*le"})

        self.host('example', 200)
        self.host('www.example', 404)
        self.host('example.com', 404)
        self.host('exampl', 404)

    def test_routes_match_method_case_insensitive(self):
        self.route_match({"method": "get"})

        assert self.get()['status'] == 200, 'GET'

    def test_routes_match_wildcard_left_case_insensitive(self):
        self.route_match({"method": "*get"})
        assert self.get()['status'] == 200, 'GET'

        self.route_match({"method": "*et"})
        assert self.get()['status'] == 200, 'GET'

    def test_routes_match_wildcard_middle_case_insensitive(self):
        self.route_match({"method": "g*t"})

        assert self.get()['status'] == 200, 'GET'

    def test_routes_match_wildcard_right_case_insensitive(self):
        self.route_match({"method": "get*"})
        assert self.get()['status'] == 200, 'GET'

        self.route_match({"method": "ge*"})
        assert self.get()['status'] == 200, 'GET'

    def test_routes_match_wildcard_substring_case_insensitive(self):
        self.route_match({"method": "*et*"})

        assert self.get()['status'] == 200, 'GET'

    def test_routes_match_wildcard_left_case_sensitive(self):
        self.route_match({"uri": "*blah"})

        assert self.get(url='/blah')['status'] == 200, '/blah'
        assert self.get(url='/BLAH')['status'] == 404, '/BLAH'

    def test_routes_match_wildcard_middle_case_sensitive(self):
        self.route_match({"uri": "/b*h"})

        assert self.get(url='/blah')['status'] == 200, '/blah'
        assert self.get(url='/BLAH')['status'] == 404, '/BLAH'

    def test_route_match_wildcards_ordered(self):
        self.route_match({"uri": "/a*x*y*"})

        assert self.get(url='/axy')['status'] == 200, '/axy'
        assert self.get(url='/ayx')['status'] == 404, '/ayx'

    def test_route_match_wildcards_adjust_start(self):
        self.route_match({"uri": "/bla*bla*"})

        assert self.get(url='/bla_foo')['status'] == 404, '/bla_foo'

    def test_route_match_wildcards_adjust_start_substr(self):
        self.route_match({"uri": "*bla*bla*"})

        assert self.get(url='/bla_foo')['status'] == 404, '/bla_foo'

    def test_route_match_wildcards_adjust_end(self):
        self.route_match({"uri": "/bla*bla"})

        assert self.get(url='/foo_bla')['status'] == 404, '/foo_bla'

    def test_routes_match_wildcard_right_case_sensitive(self):
        self.route_match({"uri": "/bla*"})

        assert self.get(url='/blah')['status'] == 200, '/blah'
        assert self.get(url='/BLAH')['status'] == 404, '/BLAH'

    def test_routes_match_wildcard_substring_case_sensitive(self):
        self.route_match({"uri": "*bla*"})

        assert self.get(url='/blah')['status'] == 200, '/blah'
        assert self.get(url='/BLAH')['status'] == 404, '/BLAH'

    def test_routes_match_many_wildcard_substrings_case_sensitive(self):
        self.route_match({"uri": "*a*B*c*"})

        assert self.get(url='/blah-a-B-c-blah')['status'] == 200
        assert self.get(url='/a-B-c')['status'] == 200
        assert self.get(url='/aBc')['status'] == 200
        assert self.get(url='/aBCaBbc')['status'] == 200
        assert self.get(url='/ABc')['status'] == 404

    def test_routes_empty_regex(self):
        if not option.available['modules']['regex']:
            pytest.skip('requires regex')

        self.route_match({"uri": "~"})
        assert self.get(url='/')['status'] == 200, 'empty regexp'
        assert self.get(url='/anything')['status'] == 200, '/anything'

        self.route_match({"uri": "!~"})
        assert self.get(url='/')['status'] == 404, 'empty regexp 2'
        assert self.get(url='/nothing')['status'] == 404, '/nothing'

    def test_routes_bad_regex(self):
        if not option.available['modules']['regex']:
            pytest.skip('requires regex')

        assert 'error' in self.route(
            {"match": {"uri": "~/bl[ah"}, "action": {"return": 200}}
        ), 'bad regex'

        status = self.route(
            {"match": {"uri": "~(?R)?z"}, "action": {"return": 200}}
        )
        if 'error' not in status:
            assert self.get(url='/nothing_z')['status'] == 500, '/nothing_z'

        status = self.route(
            {"match": {"uri": "~((?1)?z)"}, "action": {"return": 200}}
        )
        if 'error' not in status:
            assert self.get(url='/nothing_z')['status'] == 500, '/nothing_z'

    def test_routes_match_regex_case_sensitive(self):
        if not option.available['modules']['regex']:
            pytest.skip('requires regex')

        self.route_match({"uri": "~/bl[ah]"})

        assert self.get(url='/rlah')['status'] == 404, '/rlah'
        assert self.get(url='/blah')['status'] == 200, '/blah'
        assert self.get(url='/blh')['status'] == 200, '/blh'
        assert self.get(url='/BLAH')['status'] == 404, '/BLAH'

    def test_routes_match_regex_negative_case_sensitive(self):
        if not option.available['modules']['regex']:
            pytest.skip('requires regex')

        self.route_match({"uri": "!~/bl[ah]"})

        assert self.get(url='/rlah')['status'] == 200, '/rlah'
        assert self.get(url='/blah')['status'] == 404, '/blah'
        assert self.get(url='/blh')['status'] == 404, '/blh'
        assert self.get(url='/BLAH')['status'] == 200, '/BLAH'

    def test_routes_pass_encode(self):
        def check_pass(path, name):
            assert 'success' in self.conf(
                {
                    "listeners": {"*:7080": {"pass": "applications/" + path}},
                    "applications": {
                        name: {
                            "type": self.get_application_type(),
                            "processes": {"spare": 0},
                            "path": option.test_dir + '/python/empty',
                            "working_directory": option.test_dir
                            + '/python/empty',
                            "module": "wsgi",
                        }
                    },
                }
            )

            assert self.get()['status'] == 200

        check_pass("%25", "%")
        check_pass("blah%2Fblah", "blah/blah")
        check_pass("%2Fblah%2F%2Fblah%2F", "/blah//blah/")
        check_pass("%20blah%252Fblah%7E", " blah%2Fblah~")

        def check_pass_error(path, name):
            assert 'error' in self.conf(
                {
                    "listeners": {"*:7080": {"pass": "applications/" + path}},
                    "applications": {
                        name: {
                            "type": self.get_application_type(),
                            "processes": {"spare": 0},
                            "path": option.test_dir + '/python/empty',
                            "working_directory": option.test_dir
                            + '/python/empty',
                            "module": "wsgi",
                        }
                    },
                }
            )

        check_pass_error("%", "%")
        check_pass_error("%1", "%1")

    def test_routes_absent(self):
        assert 'success' in self.conf(
            {
                "listeners": {"*:7081": {"pass": "applications/empty"}},
                "applications": {
                    "empty": {
                        "type": self.get_application_type(),
                        "processes": {"spare": 0},
                        "path": option.test_dir + '/python/empty',
                        "working_directory": option.test_dir + '/python/empty',
                        "module": "wsgi",
                    }
                },
            }
        )

        assert self.get(port=7081)['status'] == 200, 'routes absent'

    def test_routes_pass_invalid(self):
        assert 'error' in self.conf(
            {"pass": "routes/blah"}, 'listeners/*:7080'
        ), 'routes invalid'

    def test_route_empty(self):
        assert 'success' in self.conf(
            {
                "listeners": {"*:7080": {"pass": "routes/main"}},
                "routes": {"main": []},
                "applications": {},
            }
        ), 'route empty configure'

        assert self.get()['status'] == 404, 'route empty'

    def test_routes_route_empty(self):
        assert 'success' in self.conf(
            {}, 'listeners'
        ), 'routes empty listeners configure'

        assert 'success' in self.conf({}, 'routes'), 'routes empty configure'

    def test_routes_route_match_absent(self):
        assert 'success' in self.conf(
            [{"action": {"return": 200}}], 'routes'
        ), 'route match absent configure'

        assert self.get()['status'] == 200, 'route match absent'

    def test_routes_route_action_absent(self, skip_alert):
        skip_alert(r'failed to apply new conf')

        assert 'error' in self.conf(
            [{"match": {"method": "GET"}}], 'routes'
        ), 'route pass absent configure'

    def test_routes_route_pass(self):
        assert 'success' in self.conf(
            {
                "applications": {
                    "app": {
                        "type": self.get_application_type(),
                        "processes": {"spare": 0},
                        "path": "/app",
                        "module": "wsgi",
                    }
                },
                "upstreams": {
                    "one": {
                        "servers": {
                            "127.0.0.1:7081": {},
                            "127.0.0.1:7082": {},
                        },
                    },
                    "two": {
                        "servers": {
                            "127.0.0.1:7081": {},
                            "127.0.0.1:7082": {},
                        },
                    },
                },
            }
        )

        assert 'success' in self.conf(
            [{"action": {"pass": "routes"}}], 'routes'
        )
        assert 'success' in self.conf(
            [{"action": {"pass": "applications/app"}}], 'routes'
        )
        assert 'success' in self.conf(
            [{"action": {"pass": "upstreams/one"}}], 'routes'
        )

    def test_routes_route_pass_absent(self):
        assert 'error' in self.conf(
            [{"match": {"method": "GET"}, "action": {}}], 'routes'
        ), 'route pass absent configure'

    def test_routes_route_pass_invalid(self):
        assert 'success' in self.conf(
            {
                "applications": {
                    "app": {
                        "type": self.get_application_type(),
                        "processes": {"spare": 0},
                        "path": "/app",
                        "module": "wsgi",
                    }
                },
                "upstreams": {
                    "one": {
                        "servers": {
                            "127.0.0.1:7081": {},
                            "127.0.0.1:7082": {},
                        },
                    },
                    "two": {
                        "servers": {
                            "127.0.0.1:7081": {},
                            "127.0.0.1:7082": {},
                        },
                    },
                },
            }
        )

        assert 'error' in self.conf(
            [{"action": {"pass": "blah"}}], 'routes'
        ), 'route pass invalid'
        assert 'error' in self.conf(
            [{"action": {"pass": "routes/blah"}}], 'routes'
        ), 'route pass routes invalid'
        assert 'error' in self.conf(
            [{"action": {"pass": "applications/blah"}}], 'routes'
        ), 'route pass applications invalid'
        assert 'error' in self.conf(
            [{"action": {"pass": "upstreams/blah"}}], 'routes'
        ), 'route pass upstreams invalid'

    def test_routes_action_unique(self, temp_dir):
        assert 'success' in self.conf(
            {
                "listeners": {
                    "*:7080": {"pass": "routes"},
                    "*:7081": {"pass": "applications/app"},
                },
                "routes": [{"action": {"proxy": "http://127.0.0.1:7081"}}],
                "applications": {
                    "app": {
                        "type": self.get_application_type(),
                        "processes": {"spare": 0},
                        "path": "/app",
                        "module": "wsgi",
                    }
                },
            }
        )

        assert 'error' in self.conf(
            {"proxy": "http://127.0.0.1:7081", "share": temp_dir},
            'routes/0/action',
        ), 'proxy share'
        assert 'error' in self.conf(
            {
                "proxy": "http://127.0.0.1:7081",
                "pass": "applications/app",
            },
            'routes/0/action',
        ), 'proxy pass'
        assert 'error' in self.conf(
            {"share": temp_dir, "pass": "applications/app"},
            'routes/0/action',
        ), 'share pass'

    def test_routes_rules_two(self):
        assert 'success' in self.conf(
            [
                {"match": {"method": "GET"}, "action": {"return": 200}},
                {"match": {"method": "POST"}, "action": {"return": 201}},
            ],
            'routes',
        ), 'rules two configure'

        assert self.get()['status'] == 200, 'rules two match first'
        assert self.post()['status'] == 201, 'rules two match second'

    def test_routes_two(self):
        assert 'success' in self.conf(
            {
                "listeners": {"*:7080": {"pass": "routes/first"}},
                "routes": {
                    "first": [
                        {
                            "match": {"method": "GET"},
                            "action": {"pass": "routes/second"},
                        }
                    ],
                    "second": [
                        {
                            "match": {"host": "localhost"},
                            "action": {"return": 200},
                        }
                    ],
                },
                "applications": {},
            }
        ), 'routes two configure'

        assert self.get()['status'] == 200, 'routes two'

    def test_routes_match_host_positive(self):
        self.route_match({"host": "localhost"})

        assert self.get()['status'] == 200, 'localhost'
        self.host('localhost.', 200)
        self.host('localhost.', 200)
        self.host('.localhost', 404)
        self.host('www.localhost', 404)
        self.host('localhost1', 404)

    @pytest.mark.skip('not yet')
    def test_routes_match_host_absent(self):
        self.route_match({"host": "localhost"})

        assert (
            self.get(headers={'Connection': 'close'})['status'] == 400
        ), 'match host absent'

    def test_routes_match_host_ipv4(self):
        self.route_match({"host": "127.0.0.1"})

        self.host('127.0.0.1', 200)
        self.host('127.0.0.1:7080', 200)

    def test_routes_match_host_ipv6(self):
        self.route_match({"host": "[::1]"})

        self.host('[::1]', 200)
        self.host('[::1]:7080', 200)

    def test_routes_match_host_positive_many(self):
        self.route_match({"host": ["localhost", "example.com"]})

        assert self.get()['status'] == 200, 'localhost'
        self.host('example.com', 200)

    def test_routes_match_host_positive_and_negative(self):
        self.route_match({"host": ["*example.com", "!www.example.com"]})

        assert self.get()['status'] == 404, 'localhost'
        self.host('example.com', 200)
        self.host('www.example.com', 404)
        self.host('!www.example.com', 200)

    def test_routes_match_host_positive_and_negative_wildcard(self):
        self.route_match({"host": ["*example*", "!www.example*"]})

        self.host('example.com', 200)
        self.host('www.example.com', 404)

    def test_routes_match_host_case_insensitive(self):
        self.route_match({"host": "Example.com"})

        self.host('example.com', 200)
        self.host('EXAMPLE.COM', 200)

    def test_routes_match_host_port(self):
        self.route_match({"host": "example.com"})

        self.host('example.com:7080', 200)

    def test_routes_match_host_empty(self):
        self.route_match({"host": ""})

        self.host('', 200)
        assert (
            self.get(http_10=True, headers={})['status'] == 200
        ), 'match host empty 2'
        assert self.get()['status'] == 404, 'match host empty 3'

    def test_routes_match_uri_positive(self):
        self.route_match({"uri": ["/blah", "/slash/"]})

        assert self.get()['status'] == 404, '/'
        assert self.get(url='/blah')['status'] == 200, '/blah'
        assert self.get(url='/blah#foo')['status'] == 200, '/blah#foo'
        assert self.get(url='/blah?var')['status'] == 200, '/blah?var'
        assert self.get(url='//blah')['status'] == 200, '//blah'
        assert self.get(url='/slash/foo/../')['status'] == 200, 'relative'
        assert self.get(url='/slash/./')['status'] == 200, '/slash/./'
        assert self.get(url='/slash//.//')['status'] == 200, 'adjacent slashes'
        assert self.get(url='/%')['status'] == 400, 'percent'
        assert self.get(url='/%1')['status'] == 400, 'percent digit'
        assert self.get(url='/%A')['status'] == 400, 'percent letter'
        assert self.get(url='/slash/.?args')['status'] == 200, 'dot args'
        assert self.get(url='/slash/.#frag')['status'] == 200, 'dot frag'
        assert (
            self.get(url='/slash/foo/..?args')['status'] == 200
        ), 'dot dot args'
        assert (
            self.get(url='/slash/foo/..#frag')['status'] == 200
        ), 'dot dot frag'
        assert self.get(url='/slash/.')['status'] == 200, 'trailing dot'
        assert (
            self.get(url='/slash/foo/..')['status'] == 200
        ), 'trailing dot dot'

    def test_routes_match_uri_case_sensitive(self):
        self.route_match({"uri": "/BLAH"})

        assert self.get(url='/blah')['status'] == 404, '/blah'
        assert self.get(url='/BlaH')['status'] == 404, '/BlaH'
        assert self.get(url='/BLAH')['status'] == 200, '/BLAH'

    def test_routes_match_uri_normalize(self):
        self.route_match({"uri": "/blah"})

        assert self.get(url='/%62%6c%61%68')['status'] == 200, 'normalize'

    def test_routes_match_empty_array(self):
        self.route_match({"uri": []})

        assert self.get(url='/blah')['status'] == 200, 'empty array'

    def test_routes_reconfigure(self):
        assert 'success' in self.conf([], 'routes'), 'redefine'
        assert self.get()['status'] == 404, 'redefine request'

        assert 'success' in self.conf(
            [{"action": {"return": 200}}], 'routes'
        ), 'redefine 2'
        assert self.get()['status'] == 200, 'redefine request 2'

        assert 'success' in self.conf([], 'routes'), 'redefine 3'
        assert self.get()['status'] == 404, 'redefine request 3'

        assert 'success' in self.conf(
            {
                "listeners": {"*:7080": {"pass": "routes/main"}},
                "routes": {"main": [{"action": {"return": 200}}]},
                "applications": {},
            }
        ), 'redefine 4'
        assert self.get()['status'] == 200, 'redefine request 4'

        assert 'success' in self.conf_delete('routes/main/0'), 'redefine 5'
        assert self.get()['status'] == 404, 'redefine request 5'

        assert 'success' in self.conf_post(
            {"action": {"return": 200}}, 'routes/main'
        ), 'redefine 6'
        assert self.get()['status'] == 200, 'redefine request 6'

        assert 'error' in self.conf(
            {"action": {"return": 200}}, 'routes/main/2'
        ), 'redefine 7'
        assert 'success' in self.conf(
            {"action": {"return": 201}}, 'routes/main/1'
        ), 'redefine 8'

        assert len(self.conf_get('routes/main')) == 2, 'redefine conf 8'
        assert self.get()['status'] == 200, 'redefine request 8'

    def test_routes_edit(self):
        self.route_match({"method": "GET"})

        assert self.get()['status'] == 200, 'routes edit GET'
        assert self.post()['status'] == 404, 'routes edit POST'

        assert 'success' in self.conf_post(
            {"match": {"method": "POST"}, "action": {"return": 200}},
            'routes',
        ), 'routes edit configure 2'
        assert 'GET' == self.conf_get(
            'routes/0/match/method'
        ), 'routes edit configure 2 check'
        assert 'POST' == self.conf_get(
            'routes/1/match/method'
        ), 'routes edit configure 2 check 2'

        assert self.get()['status'] == 200, 'routes edit GET 2'
        assert self.post()['status'] == 200, 'routes edit POST 2'

        assert 'success' in self.conf_delete(
            'routes/0'
        ), 'routes edit configure 3'

        assert self.get()['status'] == 404, 'routes edit GET 3'
        assert self.post()['status'] == 200, 'routes edit POST 3'

        assert 'error' in self.conf_delete(
            'routes/1'
        ), 'routes edit configure invalid'
        assert 'error' in self.conf_delete(
            'routes/-1'
        ), 'routes edit configure invalid 2'
        assert 'error' in self.conf_delete(
            'routes/blah'
        ), 'routes edit configure invalid 3'

        assert self.get()['status'] == 404, 'routes edit GET 4'
        assert self.post()['status'] == 200, 'routes edit POST 4'

        assert 'success' in self.conf_delete(
            'routes/0'
        ), 'routes edit configure 5'

        assert self.get()['status'] == 404, 'routes edit GET 5'
        assert self.post()['status'] == 404, 'routes edit POST 5'

        assert 'success' in self.conf_post(
            {"match": {"method": "POST"}, "action": {"return": 200}},
            'routes',
        ), 'routes edit configure 6'

        assert self.get()['status'] == 404, 'routes edit GET 6'
        assert self.post()['status'] == 200, 'routes edit POST 6'

        assert 'success' in self.conf(
            {
                "listeners": {"*:7080": {"pass": "routes/main"}},
                "routes": {"main": [{"action": {"return": 200}}]},
                "applications": {},
            }
        ), 'route edit configure 7'

        assert 'error' in self.conf_delete(
            'routes/0'
        ), 'routes edit configure invalid 4'
        assert 'error' in self.conf_delete(
            'routes/main'
        ), 'routes edit configure invalid 5'

        assert self.get()['status'] == 200, 'routes edit GET 7'

        assert 'success' in self.conf_delete(
            'listeners/*:7080'
        ), 'route edit configure 8'
        assert 'success' in self.conf_delete(
            'routes/main'
        ), 'route edit configure 9'

    def test_match_edit(self, skip_alert):
        skip_alert(r'failed to apply new conf')

        self.route_match({"method": ["GET", "POST"]})

        assert self.get()['status'] == 200, 'match edit GET'
        assert self.post()['status'] == 200, 'match edit POST'
        assert self.put()['status'] == 404, 'match edit PUT'

        assert 'success' in self.conf_post(
            '\"PUT\"', 'routes/0/match/method'
        ), 'match edit configure 2'
        assert ['GET', 'POST', 'PUT'] == self.conf_get(
            'routes/0/match/method'
        ), 'match edit configure 2 check'

        assert self.get()['status'] == 200, 'match edit GET 2'
        assert self.post()['status'] == 200, 'match edit POST 2'
        assert self.put()['status'] == 200, 'match edit PUT 2'

        assert 'success' in self.conf_delete(
            'routes/0/match/method/1'
        ), 'match edit configure 3'
        assert ['GET', 'PUT'] == self.conf_get(
            'routes/0/match/method'
        ), 'match edit configure 3 check'

        assert self.get()['status'] == 200, 'match edit GET 3'
        assert self.post()['status'] == 404, 'match edit POST 3'
        assert self.put()['status'] == 200, 'match edit PUT 3'

        assert 'success' in self.conf_delete(
            'routes/0/match/method/1'
        ), 'match edit configure 4'
        assert ['GET'] == self.conf_get(
            'routes/0/match/method'
        ), 'match edit configure 4 check'

        assert self.get()['status'] == 200, 'match edit GET 4'
        assert self.post()['status'] == 404, 'match edit POST 4'
        assert self.put()['status'] == 404, 'match edit PUT 4'

        assert 'error' in self.conf_delete(
            'routes/0/match/method/1'
        ), 'match edit configure invalid'
        assert 'error' in self.conf_delete(
            'routes/0/match/method/-1'
        ), 'match edit configure invalid 2'
        assert 'error' in self.conf_delete(
            'routes/0/match/method/blah'
        ), 'match edit configure invalid 3'
        assert ['GET'] == self.conf_get(
            'routes/0/match/method'
        ), 'match edit configure 5 check'

        assert self.get()['status'] == 200, 'match edit GET 5'
        assert self.post()['status'] == 404, 'match edit POST 5'
        assert self.put()['status'] == 404, 'match edit PUT 5'

        assert 'success' in self.conf_delete(
            'routes/0/match/method/0'
        ), 'match edit configure 6'
        assert [] == self.conf_get(
            'routes/0/match/method'
        ), 'match edit configure 6 check'

        assert self.get()['status'] == 200, 'match edit GET 6'
        assert self.post()['status'] == 200, 'match edit POST 6'
        assert self.put()['status'] == 200, 'match edit PUT 6'

        assert 'success' in self.conf(
            '"GET"', 'routes/0/match/method'
        ), 'match edit configure 7'

        assert self.get()['status'] == 200, 'match edit GET 7'
        assert self.post()['status'] == 404, 'match edit POST 7'
        assert self.put()['status'] == 404, 'match edit PUT 7'

        assert 'error' in self.conf_delete(
            'routes/0/match/method/0'
        ), 'match edit configure invalid 5'
        assert 'error' in self.conf(
            {}, 'routes/0/action'
        ), 'match edit configure invalid 6'

        assert 'success' in self.conf(
            {}, 'routes/0/match'
        ), 'match edit configure 8'

        assert self.get()['status'] == 200, 'match edit GET 8'

    def test_routes_match_rules(self):
        self.route_match({"method": "GET", "host": "localhost", "uri": "/"})

        assert self.get()['status'] == 200, 'routes match rules'

    def test_routes_loop(self):
        assert 'success' in self.route(
            {"match": {"uri": "/"}, "action": {"pass": "routes"}}
        ), 'routes loop configure'

        assert self.get()['status'] == 500, 'routes loop'

    def test_routes_match_headers(self):
        self.route_match({"headers": {"host": "localhost"}})

        assert self.get()['status'] == 200, 'match headers'
        self.host('Localhost', 200)
        self.host('localhost.com', 404)
        self.host('llocalhost', 404)
        self.host('host', 404)

    def test_routes_match_headers_multiple(self):
        self.route_match({"headers": {"host": "localhost", "x-blah": "test"}})

        assert self.get()['status'] == 404, 'match headers multiple'
        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": "test",
                    "Connection": "close",
                }
            )['status']
            == 200
        ), 'match headers multiple 2'

        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": "",
                    "Connection": "close",
                }
            )['status']
            == 404
        ), 'match headers multiple 3'

    def test_routes_match_headers_multiple_values(self):
        self.route_match({"headers": {"x-blah": "test"}})

        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": ["test", "test", "test"],
                    "Connection": "close",
                }
            )['status']
            == 200
        ), 'match headers multiple values'
        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": ["test", "blah", "test"],
                    "Connection": "close",
                }
            )['status']
            == 404
        ), 'match headers multiple values 2'
        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": ["test", "", "test"],
                    "Connection": "close",
                }
            )['status']
            == 404
        ), 'match headers multiple values 3'

    def test_routes_match_headers_multiple_rules(self):
        self.route_match({"headers": {"x-blah": ["test", "blah"]}})

        assert self.get()['status'] == 404, 'match headers multiple rules'
        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": "test",
                    "Connection": "close",
                }
            )['status']
            == 200
        ), 'match headers multiple rules 2'
        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": "blah",
                    "Connection": "close",
                }
            )['status']
            == 200
        ), 'match headers multiple rules 3'
        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": ["test", "blah", "test"],
                    "Connection": "close",
                }
            )['status']
            == 200
        ), 'match headers multiple rules 4'

        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "X-blah": ["blah", ""],
                    "Connection": "close",
                }
            )['status']
            == 404
        ), 'match headers multiple rules 5'

    def test_routes_match_headers_case_insensitive(self):
        self.route_match({"headers": {"X-BLAH": "TEST"}})

        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "x-blah": "test",
                    "Connection": "close",
                }
            )['status']
            == 200
        ), 'match headers case insensitive'

    def test_routes_match_headers_invalid(self):
        self.route_match_invalid({"headers": ["blah"]})
        self.route_match_invalid({"headers": {"foo": ["bar", {}]}})
        self.route_match_invalid({"headers": {"": "blah"}})

    def test_routes_match_headers_empty_rule(self):
        self.route_match({"headers": {"host": ""}})

        assert self.get()['status'] == 404, 'localhost'
        self.host('', 200)

    def test_routes_match_headers_empty(self):
        self.route_match({"headers": {}})
        assert self.get()['status'] == 200, 'empty'

        self.route_match({"headers": []})
        assert self.get()['status'] == 200, 'empty 2'

    def test_routes_match_headers_rule_array_empty(self):
        self.route_match({"headers": {"blah": []}})

        assert self.get()['status'] == 404, 'array empty'
        assert (
            self.get(
                headers={
                    "Host": "localhost",
                    "blah": "foo",
                    "Connection": "close",
                }
            )['status']
            == 200
        ), 'match headers rule array empty 2'

    def test_routes_match_headers_array(self):
        self.route_match(
            {
                "headers": [
                    {"x-header1": "foo*"},
                    {"x-header2": "bar"},
                    {"x-header3": ["foo", "bar"]},
                    {"x-header1": "bar", "x-header4": "foo"},
                ]
            }
        )

        def check_headers(hds):
            hds = dict({"Host": "localhost", "Connection": "close"}, **hds)
            assert self.get(headers=hds)['status'] == 200, 'headers array match'

        def check_headers_404(hds):
            hds = dict({"Host": "localhost", "Connection": "close"}, **hds)
            assert (
                self.get(headers=hds)['status'] == 404
            ), 'headers array no match'

        assert self.get()['status'] == 404, 'match headers array'
        check_headers({"x-header1": "foo123"})
        check_headers({"x-header2": "bar"})
        check_headers({"x-header3": "bar"})
        check_headers_404({"x-header1": "bar"})
        check_headers({"x-header1": "bar", "x-header4": "foo"})

        assert 'success' in self.conf_delete(
            'routes/0/match/headers/1'
        ), 'match headers array configure 2'

        check_headers_404({"x-header2": "bar"})
        check_headers({"x-header3": "foo"})

    def test_routes_match_arguments(self):
        self.route_match({"arguments": {"foo": "bar"}})

        assert self.get()['status'] == 404, 'args'
        assert self.get(url='/?foo=bar')['status'] == 200, 'args 2'