summaryrefslogtreecommitdiffhomepage
path: root/test/test_configuration.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_configuration.py')
-rw-r--r--test/test_configuration.py754
1 files changed, 390 insertions, 364 deletions
diff --git a/test/test_configuration.py b/test/test_configuration.py
index e3ddc891..19a2a1a5 100644
--- a/test/test_configuration.py
+++ b/test/test_configuration.py
@@ -1,439 +1,465 @@
import socket
import pytest
-from unit.control import TestControl
-from unit.option import option
+from unit.control import Control
+prerequisites = {'modules': {'python': 'any'}}
-class TestConfiguration(TestControl):
- prerequisites = {'modules': {'python': 'any'}}
+client = Control()
- def try_addr(self, addr):
- return self.conf(
- {
- "listeners": {addr: {"pass": "routes"}},
- "routes": [{"action": {"return": 200}}],
- "applications": {},
- }
- )
- def test_json_empty(self):
- assert 'error' in self.conf(''), 'empty'
+def try_addr(addr):
+ return client.conf(
+ {
+ "listeners": {addr: {"pass": "routes"}},
+ "routes": [{"action": {"return": 200}}],
+ "applications": {},
+ }
+ )
- def test_json_leading_zero(self):
- assert 'error' in self.conf('00'), 'leading zero'
- def test_json_unicode(self):
- assert 'success' in self.conf(
- u"""
- {
- "ap\u0070": {
- "type": "\u0070ython",
- "processes": { "spare": 0 },
- "path": "\u002Fapp",
- "module": "wsgi"
- }
+def test_json_empty():
+ assert 'error' in client.conf(''), 'empty'
+
+
+def test_json_leading_zero():
+ assert 'error' in client.conf('00'), 'leading zero'
+
+
+def test_json_unicode():
+ assert 'success' in client.conf(
+ """
+ {
+ "ap\u0070": {
+ "type": "\u0070ython",
+ "processes": { "spare": 0 },
+ "path": "\u002Fapp",
+ "module": "wsgi"
}
- """,
- 'applications',
- ), 'unicode'
+ }
+ """,
+ 'applications',
+ ), 'unicode'
+
+ assert client.conf_get('applications') == {
+ "app": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "module": "wsgi",
+ }
+ }, 'unicode get'
- assert self.conf_get('applications') == {
- "app": {
+
+def test_json_unicode_2():
+ assert 'success' in client.conf(
+ {
+ "приложение": {
"type": "python",
"processes": {"spare": 0},
"path": "/app",
"module": "wsgi",
}
- }, 'unicode get'
+ },
+ 'applications',
+ ), 'unicode 2'
- def test_json_unicode_2(self):
- assert 'success' in self.conf(
- {
- "приложение": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "module": "wsgi",
- }
- },
- 'applications',
- ), 'unicode 2'
+ assert 'приложение' in client.conf_get('applications')
- assert 'приложение' in self.conf_get('applications'), 'unicode 2 get'
- def test_json_unicode_number(self):
- assert 'success' in self.conf(
- u"""
- {
- "app": {
- "type": "python",
- "processes": { "spare": \u0030 },
- "path": "/app",
- "module": "wsgi"
- }
+def test_json_unicode_number():
+ assert 'success' in client.conf(
+ """
+ {
+ "app": {
+ "type": "python",
+ "processes": { "spare": \u0030 },
+ "path": "/app",
+ "module": "wsgi"
}
- """,
- 'applications',
- ), 'unicode number'
+ }
+ """,
+ 'applications',
+ ), 'unicode number'
- def test_json_utf8_bom(self):
- assert 'success' in self.conf(
- b"""\xEF\xBB\xBF
- {
- "app": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "module": "wsgi"
- }
- }
- """,
- 'applications',
- ), 'UTF-8 BOM'
-
- def test_json_comment_single_line(self):
- assert 'success' in self.conf(
- b"""
- // this is bridge
- {
- "//app": {
- "type": "python", // end line
- "processes": {"spare": 0},
- // inside of block
- "path": "/app",
- "module": "wsgi"
- }
- // double //
+
+def test_json_utf8_bom():
+ assert 'success' in client.conf(
+ b"""\xEF\xBB\xBF
+ {
+ "app": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "module": "wsgi"
}
- // end of json \xEF\t
- """,
- 'applications',
- ), 'single line comments'
-
- def test_json_comment_multi_line(self):
- assert 'success' in self.conf(
- b"""
- /* this is bridge */
- {
- "/*app": {
- /**
- * multiple lines
- **/
- "type": "python",
- "processes": /* inline */ {"spare": 0},
- "path": "/app",
- "module": "wsgi"
- /*
- // end of block */
- }
- /* blah * / blah /* blah */
+ }
+ """,
+ 'applications',
+ ), 'UTF-8 BOM'
+
+
+def test_json_comment_single_line():
+ assert 'success' in client.conf(
+ b"""
+ // this is bridge
+ {
+ "//app": {
+ "type": "python", // end line
+ "processes": {"spare": 0},
+ // inside of block
+ "path": "/app",
+ "module": "wsgi"
}
- /* end of json \xEF\t\b */
- """,
- 'applications',
- ), 'multi line comments'
-
- def test_json_comment_invalid(self):
- assert 'error' in self.conf(b'/{}', 'applications'), 'slash'
- assert 'error' in self.conf(b'//{}', 'applications'), 'comment'
- assert 'error' in self.conf(b'{} /', 'applications'), 'slash end'
- assert 'error' in self.conf(b'/*{}', 'applications'), 'slash star'
- assert 'error' in self.conf(b'{} /*', 'applications'), 'slash star end'
-
- def test_applications_open_brace(self):
- assert 'error' in self.conf('{', 'applications'), 'open brace'
-
- def test_applications_string(self):
- assert 'error' in self.conf('"{}"', 'applications'), 'string'
-
- @pytest.mark.skip('not yet, unsafe')
- def test_applications_type_only(self):
- assert 'error' in self.conf(
- {"app": {"type": "python"}}, 'applications'
- ), 'type only'
-
- def test_applications_miss_quote(self):
- assert 'error' in self.conf(
- """
- {
- app": {
- "type": "python",
- "processes": { "spare": 0 },
- "path": "/app",
- "module": "wsgi"
- }
+ // double //
+ }
+ // end of json \xEF\t
+ """,
+ 'applications',
+ ), 'single line comments'
+
+
+def test_json_comment_multi_line():
+ assert 'success' in client.conf(
+ b"""
+ /* this is bridge */
+ {
+ "/*app": {
+ /**
+ * multiple lines
+ **/
+ "type": "python",
+ "processes": /* inline */ {"spare": 0},
+ "path": "/app",
+ "module": "wsgi"
+ /*
+ // end of block */
}
- """,
- 'applications',
- ), 'miss quote'
+ /* blah * / blah /* blah */
+ }
+ /* end of json \xEF\t\b */
+ """,
+ 'applications',
+ ), 'multi line comments'
- def test_applications_miss_colon(self):
- assert 'error' in self.conf(
- """
- {
- "app" {
- "type": "python",
- "processes": { "spare": 0 },
- "path": "/app",
- "module": "wsgi"
- }
+
+def test_json_comment_invalid():
+ assert 'error' in client.conf(b'/{}', 'applications'), 'slash'
+ assert 'error' in client.conf(b'//{}', 'applications'), 'comment'
+ assert 'error' in client.conf(b'{} /', 'applications'), 'slash end'
+ assert 'error' in client.conf(b'/*{}', 'applications'), 'slash star'
+ assert 'error' in client.conf(b'{} /*', 'applications'), 'slash star end'
+
+
+def test_applications_open_brace():
+ assert 'error' in client.conf('{', 'applications'), 'open brace'
+
+
+def test_applications_string():
+ assert 'error' in client.conf('"{}"', 'applications'), 'string'
+
+
+@pytest.mark.skip('not yet, unsafe')
+def test_applications_type_only():
+ assert 'error' in client.conf(
+ {"app": {"type": "python"}}, 'applications'
+ ), 'type only'
+
+
+def test_applications_miss_quote():
+ assert 'error' in client.conf(
+ """
+ {
+ app": {
+ "type": "python",
+ "processes": { "spare": 0 },
+ "path": "/app",
+ "module": "wsgi"
}
- """,
- 'applications',
- ), 'miss colon'
+ }
+ """,
+ 'applications',
+ ), 'miss quote'
- def test_applications_miss_comma(self):
- assert 'error' in self.conf(
- """
- {
- "app": {
- "type": "python"
- "processes": { "spare": 0 },
- "path": "/app",
- "module": "wsgi"
- }
+
+def test_applications_miss_colon():
+ assert 'error' in client.conf(
+ """
+ {
+ "app" {
+ "type": "python",
+ "processes": { "spare": 0 },
+ "path": "/app",
+ "module": "wsgi"
}
- """,
- 'applications',
- ), 'miss comma'
+ }
+ """,
+ 'applications',
+ ), 'miss colon'
- def test_applications_skip_spaces(self):
- assert 'success' in self.conf(
- b'{ \n\r\t}', 'applications'
- ), 'skip spaces'
- def test_applications_relative_path(self):
- assert 'success' in self.conf(
- {
- "app": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "../app",
- "module": "wsgi",
- }
- },
- 'applications',
- ), 'relative path'
+def test_applications_miss_comma():
+ assert 'error' in client.conf(
+ """
+ {
+ "app": {
+ "type": "python"
+ "processes": { "spare": 0 },
+ "path": "/app",
+ "module": "wsgi"
+ }
+ }
+ """,
+ 'applications',
+ ), 'miss comma'
- @pytest.mark.skip('not yet, unsafe')
- def test_listeners_empty(self):
- assert 'error' in self.conf(
- {"*:7080": {}}, 'listeners'
- ), 'listener empty'
- def test_listeners_no_app(self):
- assert 'error' in self.conf(
- {"*:7080": {"pass": "applications/app"}}, 'listeners'
- ), 'listeners no app'
+def test_applications_skip_spaces():
+ assert 'success' in client.conf(b'{ \n\r\t}', 'applications'), 'skip spaces'
- def test_listeners_unix_abstract(self):
- if option.system != 'Linux':
- assert 'error' in self.try_addr("unix:@sock"), 'abstract at'
- pytest.skip('not yet')
+def test_applications_relative_path():
+ assert 'success' in client.conf(
+ {
+ "app": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "../app",
+ "module": "wsgi",
+ }
+ },
+ 'applications',
+ ), 'relative path'
- assert 'error' in self.try_addr("unix:\0soc"), 'abstract \0'
- assert 'error' in self.try_addr("unix:\u0000soc"), 'abstract \0 unicode'
- def test_listeners_addr(self):
- assert 'success' in self.try_addr("*:7080"), 'wildcard'
- assert 'success' in self.try_addr("127.0.0.1:7081"), 'explicit'
- assert 'success' in self.try_addr("[::1]:7082"), 'explicit ipv6'
+@pytest.mark.skip('not yet, unsafe')
+def test_listeners_empty():
+ assert 'error' in client.conf({"*:7080": {}}, 'listeners'), 'listener empty'
- def test_listeners_addr_error(self):
- assert 'error' in self.try_addr("127.0.0.1"), 'no port'
- def test_listeners_addr_error_2(self, skip_alert):
- skip_alert(r'bind.*failed', r'failed to apply new conf')
+def test_listeners_no_app():
+ assert 'error' in client.conf(
+ {"*:7080": {"pass": "applications/app"}}, 'listeners'
+ ), 'listeners no app'
- assert 'error' in self.try_addr(
- "[f607:7403:1e4b:6c66:33b2:843f:2517:da27]:7080"
- )
- def test_listeners_port_release(self):
- for i in range(10):
- fail = False
- with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
- s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+def test_listeners_unix_abstract(system):
+ if system != 'Linux':
+ assert 'error' in try_addr("unix:@sock"), 'abstract at'
- self.conf(
- {
- "listeners": {"127.0.0.1:7080": {"pass": "routes"}},
- "routes": [],
- }
- )
+ pytest.skip('not yet')
- resp = self.conf({"listeners": {}, "applications": {}})
+ assert 'error' in try_addr("unix:\0soc"), 'abstract \0'
+ assert 'error' in try_addr("unix:\u0000soc"), 'abstract \0 unicode'
- try:
- s.bind(('127.0.0.1', 7080))
- s.listen()
- except OSError:
- fail = True
+def test_listeners_addr():
+ assert 'success' in try_addr("*:7080"), 'wildcard'
+ assert 'success' in try_addr("127.0.0.1:7081"), 'explicit'
+ assert 'success' in try_addr("[::1]:7082"), 'explicit ipv6'
- if fail:
- pytest.fail('cannot bind or listen to the address')
- assert 'success' in resp, 'port release'
+def test_listeners_addr_error():
+ assert 'error' in try_addr("127.0.0.1"), 'no port'
- def test_json_application_name_large(self):
- name = "X" * 1024 * 1024
- assert 'success' in self.conf(
- {
- "listeners": {"*:7080": {"pass": f"applications/{name}"}},
- "applications": {
- name: {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "module": "wsgi",
- }
- },
- }
- )
+def test_listeners_addr_error_2(skip_alert):
+ skip_alert(r'bind.*failed', r'failed to apply new conf')
- @pytest.mark.skip('not yet')
- def test_json_application_many(self):
- apps = 999
+ assert 'error' in try_addr("[f607:7403:1e4b:6c66:33b2:843f:2517:da27]:7080")
- conf = {
- "applications": {
- f"app-{a}": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "module": "wsgi",
- }
- for a in range(apps)
- },
- "listeners": {
- f"*:{(7000 + a)}": {"pass": f"applications/app-{a}"}
- for a in range(apps)
- },
- }
- assert 'success' in self.conf(conf)
+def test_listeners_port_release():
+ for _ in range(10):
+ fail = False
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- def test_json_application_python_prefix(self):
- conf = {
- "applications": {
- "sub-app": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "module": "wsgi",
- "prefix": "/app",
- }
- },
- "listeners": {"*:7080": {"pass": "routes"}},
- "routes": [
+ client.conf(
{
- "match": {"uri": "/app/*"},
- "action": {"pass": "applications/sub-app"},
+ "listeners": {"127.0.0.1:7080": {"pass": "routes"}},
+ "routes": [],
}
- ],
- }
+ )
- assert 'success' in self.conf(conf)
+ resp = client.conf({"listeners": {}, "applications": {}})
- def test_json_application_prefix_target(self):
- conf = {
- "applications": {
- "sub-app": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "targets": {
- "foo": {"module": "foo.wsgi", "prefix": "/app"},
- "bar": {
- "module": "bar.wsgi",
- "callable": "bar",
- "prefix": "/api",
- },
- },
- }
- },
- "listeners": {"*:7080": {"pass": "routes"}},
- "routes": [
- {
- "match": {"uri": "/app/*"},
- "action": {"pass": "applications/sub-app/foo"},
- },
- {
- "match": {"uri": "/api/*"},
- "action": {"pass": "applications/sub-app/bar"},
- },
- ],
- }
+ try:
+ s.bind(('127.0.0.1', 7080))
+ s.listen()
- assert 'success' in self.conf(conf)
+ except OSError:
+ fail = True
- def test_json_application_invalid_python_prefix(self):
- conf = {
- "applications": {
- "sub-app": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "module": "wsgi",
- "prefix": "app",
- }
- },
- "listeners": {"*:7080": {"pass": "applications/sub-app"}},
- }
+ if fail:
+ pytest.fail('cannot bind or listen to the address')
- assert 'error' in self.conf(conf)
+ assert 'success' in resp, 'port release'
- def test_json_application_empty_python_prefix(self):
- conf = {
- "applications": {
- "sub-app": {
- "type": "python",
- "processes": {"spare": 0},
- "path": "/app",
- "module": "wsgi",
- "prefix": "",
- }
- },
- "listeners": {"*:7080": {"pass": "applications/sub-app"}},
- }
- assert 'error' in self.conf(conf)
+def test_json_application_name_large():
+ name = "X" * 1024 * 1024
- def test_json_application_many2(self):
- conf = {
+ assert 'success' in client.conf(
+ {
+ "listeners": {"*:7080": {"pass": f"applications/{name}"}},
"applications": {
- f"app-{a}": {
+ name: {
"type": "python",
"processes": {"spare": 0},
"path": "/app",
"module": "wsgi",
}
- # Larger number of applications can cause test fail with default
- # open files limit due to the lack of file descriptors.
- for a in range(100)
},
- "listeners": {"*:7080": {"pass": "applications/app-1"}},
}
+ )
+
+
+@pytest.mark.skip('not yet')
+def test_json_application_many():
+ apps = 999
+
+ conf = {
+ "applications": {
+ f"app-{a}": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "module": "wsgi",
+ }
+ for a in range(apps)
+ },
+ "listeners": {
+ f"*:{(7000 + a)}": {"pass": f"applications/app-{a}"}
+ for a in range(apps)
+ },
+ }
- assert 'success' in self.conf(conf)
+ assert 'success' in client.conf(conf)
- def test_unprivileged_user_error(self, is_su, skip_alert):
- skip_alert(r'cannot set user "root"', r'failed to apply new conf')
- if is_su:
- pytest.skip('unprivileged tests')
- assert 'error' in self.conf(
+def test_json_application_python_prefix():
+ conf = {
+ "applications": {
+ "sub-app": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "module": "wsgi",
+ "prefix": "/app",
+ }
+ },
+ "listeners": {"*:7080": {"pass": "routes"}},
+ "routes": [
{
- "app": {
- "type": "external",
- "processes": 1,
- "executable": "/app",
- "user": "root",
- }
+ "match": {"uri": "/app/*"},
+ "action": {"pass": "applications/sub-app"},
+ }
+ ],
+ }
+
+ assert 'success' in client.conf(conf)
+
+
+def test_json_application_prefix_target():
+ conf = {
+ "applications": {
+ "sub-app": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "targets": {
+ "foo": {"module": "foo.wsgi", "prefix": "/app"},
+ "bar": {
+ "module": "bar.wsgi",
+ "callable": "bar",
+ "prefix": "/api",
+ },
+ },
+ }
+ },
+ "listeners": {"*:7080": {"pass": "routes"}},
+ "routes": [
+ {
+ "match": {"uri": "/app/*"},
+ "action": {"pass": "applications/sub-app/foo"},
+ },
+ {
+ "match": {"uri": "/api/*"},
+ "action": {"pass": "applications/sub-app/bar"},
},
- 'applications',
- ), 'setting user'
+ ],
+ }
+
+ assert 'success' in client.conf(conf)
+
+
+def test_json_application_invalid_python_prefix():
+ conf = {
+ "applications": {
+ "sub-app": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "module": "wsgi",
+ "prefix": "app",
+ }
+ },
+ "listeners": {"*:7080": {"pass": "applications/sub-app"}},
+ }
+
+ assert 'error' in client.conf(conf)
+
+
+def test_json_application_empty_python_prefix():
+ conf = {
+ "applications": {
+ "sub-app": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "module": "wsgi",
+ "prefix": "",
+ }
+ },
+ "listeners": {"*:7080": {"pass": "applications/sub-app"}},
+ }
+
+ assert 'error' in client.conf(conf)
+
+
+def test_json_application_many2():
+ conf = {
+ "applications": {
+ f"app-{a}": {
+ "type": "python",
+ "processes": {"spare": 0},
+ "path": "/app",
+ "module": "wsgi",
+ }
+ # Larger number of applications can cause test fail with default
+ # open files limit due to the lack of file descriptors.
+ for a in range(100)
+ },
+ "listeners": {"*:7080": {"pass": "applications/app-1"}},
+ }
+
+ assert 'success' in client.conf(conf)
+
+
+def test_unprivileged_user_error(require, skip_alert):
+ require({'privileged_user': False})
+
+ skip_alert(r'cannot set user "root"', r'failed to apply new conf')
+
+ assert 'error' in client.conf(
+ {
+ "app": {
+ "type": "external",
+ "processes": 1,
+ "executable": "/app",
+ "user": "root",
+ }
+ },
+ 'applications',
+ ), 'setting user'
generated by cgit (git 2.34.1) at 2024-11-16 19:57:58 +0000