diff options
Diffstat (limited to 'test/test_routing.py')
| -rw-r--r-- | test/test_routing.py | 1877 |
1 files changed, 816 insertions, 1061 deletions
diff --git a/test/test_routing.py b/test/test_routing.py index 269e8efc..4107f57e 100644 --- a/test/test_routing.py +++ b/test/test_routing.py @@ -1,117 +1,103 @@ # -*- coding: utf-8 -*- -import unittest +import pytest from unit.applications.proto import TestApplicationProto +from conftest import option, skip_alert class TestRouting(TestApplicationProto): prerequisites = {'modules': {'python': 'any'}} - def setUp(self): - super().setUp() + def setup_method(self): + super().setup_method() - self.assertIn( - 'success', - self.conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [ - { - "match": {"method": "GET"}, - "action": {"return": 200}, - } - ], - "applications": {}, - } - ), - 'routing configure', - ) + 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): - self.assertIn( - 'success', - self.route({"match": match, "action": {"return": 200}}), - 'route match configure', - ) + assert 'success' in self.route( + {"match": match, "action": {"return": 200}} + ), 'route match configure' def route_match_invalid(self, match): - self.assertIn( - 'error', - self.route({"match": match, "action": {"return": 200}}), - 'route match configure invalid', - ) + assert 'error' in self.route( + {"match": match, "action": {"return": 200}} + ), 'route match configure invalid' def host(self, host, status): - self.assertEqual( - self.get(headers={'Host': host, 'Connection': 'close'})[ - 'status' - ], - status, - 'match host', - ) + assert ( + self.get(headers={'Host': host, 'Connection': 'close'})['status'] + == status + ), 'match host' def cookie(self, cookie, status): - self.assertEqual( + assert ( self.get( headers={ 'Host': 'localhost', 'Cookie': cookie, 'Connection': 'close', }, - )['status'], - status, - 'match cookie', - ) + )['status'] + == status + ), 'match cookie' def test_routes_match_method_positive(self): - self.assertEqual(self.get()['status'], 200, 'GET') - self.assertEqual(self.post()['status'], 404, 'POST') + 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"]}) - self.assertEqual(self.get()['status'], 200, 'GET') - self.assertEqual(self.post()['status'], 200, 'POST') - self.assertEqual(self.delete()['status'], 404, 'DELETE') + 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"}) - self.assertEqual(self.get()['status'], 404, 'GET') - self.assertEqual(self.post()['status'], 200, 'POST') + 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"]}) - self.assertEqual(self.get()['status'], 404, 'GET') - self.assertEqual(self.post()['status'], 404, 'POST') - self.assertEqual(self.delete()['status'], 200, 'DELETE') + 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"}) - self.assertEqual(self.get()['status'], 200, 'GET') - self.assertEqual(self.post()['status'], 404, 'POST') + 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*"}) - self.assertEqual(self.get()['status'], 200, 'GET') - self.assertEqual(self.post()['status'], 404, 'POST') + 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*"}) - self.assertEqual(self.get()['status'], 200, 'GET') - self.assertEqual(self.post()['status'], 404, 'POST') + assert self.get()['status'] == 200, 'GET' + assert self.post()['status'] == 404, 'POST' def test_routes_match_method_wildcard(self): self.route_match({"method": "*"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' def test_routes_match_invalid(self): self.route_match_invalid({"method": "**"}) @@ -124,32 +110,32 @@ class TestRouting(TestApplicationProto): def test_routes_match_empty_exact(self): self.route_match({"uri": ""}) - self.assertEqual(self.get()['status'], 404) + assert self.get()['status'] == 404 self.route_match({"uri": "/"}) - self.assertEqual(self.get()['status'], 200) - self.assertEqual(self.get(url='/blah')['status'], 404) + assert self.get()['status'] == 200 + assert self.get(url='/blah')['status'] == 404 def test_routes_match_negative(self): self.route_match({"uri": "!"}) - self.assertEqual(self.get()['status'], 404) + assert self.get()['status'] == 404 self.route_match({"uri": "!/"}) - self.assertEqual(self.get()['status'], 404) - self.assertEqual(self.get(url='/blah')['status'], 200) + assert self.get()['status'] == 404 + assert self.get(url='/blah')['status'] == 200 self.route_match({"uri": "!*blah"}) - self.assertEqual(self.get()['status'], 200) - self.assertEqual(self.get(url='/bla')['status'], 200) - self.assertEqual(self.get(url='/blah')['status'], 404) - self.assertEqual(self.get(url='/blah1')['status'], 200) + 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*"}) - self.assertEqual(self.get()['status'], 200) - self.assertEqual(self.get(url='/blah')['status'], 200) - self.assertEqual(self.get(url='/blah1')['status'], 404) - self.assertEqual(self.get(url='/blah12')['status'], 404) - self.assertEqual(self.get(url='/blah2')['status'], 200) + 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"}) @@ -162,110 +148,105 @@ class TestRouting(TestApplicationProto): def test_routes_match_method_case_insensitive(self): self.route_match({"method": "get"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' def test_routes_match_wildcard_left_case_insensitive(self): self.route_match({"method": "*get"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' self.route_match({"method": "*et"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' def test_routes_match_wildcard_middle_case_insensitive(self): self.route_match({"method": "g*t"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' def test_routes_match_wildcard_right_case_insensitive(self): self.route_match({"method": "get*"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' self.route_match({"method": "ge*"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' def test_routes_match_wildcard_substring_case_insensitive(self): self.route_match({"method": "*et*"}) - self.assertEqual(self.get()['status'], 200, 'GET') + assert self.get()['status'] == 200, 'GET' def test_routes_match_wildcard_left_case_sensitive(self): self.route_match({"uri": "*blah"}) - self.assertEqual(self.get(url='/blah')['status'], 200, '/blah') - self.assertEqual(self.get(url='/BLAH')['status'], 404, '/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"}) - self.assertEqual(self.get(url='/blah')['status'], 200, '/blah') - self.assertEqual(self.get(url='/BLAH')['status'], 404, '/BLAH') + 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*"}) - self.assertEqual(self.get(url='/axy')['status'], 200, '/axy') - self.assertEqual(self.get(url='/ayx')['status'], 404, '/ayx') + 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*"}) - self.assertEqual(self.get(url='/bla_foo')['status'], 404, '/bla_foo') + 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*"}) - self.assertEqual(self.get(url='/bla_foo')['status'], 404, '/bla_foo') + assert self.get(url='/bla_foo')['status'] == 404, '/bla_foo' def test_route_match_wildcards_adjust_end(self): self.route_match({"uri": "/bla*bla"}) - self.assertEqual(self.get(url='/foo_bla')['status'], 404, '/foo_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*"}) - self.assertEqual(self.get(url='/blah')['status'], 200, '/blah') - self.assertEqual(self.get(url='/BLAH')['status'], 404, '/BLAH') + 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*"}) - self.assertEqual(self.get(url='/blah')['status'], 200, '/blah') - self.assertEqual(self.get(url='/BLAH')['status'], 404, '/BLAH') + 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*"}) - self.assertEqual(self.get(url='/blah-a-B-c-blah')['status'], 200) - self.assertEqual(self.get(url='/a-B-c')['status'], 200) - self.assertEqual(self.get(url='/aBc')['status'], 200) - self.assertEqual(self.get(url='/aBCaBbc')['status'], 200) - self.assertEqual(self.get(url='/ABc')['status'], 404) + 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_pass_encode(self): def check_pass(path, name): - self.assertIn( - 'success', - self.conf( - { - "listeners": { - "*:7080": {"pass": "applications/" + path} - }, - "applications": { - name: { - "type": "python", - "processes": {"spare": 0}, - "path": self.current_dir + '/python/empty', - "working_directory": self.current_dir - + '/python/empty', - "module": "wsgi", - } - }, - } - ), + assert 'success' in self.conf( + { + "listeners": {"*:7080": {"pass": "applications/" + path}}, + "applications": { + name: { + "type": "python", + "processes": {"spare": 0}, + "path": option.test_dir + '/python/empty', + "working_directory": option.test_dir + + '/python/empty', + "module": "wsgi", + } + }, + } ) - self.assertEqual(self.get()['status'], 200) + assert self.get()['status'] == 200 check_pass("%25", "%") check_pass("blah%2Fblah", "blah/blah") @@ -273,25 +254,20 @@ class TestRouting(TestApplicationProto): check_pass("%20blah%252Fblah%7E", " blah%2Fblah~") def check_pass_error(path, name): - self.assertIn( - 'error', - self.conf( - { - "listeners": { - "*:7080": {"pass": "applications/" + path} - }, - "applications": { - name: { - "type": "python", - "processes": {"spare": 0}, - "path": self.current_dir + '/python/empty', - "working_directory": self.current_dir - + '/python/empty', - "module": "wsgi", - } - }, - } - ), + assert 'error' in self.conf( + { + "listeners": {"*:7080": {"pass": "applications/" + path}}, + "applications": { + name: { + "type": "python", + "processes": {"spare": 0}, + "path": option.test_dir + '/python/empty', + "working_directory": option.test_dir + + '/python/empty', + "module": "wsgi", + } + }, + } ) check_pass_error("%", "%") @@ -305,8 +281,8 @@ class TestRouting(TestApplicationProto): "empty": { "type": "python", "processes": {"spare": 0}, - "path": self.current_dir + '/python/empty', - "working_directory": self.current_dir + "path": option.test_dir + '/python/empty', + "working_directory": option.test_dir + '/python/empty', "module": "wsgi", } @@ -314,179 +290,135 @@ class TestRouting(TestApplicationProto): } ) - self.assertEqual(self.get(port=7081)['status'], 200, 'routes absent') + assert 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', - ) + assert 'error' in 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": {}, - } - ), - 'route empty configure', - ) + assert 'success' in self.conf( + { + "listeners": {"*:7080": {"pass": "routes/main"}}, + "routes": {"main": []}, + "applications": {}, + } + ), 'route empty configure' - self.assertEqual(self.get()['status'], 404, 'route empty') + assert self.get()['status'] == 404, 'route empty' def test_routes_route_empty(self): - self.assertIn( - 'success', - self.conf({}, 'listeners'), - 'routes empty listeners configure', - ) + assert 'success' in self.conf( + {}, 'listeners' + ), 'routes empty listeners configure' - self.assertIn( - 'success', self.conf({}, 'routes'), 'routes empty configure' - ) + assert 'success' in self.conf({}, 'routes'), 'routes empty configure' def test_routes_route_match_absent(self): - self.assertIn( - 'success', - self.conf([{"action": {"return": 200}}], 'routes'), - 'route match absent configure', - ) + assert 'success' in self.conf( + [{"action": {"return": 200}}], 'routes' + ), 'route match absent configure' - self.assertEqual(self.get()['status'], 200, 'route match absent') + assert self.get()['status'] == 200, 'route match absent' def test_routes_route_action_absent(self): - self.skip_alerts.append(r'failed to apply new conf') + skip_alert(r'failed to apply new conf') - self.assertIn( - 'error', - self.conf([{"match": {"method": "GET"}}], 'routes'), - 'route pass absent configure', - ) + assert 'error' in self.conf( + [{"match": {"method": "GET"}}], 'routes' + ), 'route pass absent configure' def test_routes_route_pass_absent(self): - self.assertIn( - 'error', - self.conf([{"match": {"method": "GET"}, "action": {}}], 'routes'), - 'route pass absent configure', - ) + assert 'error' in self.conf( + [{"match": {"method": "GET"}, "action": {}}], 'routes' + ), 'route pass absent configure' def test_routes_action_unique(self): - self.assertIn( - 'success', - self.conf( - { - "listeners": { - "*:7080": {"pass": "routes"}, - "*:7081": {"pass": "applications/app"}, - }, - "routes": [{"action": {"proxy": "http://127.0.0.1:7081"}}], - "applications": { - "app": { - "type": "python", - "processes": {"spare": 0}, - "path": "/app", - "module": "wsgi", - } - }, - } - ), - ) - - self.assertIn( - 'error', - self.conf( - {"proxy": "http://127.0.0.1:7081", "share": self.testdir}, - 'routes/0/action', - ), - 'proxy share', - ) - self.assertIn( - 'error', - self.conf( - { - "proxy": "http://127.0.0.1:7081", - "pass": "applications/app", + assert 'success' in self.conf( + { + "listeners": { + "*:7080": {"pass": "routes"}, + "*:7081": {"pass": "applications/app"}, }, - 'routes/0/action', - ), - 'proxy pass', - ) - self.assertIn( - 'error', - self.conf( - {"share": self.testdir, "pass": "applications/app"}, - 'routes/0/action', - ), - 'share pass', + "routes": [{"action": {"proxy": "http://127.0.0.1:7081"}}], + "applications": { + "app": { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, + } ) + assert 'error' in self.conf( + {"proxy": "http://127.0.0.1:7081", "share": self.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": self.temp_dir, "pass": "applications/app"}, + 'routes/0/action', + ), 'share pass' + def test_routes_rules_two(self): - self.assertIn( - 'success', - self.conf( - [ - {"match": {"method": "GET"}, "action": {"return": 200}}, - {"match": {"method": "POST"}, "action": {"return": 201}}, - ], - 'routes', - ), - 'rules two configure', - ) + assert 'success' in self.conf( + [ + {"match": {"method": "GET"}, "action": {"return": 200}}, + {"match": {"method": "POST"}, "action": {"return": 201}}, + ], + 'routes', + ), 'rules two configure' - self.assertEqual(self.get()['status'], 200, 'rules two match first') - self.assertEqual(self.post()['status'], 201, 'rules two match second') + assert self.get()['status'] == 200, 'rules two match first' + assert self.post()['status'] == 201, '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": {"return": 200}, - } - ], - }, - "applications": {}, - } - ), - 'routes two configure', - ) + 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' - self.assertEqual(self.get()['status'], 200, 'routes two') + assert self.get()['status'] == 200, 'routes two' def test_routes_match_host_positive(self): self.route_match({"host": "localhost"}) - self.assertEqual(self.get()['status'], 200, '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) - @unittest.skip('not yet') + @pytest.mark.skip('not yet') def test_routes_match_host_absent(self): self.route_match({"host": "localhost"}) - self.assertEqual( - self.get(headers={'Connection': 'close'})['status'], - 400, - 'match host absent', - ) + 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"}) @@ -503,13 +435,13 @@ class TestRouting(TestApplicationProto): def test_routes_match_host_positive_many(self): self.route_match({"host": ["localhost", "example.com"]}) - self.assertEqual(self.get()['status'], 200, 'localhost') + 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"]}) - self.assertEqual(self.get()['status'], 404, 'localhost') + assert self.get()['status'] == 404, 'localhost' self.host('example.com', 200) self.host('www.example.com', 404) self.host('!www.example.com', 200) @@ -535,380 +467,278 @@ class TestRouting(TestApplicationProto): self.route_match({"host": ""}) self.host('', 200) - self.assertEqual( - self.get(http_10=True, headers={})['status'], - 200, - 'match host empty 2', - ) - self.assertEqual(self.get()['status'], 404, 'match host empty 3') + 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/"]}) - self.assertEqual(self.get()['status'], 404, '/') - self.assertEqual(self.get(url='/blah')['status'], 200, '/blah') - self.assertEqual(self.get(url='/blah#foo')['status'], 200, '/blah#foo') - self.assertEqual(self.get(url='/blah?var')['status'], 200, '/blah?var') - self.assertEqual(self.get(url='//blah')['status'], 200, '//blah') - self.assertEqual( - self.get(url='/slash/foo/../')['status'], 200, 'relative' - ) - self.assertEqual(self.get(url='/slash/./')['status'], 200, '/slash/./') - self.assertEqual( - self.get(url='/slash//.//')['status'], 200, 'adjacent slashes' - ) - self.assertEqual(self.get(url='/%')['status'], 400, 'percent') - self.assertEqual(self.get(url='/%1')['status'], 400, 'percent digit') - self.assertEqual(self.get(url='/%A')['status'], 400, 'percent letter') - self.assertEqual( - self.get(url='/slash/.?args')['status'], 200, 'dot args' - ) - self.assertEqual( - self.get(url='/slash/.#frag')['status'], 200, 'dot frag' - ) - self.assertEqual( - self.get(url='/slash/foo/..?args')['status'], - 200, - 'dot dot args', - ) - self.assertEqual( - self.get(url='/slash/foo/..#frag')['status'], - 200, - 'dot dot frag', - ) - self.assertEqual( - self.get(url='/slash/.')['status'], 200, 'trailing dot' - ) - self.assertEqual( - self.get(url='/slash/foo/..')['status'], - 200, - 'trailing dot dot', - ) + 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"}) - self.assertEqual(self.get(url='/blah')['status'], 404, '/blah') - self.assertEqual(self.get(url='/BlaH')['status'], 404, '/BlaH') - self.assertEqual(self.get(url='/BLAH')['status'], 200, '/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"}) - self.assertEqual( - self.get(url='/%62%6c%61%68')['status'], 200, 'normalize' - ) + assert self.get(url='/%62%6c%61%68')['status'] == 200, 'normalize' def test_routes_match_empty_array(self): self.route_match({"uri": []}) - self.assertEqual(self.get(url='/blah')['status'], 200, 'empty array') + assert self.get(url='/blah')['status'] == 200, 'empty array' def test_routes_reconfigure(self): - self.assertIn('success', self.conf([], 'routes'), 'redefine') - self.assertEqual(self.get()['status'], 404, 'redefine request') + assert 'success' in self.conf([], 'routes'), 'redefine' + assert self.get()['status'] == 404, 'redefine request' - self.assertIn( - 'success', - self.conf([{"action": {"return": 200}}], 'routes'), - 'redefine 2', - ) - self.assertEqual(self.get()['status'], 200, 'redefine request 2') + assert 'success' in self.conf( + [{"action": {"return": 200}}], 'routes' + ), 'redefine 2' + assert self.get()['status'] == 200, 'redefine request 2' - self.assertIn('success', self.conf([], 'routes'), 'redefine 3') - self.assertEqual(self.get()['status'], 404, 'redefine request 3') + assert 'success' in self.conf([], 'routes'), 'redefine 3' + assert self.get()['status'] == 404, 'redefine request 3' - self.assertIn( - 'success', - self.conf( - { - "listeners": {"*:7080": {"pass": "routes/main"}}, - "routes": {"main": [{"action": {"return": 200}}]}, - "applications": {}, - } - ), - 'redefine 4', - ) - self.assertEqual(self.get()['status'], 200, 'redefine request 4') + 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' - self.assertIn( - 'success', self.conf_delete('routes/main/0'), 'redefine 5' - ) - self.assertEqual(self.get()['status'], 404, 'redefine request 5') + assert 'success' in self.conf_delete('routes/main/0'), 'redefine 5' + assert self.get()['status'] == 404, 'redefine request 5' - self.assertIn( - 'success', - self.conf_post({"action": {"return": 200}}, 'routes/main'), - 'redefine 6', - ) - self.assertEqual(self.get()['status'], 200, 'redefine request 6') + assert 'success' in self.conf_post( + {"action": {"return": 200}}, 'routes/main' + ), 'redefine 6' + assert self.get()['status'] == 200, 'redefine request 6' - self.assertIn( - 'error', - self.conf({"action": {"return": 200}}, 'routes/main/2'), - 'redefine 7', - ) - self.assertIn( - 'success', - self.conf({"action": {"return": 201}}, 'routes/main/1'), - 'redefine 8', - ) + 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' - self.assertEqual( - len(self.conf_get('routes/main')), 2, 'redefine conf 8' - ) - self.assertEqual(self.get()['status'], 200, 'redefine request 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"}) - self.assertEqual(self.get()['status'], 200, 'routes edit GET') - self.assertEqual(self.post()['status'], 404, 'routes edit POST') - - self.assertIn( - 'success', - self.conf_post( - {"match": {"method": "POST"}, "action": {"return": 200}}, - 'routes', - ), - 'routes edit configure 2', - ) - self.assertEqual( - 'GET', - self.conf_get('routes/0/match/method'), - 'routes edit configure 2 check', - ) - self.assertEqual( - 'POST', - self.conf_get('routes/1/match/method'), - 'routes edit configure 2 check 2', - ) - - self.assertEqual(self.get()['status'], 200, 'routes edit GET 2') - self.assertEqual(self.post()['status'], 200, 'routes edit POST 2') - - self.assertIn( - 'success', self.conf_delete('routes/0'), 'routes edit configure 3', - ) - - self.assertEqual(self.get()['status'], 404, 'routes edit GET 3') - self.assertEqual(self.post()['status'], 200, 'routes edit POST 3') - - self.assertIn( - 'error', - self.conf_delete('routes/1'), - 'routes edit configure invalid', - ) - self.assertIn( - 'error', - self.conf_delete('routes/-1'), - 'routes edit configure invalid 2', - ) - self.assertIn( - 'error', - self.conf_delete('routes/blah'), - 'routes edit configure invalid 3', - ) - - self.assertEqual(self.get()['status'], 404, 'routes edit GET 4') - self.assertEqual(self.post()['status'], 200, 'routes edit POST 4') - - self.assertIn( - 'success', self.conf_delete('routes/0'), 'routes edit configure 5', - ) - - self.assertEqual(self.get()['status'], 404, 'routes edit GET 5') - self.assertEqual(self.post()['status'], 404, 'routes edit POST 5') - - self.assertIn( - 'success', - self.conf_post( - {"match": {"method": "POST"}, "action": {"return": 200}}, - 'routes', - ), - 'routes edit configure 6', - ) - - self.assertEqual(self.get()['status'], 404, 'routes edit GET 6') - self.assertEqual(self.post()['status'], 200, 'routes edit POST 6') - - self.assertIn( - 'success', - self.conf( - { - "listeners": {"*:7080": {"pass": "routes/main"}}, - "routes": {"main": [{"action": {"return": 200}}]}, - "applications": {}, - } - ), - 'route edit configure 7', - ) + 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' - self.assertIn( - 'error', - self.conf_delete('routes/0'), - 'routes edit configure invalid 4', - ) - self.assertIn( - 'error', - self.conf_delete('routes/main'), - 'routes edit configure invalid 5', - ) + 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' - self.assertEqual(self.get()['status'], 200, 'routes edit GET 7') + assert self.get()['status'] == 200, 'routes edit GET 7' - self.assertIn( - 'success', - self.conf_delete('listeners/*:7080'), - 'route edit configure 8', - ) - self.assertIn( - 'success', - self.conf_delete('routes/main'), - 'route edit configure 9', - ) + 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): - self.skip_alerts.append(r'failed to apply new conf') + skip_alert(r'failed to apply new conf') self.route_match({"method": ["GET", "POST"]}) - self.assertEqual(self.get()['status'], 200, 'match edit GET') - self.assertEqual(self.post()['status'], 200, 'match edit POST') - self.assertEqual(self.put()['status'], 404, 'match edit PUT') - - self.assertIn( - 'success', - self.conf_post('\"PUT\"', 'routes/0/match/method'), - 'match edit configure 2', - ) - self.assertListEqual( - ['GET', 'POST', 'PUT'], - self.conf_get('routes/0/match/method'), - 'match edit configure 2 check', - ) - - self.assertEqual(self.get()['status'], 200, 'match edit GET 2') - self.assertEqual(self.post()['status'], 200, 'match edit POST 2') - self.assertEqual(self.put()['status'], 200, 'match edit PUT 2') - - self.assertIn( - 'success', - self.conf_delete('routes/0/match/method/1'), - 'match edit configure 3', - ) - self.assertListEqual( - ['GET', 'PUT'], - self.conf_get('routes/0/match/method'), - 'match edit configure 3 check', - ) - - self.assertEqual(self.get()['status'], 200, 'match edit GET 3') - self.assertEqual(self.post()['status'], 404, 'match edit POST 3') - self.assertEqual(self.put()['status'], 200, 'match edit PUT 3') - - self.assertIn( - 'success', - self.conf_delete('routes/0/match/method/1'), - 'match edit configure 4', - ) - self.assertListEqual( - ['GET'], - self.conf_get('routes/0/match/method'), - 'match edit configure 4 check', - ) - - self.assertEqual(self.get()['status'], 200, 'match edit GET 4') - self.assertEqual(self.post()['status'], 404, 'match edit POST 4') - self.assertEqual(self.put()['status'], 404, 'match edit PUT 4') - - self.assertIn( - 'error', - self.conf_delete('routes/0/match/method/1'), - 'match edit configure invalid', - ) - self.assertIn( - 'error', - self.conf_delete('routes/0/match/method/-1'), - 'match edit configure invalid 2', - ) - self.assertIn( - 'error', - self.conf_delete('routes/0/match/method/blah'), - 'match edit configure invalid 3', - ) - self.assertListEqual( - ['GET'], - self.conf_get('routes/0/match/method'), - 'match edit configure 5 check', - ) - - self.assertEqual(self.get()['status'], 200, 'match edit GET 5') - self.assertEqual(self.post()['status'], 404, 'match edit POST 5') - self.assertEqual(self.put()['status'], 404, 'match edit PUT 5') - - self.assertIn( - 'success', - self.conf_delete('routes/0/match/method/0'), - 'match edit configure 6', - ) - self.assertListEqual( - [], - self.conf_get('routes/0/match/method'), - 'match edit configure 6 check', - ) - - self.assertEqual(self.get()['status'], 200, 'match edit GET 6') - self.assertEqual(self.post()['status'], 200, 'match edit POST 6') - self.assertEqual(self.put()['status'], 200, 'match edit PUT 6') - - self.assertIn( - 'success', - self.conf('"GET"', 'routes/0/match/method'), - 'match edit configure 7', - ) - - self.assertEqual(self.get()['status'], 200, 'match edit GET 7') - self.assertEqual(self.post()['status'], 404, 'match edit POST 7') - self.assertEqual(self.put()['status'], 404, 'match edit PUT 7') - - self.assertIn( - 'error', - self.conf_delete('routes/0/match/method/0'), - 'match edit configure invalid 5', - ) - self.assertIn( - 'error', - self.conf({}, 'routes/0/action'), - 'match edit configure invalid 6', - ) - - self.assertIn( - 'success', - self.conf({}, 'routes/0/match'), - 'match edit configure 8', - ) - - self.assertEqual(self.get()['status'], 200, 'match edit GET 8') + 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": "/"}) - self.assertEqual(self.get()['status'], 200, 'routes match rules') + assert self.get()['status'] == 200, 'routes match rules' def test_routes_loop(self): - self.assertIn( - 'success', - self.route({"match": {"uri": "/"}, "action": {"pass": "routes"}}), - 'routes loop configure', - ) + assert 'success' in self.route( + {"match": {"uri": "/"}, "action": {"pass": "routes"}} + ), 'routes loop configure' - self.assertEqual(self.get()['status'], 500, 'routes loop') + assert self.get()['status'] == 500, 'routes loop' def test_routes_match_headers(self): self.route_match({"headers": {"host": "localhost"}}) - self.assertEqual(self.get()['status'], 200, 'match headers') + assert self.get()['status'] == 200, 'match headers' self.host('Localhost', 200) self.host('localhost.com', 404) self.host('llocalhost', 404) @@ -917,134 +747,122 @@ class TestRouting(TestApplicationProto): def test_routes_match_headers_multiple(self): self.route_match({"headers": {"host": "localhost", "x-blah": "test"}}) - self.assertEqual(self.get()['status'], 404, 'match headers multiple') - self.assertEqual( + 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', - ) + )['status'] + == 200 + ), 'match headers multiple 2' - self.assertEqual( + assert ( self.get( headers={ "Host": "localhost", "X-blah": "", "Connection": "close", } - )['status'], - 404, - 'match headers multiple 3', - ) + )['status'] + == 404 + ), 'match headers multiple 3' def test_routes_match_headers_multiple_values(self): self.route_match({"headers": {"x-blah": "test"}}) - self.assertEqual( + assert ( self.get( headers={ "Host": "localhost", "X-blah": ["test", "test", "test"], "Connection": "close", } - )['status'], - 200, - 'match headers multiple values', - ) - self.assertEqual( + )['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', - ) - self.assertEqual( + )['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', - ) + )['status'] + == 404 + ), 'match headers multiple values 3' def test_routes_match_headers_multiple_rules(self): self.route_match({"headers": {"x-blah": ["test", "blah"]}}) - self.assertEqual( - self.get()['status'], 404, 'match headers multiple rules' - ) - self.assertEqual( + 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', - ) - self.assertEqual( + )['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', - ) - self.assertEqual( + )['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', - ) + )['status'] + == 200 + ), 'match headers multiple rules 4' - self.assertEqual( + assert ( self.get( headers={ "Host": "localhost", "X-blah": ["blah", ""], "Connection": "close", } - )['status'], - 404, - 'match headers multiple rules 5', - ) + )['status'] + == 404 + ), 'match headers multiple rules 5' def test_routes_match_headers_case_insensitive(self): self.route_match({"headers": {"X-BLAH": "TEST"}}) - self.assertEqual( + assert ( self.get( headers={ "Host": "localhost", "x-blah": "test", "Connection": "close", } - )['status'], - 200, - 'match headers case insensitive', - ) + )['status'] + == 200 + ), 'match headers case insensitive' def test_routes_match_headers_invalid(self): self.route_match_invalid({"headers": ["blah"]}) @@ -1054,29 +872,30 @@ class TestRouting(TestApplicationProto): def test_routes_match_headers_empty_rule(self): self.route_match({"headers": {"host": ""}}) - self.assertEqual(self.get()['status'], 404, 'localhost') + assert self.get()['status'] == 404, 'localhost' self.host('', 200) def test_routes_match_headers_empty(self): self.route_match({"headers": {}}) - self.assertEqual(self.get()['status'], 200, 'empty') + assert self.get()['status'] == 200, 'empty' self.route_match({"headers": []}) - self.assertEqual(self.get()['status'], 200, 'empty 2') + assert self.get()['status'] == 200, 'empty 2' def test_routes_match_headers_rule_array_empty(self): self.route_match({"headers": {"blah": []}}) - self.assertEqual(self.get()['status'], 404, 'array empty') - self.assertEqual( + 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' - ) + )['status'] + == 200 + ), 'match headers rule array empty 2' def test_routes_match_headers_array(self): self.route_match( @@ -1090,52 +909,48 @@ class TestRouting(TestApplicationProto): } ) - self.assertEqual(self.get()['status'], 404, 'match headers array') - self.assertEqual( + assert self.get()['status'] == 404, 'match headers array' + assert ( self.get( headers={ "Host": "localhost", "x-header1": "foo123", "Connection": "close", } - )['status'], - 200, - 'match headers array 2', - ) - self.assertEqual( + )['status'] + == 200 + ), 'match headers array 2' + assert ( self.get( headers={ "Host": "localhost", "x-header2": "bar", "Connection": "close", } - )['status'], - 200, - 'match headers array 3', - ) - self.assertEqual( + )['status'] + == 200 + ), 'match headers array 3' + assert ( self.get( headers={ "Host": "localhost", "x-header3": "bar", "Connection": "close", } - )['status'], - 200, - 'match headers array 4', - ) - self.assertEqual( + )['status'] + == 200 + ), 'match headers array 4' + assert ( self.get( headers={ "Host": "localhost", "x-header1": "bar", "Connection": "close", } - )['status'], - 404, - 'match headers array 5', - ) - self.assertEqual( + )['status'] + == 404 + ), 'match headers array 5' + assert ( self.get( headers={ "Host": "localhost", @@ -1143,49 +958,44 @@ class TestRouting(TestApplicationProto): "x-header4": "foo", "Connection": "close", } - )['status'], - 200, - 'match headers array 6', - ) + )['status'] + == 200 + ), 'match headers array 6' - self.assertIn( - 'success', - self.conf_delete('routes/0/match/headers/1'), - 'match headers array configure 2', - ) + assert 'success' in self.conf_delete( + 'routes/0/match/headers/1' + ), 'match headers array configure 2' - self.assertEqual( + assert ( self.get( headers={ "Host": "localhost", "x-header2": "bar", "Connection": "close", } - )['status'], - 404, - 'match headers array 7', - ) - self.assertEqual( + )['status'] + == 404 + ), 'match headers array 7' + assert ( self.get( headers={ "Host": "localhost", "x-header3": "foo", "Connection": "close", } - )['status'], - 200, - 'match headers array 8', - ) + )['status'] + == 200 + ), 'match headers array 8' def test_routes_match_arguments(self): self.route_match({"arguments": {"foo": "bar"}}) - self.assertEqual(self.get()['status'], 404, 'args') - self.assertEqual(self.get(url='/?foo=bar')['status'], 200, 'args 2') - self.assertEqual(self.get(url='/?foo=bar1')['status'], 404, 'args 3') - self.assertEqual(self.get(url='/?1foo=bar')['status'], 404, 'args 4') - self.assertEqual(self.get(url='/?Foo=bar')['status'], 404, 'case') - self.assertEqual(self.get(url='/?foo=Bar')['status'], 404, 'case 2') + assert self.get()['status'] == 404, 'args' + assert self.get(url='/?foo=bar')['status'] == 200, 'args 2' + assert self.get(url='/?foo=bar1')['status'] == 404, 'args 3' + assert self.get(url='/?1foo=bar')['status'] == 404, 'args 4' + assert self.get(url='/?Foo=bar')['status'] == 404, 'case' + assert self.get(url='/?foo=Bar')['status'] == 404, 'case 2' def test_routes_match_arguments_chars(self): chars = ( @@ -1195,15 +1005,30 @@ class TestRouting(TestApplicationProto): chars_enc = "" for h1 in ["2", "3", "4", "5", "6", "7"]: - for h2 in ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "A", - "B", "C", "D", "E", "F", + for h2 in [ + "0", + "1", + "2", + "3", + "4", + "5", + "6", + "7", + "8", + "9", + "A", + "B", + "C", + "D", + "E", + "F", ]: chars_enc += "%" + h1 + h2 chars_enc = chars_enc[:-3] def check_args(args, query): self.route_match({"arguments": args}) - self.assertEqual(self.get(url='/?' + query)['status'], 200) + assert self.get(url='/?' + query)['status'] == 200 check_args({chars: chars}, chars + '=' + chars) check_args({chars: chars}, chars + '=' + chars_enc) @@ -1216,175 +1041,155 @@ class TestRouting(TestApplicationProto): def test_routes_match_arguments_empty(self): self.route_match({"arguments": {}}) - self.assertEqual(self.get()['status'], 200, 'arguments empty') + assert self.get()['status'] == 200, 'arguments empty' self.route_match({"arguments": []}) - self.assertEqual(self.get()['status'], 200, 'arguments empty 2') + assert self.get()['status'] == 200, 'arguments empty 2' def test_routes_match_arguments_space(self): self.route_match({"arguments": {"+fo o%20": "%20b+a r"}}) - self.assertEqual(self.get(url='/? fo o = b a r&')['status'], 200) - self.assertEqual(self.get(url='/?+fo+o+=+b+a+r&')['status'], 200) - self.assertEqual( - self.get(url='/?%20fo%20o%20=%20b%20a%20r&')['status'], 200 - ) + assert self.get(url='/? fo o = b a r&')['status'] == 200 + assert self.get(url='/?+fo+o+=+b+a+r&')['status'] == 200 + assert self.get(url='/?%20fo%20o%20=%20b%20a%20r&')['status'] == 200 self.route_match({"arguments": {"%20foo": " bar"}}) - self.assertEqual(self.get(url='/? foo= bar')['status'], 200) - self.assertEqual(self.get(url='/?+foo=+bar')['status'], 200) - self.assertEqual(self.get(url='/?%20foo=%20bar')['status'], 200) - self.assertEqual(self.get(url='/?+foo= bar')['status'], 200) - self.assertEqual(self.get(url='/?%20foo=+bar')['status'], 200) + assert self.get(url='/? foo= bar')['status'] == 200 + assert self.get(url='/?+foo=+bar')['status'] == 200 + assert self.get(url='/?%20foo=%20bar')['status'] == 200 + assert self.get(url='/?+foo= bar')['status'] == 200 + assert self.get(url='/?%20foo=+bar')['status'] == 200 def test_routes_match_arguments_equal(self): self.route_match({"arguments": {"=": "="}}) - self.assertEqual(self.get(url='/?%3D=%3D')['status'], 200) - self.assertEqual(self.get(url='/?%3D==')['status'], 200) - self.assertEqual(self.get(url='/?===')['status'], 404) - self.assertEqual(self.get(url='/?%3D%3D%3D')['status'], 404) - self.assertEqual(self.get(url='/?==%3D')['status'], 404) + assert self.get(url='/?%3D=%3D')['status'] == 200 + assert self.get(url='/?%3D==')['status'] == 200 + assert self.get(url='/?===')['status'] == 404 + assert self.get(url='/?%3D%3D%3D')['status'] == 404 + assert self.get(url='/?==%3D')['status'] == 404 def test_routes_match_arguments_enc(self): self.route_match({"arguments": {"Ю": "н"}}) - self.assertEqual(self.get(url='/?%D0%AE=%D0%BD')['status'], 200) - self.assertEqual(self.get(url='/?%d0%ae=%d0%Bd')['status'], 200) + assert self.get(url='/?%D0%AE=%D0%BD')['status'] == 200 + assert self.get(url='/?%d0%ae=%d0%Bd')['status'] == 200 def test_routes_match_arguments_hash(self): self.route_match({"arguments": {"#": "#"}}) - self.assertEqual(self.get(url='/?%23=%23')['status'], 200) - self.assertEqual(self.get(url='/?%23=%23#')['status'], 200) - self.assertEqual(self.get(url='/?#=#')['status'], 404) - self.assertEqual(self.get(url='/?%23=#')['status'], 404) + assert self.get(url='/?%23=%23')['status'] == 200 + assert self.get(url='/?%23=%23#')['status'] == 200 + assert self.get(url='/?#=#')['status'] == 404 + assert self.get(url='/?%23=#')['status'] == 404 def test_routes_match_arguments_wildcard(self): self.route_match({"arguments": {"foo": "*"}}) - self.assertEqual(self.get(url='/?foo')['status'], 200) - self.assertEqual(self.get(url='/?foo=')['status'], 200) - self.assertEqual(self.get(url='/?foo=blah')['status'], 200) - self.assertEqual(self.get(url='/?blah=foo')['status'], 404) + assert self.get(url='/?foo')['status'] == 200 + assert self.get(url='/?foo=')['status'] == 200 + assert self.get(url='/?foo=blah')['status'] == 200 + assert self.get(url='/?blah=foo')['status'] == 404 self.route_match({"arguments": {"foo": "%25*"}}) - self.assertEqual(self.get(url='/?foo=%xx')['status'], 200) + assert self.get(url='/?foo=%xx')['status'] == 200 self.route_match({"arguments": {"foo": "%2A*"}}) - self.assertEqual(self.get(url='/?foo=*xx')['status'], 200) - self.assertEqual(self.get(url='/?foo=xx')['status'], 404) + assert self.get(url='/?foo=*xx')['status'] == 200 + assert self.get(url='/?foo=xx')['status'] == 404 self.route_match({"arguments": {"foo": "*%2A"}}) - self.assertEqual(self.get(url='/?foo=xx*')['status'], 200) - self.assertEqual(self.get(url='/?foo=xx*x')['status'], 404) + assert self.get(url='/?foo=xx*')['status'] == 200 + assert self.get(url='/?foo=xx*x')['status'] == 404 self.route_match({"arguments": {"foo": "1*2"}}) - self.assertEqual(self.get(url='/?foo=12')['status'], 200) - self.assertEqual(self.get(url='/?foo=1blah2')['status'], 200) - self.assertEqual(self.get(url='/?foo=1%2A2')['status'], 200) - self.assertEqual(self.get(url='/?foo=x12')['status'], 404) + assert self.get(url='/?foo=12')['status'] == 200 + assert self.get(url='/?foo=1blah2')['status'] == 200 + assert self.get(url='/?foo=1%2A2')['status'] == 200 + assert self.get(url='/?foo=x12')['status'] == 404 self.route_match({"arguments": {"foo": "bar*", "%25": "%25"}}) - self.assertEqual(self.get(url='/?foo=barxx&%=%')['status'], 200) - self.assertEqual(self.get(url='/?foo=barxx&x%=%')['status'], 404) + assert self.get(url='/?foo=barxx&%=%')['status'] == 200 + assert self.get(url='/?foo=barxx&x%=%')['status'] == 404 def test_routes_match_arguments_negative(self): self.route_match({"arguments": {"foo": "!%25"}}) - self.assertEqual(self.get(url='/?foo=blah')['status'], 200) - self.assertEqual(self.get(url='/?foo=%')['status'], 404) + assert self.get(url='/?foo=blah')['status'] == 200 + assert self.get(url='/?foo=%')['status'] == 404 self.route_match({"arguments": {"foo": "%21blah"}}) - self.assertEqual(self.get(url='/?foo=%21blah')['status'], 200) - self.assertEqual(self.get(url='/?foo=!blah')['status'], 200) - self.assertEqual(self.get(url='/?foo=bar')['status'], 404) + assert self.get(url='/?foo=%21blah')['status'] == 200 + assert self.get(url='/?foo=!blah')['status'] == 200 + assert self.get(url='/?foo=bar')['status'] == 404 self.route_match({"arguments": {"foo": "!!%21*a"}}) - self.assertEqual(self.get(url='/?foo=blah')['status'], 200) - self.assertEqual(self.get(url='/?foo=!blah')['status'], 200) - self.assertEqual(self.get(url='/?foo=!!a')['status'], 404) - self.assertEqual(self.get(url='/?foo=!!bla')['status'], 404) + assert self.get(url='/?foo=blah')['status'] == 200 + assert self.get(url='/?foo=!blah')['status'] == 200 + assert self.get(url='/?foo=!!a')['status'] == 404 + assert self.get(url='/?foo=!!bla')['status'] == 404 def test_routes_match_arguments_percent(self): self.route_match({"arguments": {"%25": "%25"}}) - self.assertEqual(self.get(url='/?%=%')['status'], 200) - self.assertEqual(self.get(url='/?%25=%25')['status'], 200) - self.assertEqual(self.get(url='/?%25=%')['status'], 200) + assert self.get(url='/?%=%')['status'] == 200 + assert self.get(url='/?%25=%25')['status'] == 200 + assert self.get(url='/?%25=%')['status'] == 200 self.route_match({"arguments": {"%251": "%252"}}) - self.assertEqual(self.get(url='/?%1=%2')['status'], 200) - self.assertEqual(self.get(url='/?%251=%252')['status'], 200) - self.assertEqual(self.get(url='/?%251=%2')['status'], 200) + assert self.get(url='/?%1=%2')['status'] == 200 + assert self.get(url='/?%251=%252')['status'] == 200 + assert self.get(url='/?%251=%2')['status'] == 200 self.route_match({"arguments": {"%25%21%251": "%25%24%252"}}) - self.assertEqual(self.get(url='/?%!%1=%$%2')['status'], 200) - self.assertEqual(self.get(url='/?%25!%251=%25$%252')['status'], 200) - self.assertEqual(self.get(url='/?%25!%1=%$%2')['status'], 200) + assert self.get(url='/?%!%1=%$%2')['status'] == 200 + assert self.get(url='/?%25!%251=%25$%252')['status'] == 200 + assert self.get(url='/?%25!%1=%$%2')['status'] == 200 def test_routes_match_arguments_ampersand(self): self.route_match({"arguments": {"foo": "&"}}) - self.assertEqual(self.get(url='/?foo=%26')['status'], 200) - self.assertEqual(self.get(url='/?foo=%26&')['status'], 200) - self.assertEqual(self.get(url='/?foo=%26%26')['status'], 404) - self.assertEqual(self.get(url='/?foo=&')['status'], 404) + assert self.get(url='/?foo=%26')['status'] == 200 + assert self.get(url='/?foo=%26&')['status'] == 200 + assert self.get(url='/?foo=%26%26')['status'] == 404 + assert self.get(url='/?foo=&')['status'] == 404 self.route_match({"arguments": {"&": ""}}) - self.assertEqual(self.get(url='/?%26=')['status'], 200) - self.assertEqual(self.get(url='/?%26=&')['status'], 200) - self.assertEqual(self.get(url='/?%26=%26')['status'], 404) - self.assertEqual(self.get(url='/?&=')['status'], 404) + assert self.get(url='/?%26=')['status'] == 200 + assert self.get(url='/?%26=&')['status'] == 200 + assert self.get(url='/?%26=%26')['status'] == 404 + assert self.get(url='/?&=')['status'] == 404 def test_routes_match_arguments_complex(self): self.route_match({"arguments": {"foo": ""}}) - self.assertEqual(self.get(url='/?foo')['status'], 200, 'complex') - self.assertEqual( - self.get(url='/?blah=blah&foo=')['status'], 200, 'complex 2' - ) - self.assertEqual( - self.get(url='/?&&&foo&&&')['status'], 200, 'complex 3' - ) - self.assertEqual( - self.get(url='/?foo&foo=bar&foo')['status'], 404, 'complex 4' - ) - self.assertEqual( - self.get(url='/?foo=&foo')['status'], 200, 'complex 5' - ) - self.assertEqual( - self.get(url='/?&=&foo&==&')['status'], 200, 'complex 6' - ) - self.assertEqual( - self.get(url='/?&=&bar&==&')['status'], 404, 'complex 7' - ) + assert self.get(url='/?foo')['status'] == 200, 'complex' + assert self.get(url='/?blah=blah&foo=')['status'] == 200, 'complex 2' + assert self.get(url='/?&&&foo&&&')['status'] == 200, 'complex 3' + assert self.get(url='/?foo&foo=bar&foo')['status'] == 404, 'complex 4' + assert self.get(url='/?foo=&foo')['status'] == 200, 'complex 5' + assert self.get(url='/?&=&foo&==&')['status'] == 200, 'complex 6' + assert self.get(url='/?&=&bar&==&')['status'] == 404, 'complex 7' def test_routes_match_arguments_multiple(self): self.route_match({"arguments": {"foo": "bar", "blah": "test"}}) - self.assertEqual(self.get()['status'], 404, 'multiple') - self.assertEqual( - self.get(url='/?foo=bar&blah=test')['status'], 200, 'multiple 2' - ) - self.assertEqual( - self.get(url='/?foo=bar&blah')['status'], 404, 'multiple 3' - ) - self.assertEqual( - self.get(url='/?foo=bar&blah=tes')['status'], 404, 'multiple 4' - ) - self.assertEqual( - self.get(url='/?foo=b%61r&bl%61h=t%65st')['status'], - 200, - 'multiple 5', - ) + assert self.get()['status'] == 404, 'multiple' + assert ( + self.get(url='/?foo=bar&blah=test')['status'] == 200 + ), 'multiple 2' + assert self.get(url='/?foo=bar&blah')['status'] == 404, 'multiple 3' + assert ( + self.get(url='/?foo=bar&blah=tes')['status'] == 404 + ), 'multiple 4' + assert ( + self.get(url='/?foo=b%61r&bl%61h=t%65st')['status'] == 200 + ), 'multiple 5' def test_routes_match_arguments_multiple_rules(self): self.route_match({"arguments": {"foo": ["bar", "blah"]}}) - self.assertEqual(self.get()['status'], 404, 'rules') - self.assertEqual(self.get(url='/?foo=bar')['status'], 200, 'rules 2') - self.assertEqual(self.get(url='/?foo=blah')['status'], 200, 'rules 3') - self.assertEqual( - self.get(url='/?foo=blah&foo=bar&foo=blah')['status'], - 200, - 'rules 4', - ) - self.assertEqual( - self.get(url='/?foo=blah&foo=bar&foo=')['status'], 404, 'rules 5' - ) + assert self.get()['status'] == 404, 'rules' + assert self.get(url='/?foo=bar')['status'] == 200, 'rules 2' + assert self.get(url='/?foo=blah')['status'] == 200, 'rules 3' + assert ( + self.get(url='/?foo=blah&foo=bar&foo=blah')['status'] == 200 + ), 'rules 4' + assert ( + self.get(url='/?foo=blah&foo=bar&foo=')['status'] == 404 + ), 'rules 5' def test_routes_match_arguments_array(self): self.route_match( @@ -1398,27 +1203,23 @@ class TestRouting(TestApplicationProto): } ) - self.assertEqual(self.get()['status'], 404, 'arr') - self.assertEqual(self.get(url='/?var1=val123')['status'], 200, 'arr 2') - self.assertEqual(self.get(url='/?var2=val2')['status'], 200, 'arr 3') - self.assertEqual(self.get(url='/?var3=bar')['status'], 200, 'arr 4') - self.assertEqual(self.get(url='/?var1=bar')['status'], 404, 'arr 5') - self.assertEqual( - self.get(url='/?var1=bar&var4=foo')['status'], 200, 'arr 6' - ) + assert self.get()['status'] == 404, 'arr' + assert self.get(url='/?var1=val123')['status'] == 200, 'arr 2' + assert self.get(url='/?var2=val2')['status'] == 200, 'arr 3' + assert self.get(url='/?var3=bar')['status'] == 200, 'arr 4' + assert self.get(url='/?var1=bar')['status'] == 404, 'arr 5' + assert self.get(url='/?var1=bar&var4=foo')['status'] == 200, 'arr 6' - self.assertIn( - 'success', - self.conf_delete('routes/0/match/arguments/1'), - 'match arguments array configure 2', - ) + assert 'success' in self.conf_delete( + 'routes/0/match/arguments/1' + ), 'match arguments array configure 2' - self.assertEqual(self.get(url='/?var2=val2')['status'], 404, 'arr 7') - self.assertEqual(self.get(url='/?var3=foo')['status'], 200, 'arr 8') + assert self.get(url='/?var2=val2')['status'] == 404, 'arr 7' + assert self.get(url='/?var3=foo')['status'] == 200, 'arr 8' def test_routes_match_arguments_invalid(self): # TODO remove it after controller fixed - self.skip_alerts.append(r'failed to apply new conf') + skip_alert(r'failed to apply new conf') self.route_match_invalid({"arguments": ["var"]}) self.route_match_invalid({"arguments": [{"var1": {}}]}) @@ -1434,7 +1235,7 @@ class TestRouting(TestApplicationProto): def test_routes_match_cookies(self): self.route_match({"cookies": {"foO": "bar"}}) - self.assertEqual(self.get()['status'], 404, 'cookie') + assert self.get()['status'] == 404, 'cookie' self.cookie('foO=bar', 200) self.cookie('foO=bar;1', 200) self.cookie(['foO=bar', 'blah=blah'], 200) @@ -1446,10 +1247,10 @@ class TestRouting(TestApplicationProto): def test_routes_match_cookies_empty(self): self.route_match({"cookies": {}}) - self.assertEqual(self.get()['status'], 200, 'cookies empty') + assert self.get()['status'] == 200, 'cookies empty' self.route_match({"cookies": []}) - self.assertEqual(self.get()['status'], 200, 'cookies empty 2') + assert self.get()['status'] == 200, 'cookies empty 2' def test_routes_match_cookies_invalid(self): self.route_match_invalid({"cookies": ["var"]}) @@ -1458,7 +1259,7 @@ class TestRouting(TestApplicationProto): def test_routes_match_cookies_multiple(self): self.route_match({"cookies": {"foo": "bar", "blah": "blah"}}) - self.assertEqual(self.get()['status'], 404, 'multiple') + assert self.get()['status'] == 404, 'multiple' self.cookie('foo=bar; blah=blah', 200) self.cookie(['foo=bar', 'blah=blah'], 200) self.cookie(['foo=bar; blah', 'blah'], 404) @@ -1474,12 +1275,12 @@ class TestRouting(TestApplicationProto): def test_routes_match_cookies_multiple_rules(self): self.route_match({"cookies": {"blah": ["test", "blah"]}}) - self.assertEqual(self.get()['status'], 404, 'multiple rules') + assert self.get()['status'] == 404, 'multiple rules' self.cookie('blah=test', 200) self.cookie('blah=blah', 200) self.cookie(['blah=blah', 'blah=test', 'blah=blah'], 200) self.cookie(['blah=blah; blah=test', 'blah=blah'], 200) - self.cookie(['blah=blah', 'blah'], 200) # invalid cookie + self.cookie(['blah=blah', 'blah'], 200) # invalid cookie def test_routes_match_cookies_array(self): self.route_match( @@ -1493,7 +1294,7 @@ class TestRouting(TestApplicationProto): } ) - self.assertEqual(self.get()['status'], 404, 'cookies array') + assert self.get()['status'] == 404, 'cookies array' self.cookie('var1=val123', 200) self.cookie('var2=val2', 200) self.cookie(' var2=val2 ', 200) @@ -1503,11 +1304,9 @@ class TestRouting(TestApplicationProto): self.cookie('var1=bar; var4=foo;', 200) self.cookie(['var1=bar', 'var4=foo'], 200) - self.assertIn( - 'success', - self.conf_delete('routes/0/match/cookies/1'), - 'match cookies array configure 2', - ) + assert 'success' in self.conf_delete( + 'routes/0/match/cookies/1' + ), 'match cookies array configure 2' self.cookie('var2=val2', 404) self.cookie('var3=foo', 200) @@ -1535,22 +1334,22 @@ class TestRouting(TestApplicationProto): sock2, port2 = sock_port() self.route_match({"source": "127.0.0.1:" + str(port)}) - self.assertEqual(self.get(sock=sock)['status'], 200, 'exact') - self.assertEqual(self.get(sock=sock2)['status'], 404, 'exact 2') + assert self.get(sock=sock)['status'] == 200, 'exact' + assert self.get(sock=sock2)['status'] == 404, 'exact 2' sock, port = sock_port() sock2, port2 = sock_port() self.route_match({"source": "!127.0.0.1:" + str(port)}) - self.assertEqual(self.get(sock=sock)['status'], 404, 'negative') - self.assertEqual(self.get(sock=sock2)['status'], 200, 'negative 2') + assert self.get(sock=sock)['status'] == 404, 'negative' + assert self.get(sock=sock2)['status'] == 200, 'negative 2' sock, port = sock_port() sock2, port2 = sock_port() self.route_match({"source": ["*:" + str(port), "!127.0.0.1"]}) - self.assertEqual(self.get(sock=sock)['status'], 404, 'negative 3') - self.assertEqual(self.get(sock=sock2)['status'], 404, 'negative 4') + assert self.get(sock=sock)['status'] == 404, 'negative 3' + assert self.get(sock=sock2)['status'] == 404, 'negative 4' sock, port = sock_port() sock2, port2 = sock_port() @@ -1558,8 +1357,8 @@ class TestRouting(TestApplicationProto): self.route_match( {"source": "127.0.0.1:" + str(port) + "-" + str(port)} ) - self.assertEqual(self.get(sock=sock)['status'], 200, 'range single') - self.assertEqual(self.get(sock=sock2)['status'], 404, 'range single 2') + assert self.get(sock=sock)['status'] == 200, 'range single' + assert self.get(sock=sock2)['status'] == 404, 'range single 2' socks = [ sock_port(), @@ -1578,11 +1377,11 @@ class TestRouting(TestApplicationProto): + str(socks[3][1]) # fourth port number } ) - self.assertEqual(self.get(sock=socks[0][0])['status'], 404, 'range') - self.assertEqual(self.get(sock=socks[1][0])['status'], 200, 'range 2') - self.assertEqual(self.get(sock=socks[2][0])['status'], 200, 'range 3') - self.assertEqual(self.get(sock=socks[3][0])['status'], 200, 'range 4') - self.assertEqual(self.get(sock=socks[4][0])['status'], 404, 'range 5') + assert self.get(sock=socks[0][0])['status'] == 404, 'range' + assert self.get(sock=socks[1][0])['status'] == 200, 'range 2' + assert self.get(sock=socks[2][0])['status'] == 200, 'range 3' + assert self.get(sock=socks[3][0])['status'] == 200, 'range 4' + assert self.get(sock=socks[4][0])['status'] == 404, 'range 5' socks = [ sock_port(), @@ -1599,218 +1398,194 @@ class TestRouting(TestApplicationProto): ] } ) - self.assertEqual(self.get(sock=socks[0][0])['status'], 200, 'array') - self.assertEqual(self.get(sock=socks[1][0])['status'], 404, 'array 2') - self.assertEqual(self.get(sock=socks[2][0])['status'], 200, 'array 3') + assert self.get(sock=socks[0][0])['status'] == 200, 'array' + assert self.get(sock=socks[1][0])['status'] == 404, 'array 2' + assert self.get(sock=socks[2][0])['status'] == 200, 'array 3' def test_routes_source_addr(self): - self.assertIn( - 'success', - self.conf( - { - "*:7080": {"pass": "routes"}, - "[::1]:7081": {"pass": "routes"}, - }, - 'listeners', - ), - 'source listeners configure', - ) + assert 'success' in self.conf( + {"*:7080": {"pass": "routes"}, "[::1]:7081": {"pass": "routes"},}, + 'listeners', + ), 'source listeners configure' def get_ipv6(): return self.get(sock_type='ipv6', port=7081) self.route_match({"source": "127.0.0.1"}) - self.assertEqual(self.get()['status'], 200, 'exact') - self.assertEqual(get_ipv6()['status'], 404, 'exact ipv6') + assert self.get()['status'] == 200, 'exact' + assert get_ipv6()['status'] == 404, 'exact ipv6' self.route_match({"source": ["127.0.0.1"]}) - self.assertEqual(self.get()['status'], 200, 'exact 2') - self.assertEqual(get_ipv6()['status'], 404, 'exact 2 ipv6') + assert self.get()['status'] == 200, 'exact 2' + assert get_ipv6()['status'] == 404, 'exact 2 ipv6' self.route_match({"source": "!127.0.0.1"}) - self.assertEqual(self.get()['status'], 404, 'exact neg') - self.assertEqual(get_ipv6()['status'], 200, 'exact neg ipv6') + assert self.get()['status'] == 404, 'exact neg' + assert get_ipv6()['status'] == 200, 'exact neg ipv6' self.route_match({"source": "127.0.0.2"}) - self.assertEqual(self.get()['status'], 404, 'exact 3') - self.assertEqual(get_ipv6()['status'], 404, 'exact 3 ipv6') + assert self.get()['status'] == 404, 'exact 3' + assert get_ipv6()['status'] == 404, 'exact 3 ipv6' self.route_match({"source": "127.0.0.1-127.0.0.1"}) - self.assertEqual(self.get()['status'], 200, 'range single') - self.assertEqual(get_ipv6()['status'], 404, 'range single ipv6') + assert self.get()['status'] == 200, 'range single' + assert get_ipv6()['status'] == 404, 'range single ipv6' self.route_match({"source": "127.0.0.2-127.0.0.2"}) - self.assertEqual(self.get()['status'], 404, 'range single 2') - self.assertEqual(get_ipv6()['status'], 404, 'range single 2 ipv6') + assert self.get()['status'] == 404, 'range single 2' + assert get_ipv6()['status'] == 404, 'range single 2 ipv6' self.route_match({"source": "127.0.0.2-127.0.0.3"}) - self.assertEqual(self.get()['status'], 404, 'range') - self.assertEqual(get_ipv6()['status'], 404, 'range ipv6') + assert self.get()['status'] == 404, 'range' + assert get_ipv6()['status'] == 404, 'range ipv6' self.route_match({"source": "127.0.0.1-127.0.0.2"}) - self.assertEqual(self.get()['status'], 200, 'range 2') - self.assertEqual(get_ipv6()['status'], 404, 'range 2 ipv6') + assert self.get()['status'] == 200, 'range 2' + assert get_ipv6()['status'] == 404, 'range 2 ipv6' self.route_match({"source": "127.0.0.0-127.0.0.2"}) - self.assertEqual(self.get()['status'], 200, 'range 3') - self.assertEqual(get_ipv6()['status'], 404, 'range 3 ipv6') + assert self.get()['status'] == 200, 'range 3' + assert get_ipv6()['status'] == 404, 'range 3 ipv6' self.route_match({"source": "127.0.0.0-127.0.0.1"}) - self.assertEqual(self.get()['status'], 200, 'range 4') - self.assertEqual(get_ipv6()['status'], 404, 'range 4 ipv6') + assert self.get()['status'] == 200, 'range 4' + assert get_ipv6()['status'] == 404, 'range 4 ipv6' self.route_match({"source": "126.0.0.0-127.0.0.0"}) - self.assertEqual(self.get()['status'], 404, 'range 5') - self.assertEqual(get_ipv6()['status'], 404, 'range 5 ipv6') + assert self.get()['status'] == 404, 'range 5' + assert get_ipv6()['status'] == 404, 'range 5 ipv6' self.route_match({"source": "126.126.126.126-127.0.0.2"}) - self.assertEqual(self.get()['status'], 200, 'range 6') - self.assertEqual(get_ipv6()['status'], 404, 'range 6 ipv6') + assert self.get()['status'] == 200, 'range 6' + assert get_ipv6()['status'] == 404, 'range 6 ipv6' def test_routes_source_ipv6(self): - self.assertIn( - 'success', - self.conf( - { - "[::1]:7080": {"pass": "routes"}, - "127.0.0.1:7081": {"pass": "routes"}, - }, - 'listeners', - ), - 'source listeners configure', - ) + assert 'success' in self.conf( + { + "[::1]:7080": {"pass": "routes"}, + "127.0.0.1:7081": {"pass": "routes"}, + }, + 'listeners', + ), 'source listeners configure' self.route_match({"source": "::1"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, 'exact') - self.assertEqual(self.get(port=7081)['status'], 404, 'exact ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, 'exact' + assert self.get(port=7081)['status'] == 404, 'exact ipv4' self.route_match({"source": ["::1"]}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, 'exact 2') - self.assertEqual(self.get(port=7081)['status'], 404, 'exact 2 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, 'exact 2' + assert self.get(port=7081)['status'] == 404, 'exact 2 ipv4' self.route_match({"source": "!::1"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 404, 'exact neg') - self.assertEqual(self.get(port=7081)['status'], 200, 'exact neg ipv4') + assert self.get(sock_type='ipv6')['status'] == 404, 'exact neg' + assert self.get(port=7081)['status'] == 200, 'exact neg ipv4' self.route_match({"source": "::2"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 404, 'exact 3') - self.assertEqual(self.get(port=7081)['status'], 404, 'exact 3 ipv4') + assert self.get(sock_type='ipv6')['status'] == 404, 'exact 3' + assert self.get(port=7081)['status'] == 404, 'exact 3 ipv4' self.route_match({"source": "::1-::1"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, 'range') - self.assertEqual(self.get(port=7081)['status'], 404, 'range ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, 'range' + assert self.get(port=7081)['status'] == 404, 'range ipv4' self.route_match({"source": "::2-::2"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 404, 'range 2') - self.assertEqual(self.get(port=7081)['status'], 404, 'range 2 ipv4') + assert self.get(sock_type='ipv6')['status'] == 404, 'range 2' + assert self.get(port=7081)['status'] == 404, 'range 2 ipv4' self.route_match({"source": "::2-::3"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 404, 'range 3') - self.assertEqual(self.get(port=7081)['status'], 404, 'range 3 ipv4') + assert self.get(sock_type='ipv6')['status'] == 404, 'range 3' + assert self.get(port=7081)['status'] == 404, 'range 3 ipv4' self.route_match({"source": "::1-::2"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, 'range 4') - self.assertEqual(self.get(port=7081)['status'], 404, 'range 4 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, 'range 4' + assert self.get(port=7081)['status'] == 404, 'range 4 ipv4' self.route_match({"source": "::0-::2"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, 'range 5') - self.assertEqual(self.get(port=7081)['status'], 404, 'range 5 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, 'range 5' + assert self.get(port=7081)['status'] == 404, 'range 5 ipv4' self.route_match({"source": "::0-::1"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, 'range 6') - self.assertEqual(self.get(port=7081)['status'], 404, 'range 6 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, 'range 6' + assert self.get(port=7081)['status'] == 404, 'range 6 ipv4' def test_routes_source_cidr(self): - self.assertIn( - 'success', - self.conf( - { - "*:7080": {"pass": "routes"}, - "[::1]:7081": {"pass": "routes"}, - }, - 'listeners', - ), - 'source listeners configure', - ) + assert 'success' in self.conf( + {"*:7080": {"pass": "routes"}, "[::1]:7081": {"pass": "routes"},}, + 'listeners', + ), 'source listeners configure' def get_ipv6(): return self.get(sock_type='ipv6', port=7081) self.route_match({"source": "127.0.0.1/32"}) - self.assertEqual(self.get()['status'], 200, '32') - self.assertEqual(get_ipv6()['status'], 404, '32 ipv6') + assert self.get()['status'] == 200, '32' + assert get_ipv6()['status'] == 404, '32 ipv6' self.route_match({"source": "127.0.0.0/32"}) - self.assertEqual(self.get()['status'], 404, '32 2') - self.assertEqual(get_ipv6()['status'], 404, '32 2 ipv6') + assert self.get()['status'] == 404, '32 2' + assert get_ipv6()['status'] == 404, '32 2 ipv6' self.route_match({"source": "127.0.0.0/31"}) - self.assertEqual(self.get()['status'], 200, '31') - self.assertEqual(get_ipv6()['status'], 404, '31 ipv6') + assert self.get()['status'] == 200, '31' + assert get_ipv6()['status'] == 404, '31 ipv6' self.route_match({"source": "0.0.0.0/1"}) - self.assertEqual(self.get()['status'], 200, '1') - self.assertEqual(get_ipv6()['status'], 404, '1 ipv6') + assert self.get()['status'] == 200, '1' + assert get_ipv6()['status'] == 404, '1 ipv6' self.route_match({"source": "0.0.0.0/0"}) - self.assertEqual(self.get()['status'], 200, '0') - self.assertEqual(get_ipv6()['status'], 404, '0 ipv6') + assert self.get()['status'] == 200, '0' + assert get_ipv6()['status'] == 404, '0 ipv6' def test_routes_source_cidr_ipv6(self): - self.assertIn( - 'success', - self.conf( - { - "[::1]:7080": {"pass": "routes"}, - "127.0.0.1:7081": {"pass": "routes"}, - }, - 'listeners', - ), - 'source listeners configure', - ) + assert 'success' in self.conf( + { + "[::1]:7080": {"pass": "routes"}, + "127.0.0.1:7081": {"pass": "routes"}, + }, + 'listeners', + ), 'source listeners configure' self.route_match({"source": "::1/128"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, '128') - self.assertEqual(self.get(port=7081)['status'], 404, '128 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, '128' + assert self.get(port=7081)['status'] == 404, '128 ipv4' self.route_match({"source": "::0/128"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 404, '128 2') - self.assertEqual(self.get(port=7081)['status'], 404, '128 ipv4') + assert self.get(sock_type='ipv6')['status'] == 404, '128 2' + assert self.get(port=7081)['status'] == 404, '128 ipv4' self.route_match({"source": "::0/127"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, '127') - self.assertEqual(self.get(port=7081)['status'], 404, '127 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, '127' + assert self.get(port=7081)['status'] == 404, '127 ipv4' self.route_match({"source": "::0/32"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, '32') - self.assertEqual(self.get(port=7081)['status'], 404, '32 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, '32' + assert self.get(port=7081)['status'] == 404, '32 ipv4' self.route_match({"source": "::0/1"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, '1') - self.assertEqual(self.get(port=7081)['status'], 404, '1 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, '1' + assert self.get(port=7081)['status'] == 404, '1 ipv4' self.route_match({"source": "::/0"}) - self.assertEqual(self.get(sock_type='ipv6')['status'], 200, '0') - self.assertEqual(self.get(port=7081)['status'], 404, '0 ipv4') + assert self.get(sock_type='ipv6')['status'] == 200, '0' + assert self.get(port=7081)['status'] == 404, '0 ipv4' def test_routes_source_unix(self): - addr = self.testdir + '/sock' + addr = self.temp_dir + '/sock' - self.assertIn( - 'success', - self.conf({"unix:" + addr: {"pass": "routes"}}, 'listeners'), - 'source listeners configure', - ) + assert 'success' in self.conf( + {"unix:" + addr: {"pass": "routes"}}, 'listeners' + ), 'source listeners configure' self.route_match({"source": "!0.0.0.0/0"}) - self.assertEqual( - self.get(sock_type='unix', addr=addr)['status'], 200, 'unix ipv4' - ) + assert ( + self.get(sock_type='unix', addr=addr)['status'] == 200 + ), 'unix ipv4' self.route_match({"source": "!::/0"}) - self.assertEqual( - self.get(sock_type='unix', addr=addr)['status'], 200, 'unix ipv6' - ) + assert ( + self.get(sock_type='unix', addr=addr)['status'] == 200 + ), 'unix ipv6' def test_routes_match_source(self): self.route_match({"source": "::"}) @@ -1863,7 +1638,7 @@ class TestRouting(TestApplicationProto): } ) self.route_match({"source": "*:0-65535"}) - self.assertEqual(self.get()['status'], 200, 'source any') + assert self.get()['status'] == 200, 'source any' def test_routes_match_source_invalid(self): self.route_match_invalid({"source": "127"}) @@ -1888,104 +1663,84 @@ class TestRouting(TestApplicationProto): self.route_match_invalid({"source": "*:65536"}) def test_routes_match_destination(self): - self.assertIn( - 'success', - self.conf( - {"*:7080": {"pass": "routes"}, "*:7081": {"pass": "routes"}}, - 'listeners', - ), - 'listeners configure', - ) + assert 'success' in self.conf( + {"*:7080": {"pass": "routes"}, "*:7081": {"pass": "routes"}}, + 'listeners', + ), 'listeners configure' self.route_match({"destination": "*:7080"}) - self.assertEqual(self.get()['status'], 200, 'dest') - self.assertEqual(self.get(port=7081)['status'], 404, 'dest 2') + assert self.get()['status'] == 200, 'dest' + assert self.get(port=7081)['status'] == 404, 'dest 2' self.route_match({"destination": ["127.0.0.1:7080"]}) - self.assertEqual(self.get()['status'], 200, 'dest 3') - self.assertEqual(self.get(port=7081)['status'], 404, 'dest 4') + assert self.get()['status'] == 200, 'dest 3' + assert self.get(port=7081)['status'] == 404, 'dest 4' self.route_match({"destination": "!*:7080"}) - self.assertEqual(self.get()['status'], 404, 'dest neg') - self.assertEqual(self.get(port=7081)['status'], 200, 'dest neg 2') + assert self.get()['status'] == 404, 'dest neg' + assert self.get(port=7081)['status'] == 200, 'dest neg 2' self.route_match({"destination": ['!*:7080', '!*:7081']}) - self.assertEqual(self.get()['status'], 404, 'dest neg 3') - self.assertEqual(self.get(port=7081)['status'], 404, 'dest neg 4') + assert self.get()['status'] == 404, 'dest neg 3' + assert self.get(port=7081)['status'] == 404, 'dest neg 4' self.route_match({"destination": ['!*:7081', '!*:7082']}) - self.assertEqual(self.get()['status'], 200, 'dest neg 5') + assert self.get()['status'] == 200, 'dest neg 5' self.route_match({"destination": ['*:7080', '!*:7080']}) - self.assertEqual(self.get()['status'], 404, 'dest neg 6') + assert self.get()['status'] == 404, 'dest neg 6' self.route_match( {"destination": ['127.0.0.1:7080', '*:7081', '!*:7080']} ) - self.assertEqual(self.get()['status'], 404, 'dest neg 7') - self.assertEqual(self.get(port=7081)['status'], 200, 'dest neg 8') + assert self.get()['status'] == 404, 'dest neg 7' + assert self.get(port=7081)['status'] == 200, 'dest neg 8' self.route_match({"destination": ['!*:7081', '!*:7082', '*:7083']}) - self.assertEqual(self.get()['status'], 404, 'dest neg 9') + assert self.get()['status'] == 404, 'dest neg 9' self.route_match( {"destination": ['*:7081', '!127.0.0.1:7080', '*:7080']} ) - self.assertEqual(self.get()['status'], 404, 'dest neg 10') - self.assertEqual(self.get(port=7081)['status'], 200, 'dest neg 11') + assert self.get()['status'] == 404, 'dest neg 10' + assert self.get(port=7081)['status'] == 200, 'dest neg 11' - self.assertIn( - 'success', - self.conf_delete('routes/0/match/destination/0'), - 'remove destination rule', - ) - self.assertEqual(self.get()['status'], 404, 'dest neg 12') - self.assertEqual(self.get(port=7081)['status'], 404, 'dest neg 13') + assert 'success' in self.conf_delete( + 'routes/0/match/destination/0' + ), 'remove destination rule' + assert self.get()['status'] == 404, 'dest neg 12' + assert self.get(port=7081)['status'] == 404, 'dest neg 13' - self.assertIn( - 'success', - self.conf_delete('routes/0/match/destination/0'), - 'remove destination rule 2', - ) - self.assertEqual(self.get()['status'], 200, 'dest neg 14') - self.assertEqual(self.get(port=7081)['status'], 404, 'dest neg 15') + assert 'success' in self.conf_delete( + 'routes/0/match/destination/0' + ), 'remove destination rule 2' + assert self.get()['status'] == 200, 'dest neg 14' + assert self.get(port=7081)['status'] == 404, 'dest neg 15' - self.assertIn( - 'success', - self.conf_post("\"!127.0.0.1\"", 'routes/0/match/destination'), - 'add destination rule', - ) - self.assertEqual(self.get()['status'], 404, 'dest neg 16') - self.assertEqual(self.get(port=7081)['status'], 404, 'dest neg 17') + assert 'success' in self.conf_post( + "\"!127.0.0.1\"", 'routes/0/match/destination' + ), 'add destination rule' + assert self.get()['status'] == 404, 'dest neg 16' + assert self.get(port=7081)['status'] == 404, 'dest neg 17' def test_routes_match_destination_proxy(self): - self.assertIn( - 'success', - self.conf( - { - "listeners": { - "*:7080": {"pass": "routes/first"}, - "*:7081": {"pass": "routes/second"}, - }, - "routes": { - "first": [ - {"action": {"proxy": "http://127.0.0.1:7081"}} - ], - "second": [ - { - "match": {"destination": ["127.0.0.1:7081"]}, - "action": {"return": 200}, - } - ], - }, - "applications": {}, - } - ), - 'proxy configure', - ) - - self.assertEqual(self.get()['status'], 200, 'proxy') - + assert 'success' in self.conf( + { + "listeners": { + "*:7080": {"pass": "routes/first"}, + "*:7081": {"pass": "routes/second"}, + }, + "routes": { + "first": [{"action": {"proxy": "http://127.0.0.1:7081"}}], + "second": [ + { + "match": {"destination": ["127.0.0.1:7081"]}, + "action": {"return": 200}, + } + ], + }, + "applications": {}, + } + ), 'proxy configure' -if __name__ == '__main__': - TestRouting.main() + assert self.get()['status'] == 200, 'proxy' |