diff options
Diffstat (limited to '')
| -rw-r--r-- | test/test_configuration.py | 292 | ||||
| -rw-r--r-- | test/test_php_basic.py | 332 | ||||
| -rw-r--r-- | test/test_python_application.py | 180 | ||||
| -rw-r--r-- | test/test_python_atexit.py | 130 | ||||
| -rw-r--r-- | test/test_python_basic.py | 332 | ||||
| -rw-r--r-- | test/unit.py | 472 |
6 files changed, 869 insertions, 869 deletions
diff --git a/test/test_configuration.py b/test/test_configuration.py index 8a3612a8..f10a5e8b 100644 --- a/test/test_configuration.py +++ b/test/test_configuration.py @@ -1,146 +1,146 @@ -import unittest
-import unit
-
-class TestUnitConfiguration(unit.TestUnitControl):
-
- def setUpClass():
- unit.TestUnit().check_modules('python')
-
- def test_json(self):
- self.assertIn('error', self.put('/', '00'), 'leading zero')
-
- def test_json_applications(self):
- self.assertIn('error', self.put('/applications', '"{}"'),
- 'applications string')
- self.assertIn('error', self.put('/applications', '{'),
- 'applications miss brace')
-
- self.assertTry('assertIn', 'negative workers', 'error',
- self.put('/applications', """
- {
- "app": {
- "type": "python",
- "workers": -1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """))
-
- self.assertTry('assertIn', 'application type only', 'error',
- self.put('/applications', """
- {
- "app": {
- "type": "python"
- }
- }
- """))
-
- self.assertIn('error', self.put('/applications', """
- {
- app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """), 'applications miss quote')
-
- self.assertIn('error', self.put('/applications', """
- {
- "app" {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """), 'applications miss colon')
-
- self.assertIn('error', self.put('/applications', """
- {
- "app": {
- "type": "python"
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """), 'applications miss comma')
-
- self.assertIn('success', self.put('/applications', b'{ \n\r\t}'),
- 'skip space')
-
- self.assertIn('success', self.put('/applications', """
- {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "../app",
- "module": "wsgi"
- }
- }
- """), 'relative path')
-
- self.assertIn('success', self.put('/applications', b"""
- {
- "ap\u0070": {
- "type": "\u0070ython",
- "workers": 1,
- "path": "\u002Fapp",
- "module": "wsgi"
- }
- }
- """), 'unicode')
-
- self.assertIn('success', self.put('/applications', """
- {
- "приложение": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """), 'unicode 2')
-
- self.assertIn('error', self.put('/applications', b"""
- {
- "app": {
- "type": "python",
- "workers": \u0031,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """), 'unicode number')
-
- def test_json_listeners(self):
- self.assertTry('assertIn', 'listener empty', 'error',
- self.put('/listeners', '{"*:7080":{}}'))
- self.assertIn('error', self.put('/listeners',
- '{"*:7080":{"application":"app"}}'), 'listeners no app')
-
- self.put('/applications', """
- {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """)
-
- self.assertIn('success', self.put('/listeners',
- '{"*:7080":{"application":"app"}}'), 'listeners wildcard')
- self.assertIn('success', self.put('/listeners',
- '{"127.0.0.1:7081":{"application":"app"}}'), 'listeners explicit')
- self.assertIn('success', self.put('/listeners',
- '{"[::1]:7082":{"application":"app"}}'), 'listeners explicit ipv6')
- self.assertIn('error', self.put('/listeners',
- '{"127.0.0.1":{"application":"app"}}'), 'listeners no port')
-
-if __name__ == '__main__':
- unittest.main()
+import unittest +import unit + +class TestUnitConfiguration(unit.TestUnitControl): + + def setUpClass(): + unit.TestUnit().check_modules('python') + + def test_json(self): + self.assertIn('error', self.put('/', '00'), 'leading zero') + + def test_json_applications(self): + self.assertIn('error', self.put('/applications', '"{}"'), + 'applications string') + self.assertIn('error', self.put('/applications', '{'), + 'applications miss brace') + + self.assertTry('assertIn', 'negative workers', 'error', + self.put('/applications', """ + { + "app": { + "type": "python", + "workers": -1, + "path": "/app", + "module": "wsgi" + } + } + """)) + + self.assertTry('assertIn', 'application type only', 'error', + self.put('/applications', """ + { + "app": { + "type": "python" + } + } + """)) + + self.assertIn('error', self.put('/applications', """ + { + app": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + """), 'applications miss quote') + + self.assertIn('error', self.put('/applications', """ + { + "app" { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + """), 'applications miss colon') + + self.assertIn('error', self.put('/applications', """ + { + "app": { + "type": "python" + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + """), 'applications miss comma') + + self.assertIn('success', self.put('/applications', b'{ \n\r\t}'), + 'skip space') + + self.assertIn('success', self.put('/applications', """ + { + "app": { + "type": "python", + "workers": 1, + "path": "../app", + "module": "wsgi" + } + } + """), 'relative path') + + self.assertIn('success', self.put('/applications', b""" + { + "ap\u0070": { + "type": "\u0070ython", + "workers": 1, + "path": "\u002Fapp", + "module": "wsgi" + } + } + """), 'unicode') + + self.assertIn('success', self.put('/applications', """ + { + "приложение": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + """), 'unicode 2') + + self.assertIn('error', self.put('/applications', b""" + { + "app": { + "type": "python", + "workers": \u0031, + "path": "/app", + "module": "wsgi" + } + } + """), 'unicode number') + + def test_json_listeners(self): + self.assertTry('assertIn', 'listener empty', 'error', + self.put('/listeners', '{"*:7080":{}}')) + self.assertIn('error', self.put('/listeners', + '{"*:7080":{"application":"app"}}'), 'listeners no app') + + self.put('/applications', """ + { + "app": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + """) + + self.assertIn('success', self.put('/listeners', + '{"*:7080":{"application":"app"}}'), 'listeners wildcard') + self.assertIn('success', self.put('/listeners', + '{"127.0.0.1:7081":{"application":"app"}}'), 'listeners explicit') + self.assertIn('success', self.put('/listeners', + '{"[::1]:7082":{"application":"app"}}'), 'listeners explicit ipv6') + self.assertIn('error', self.put('/listeners', + '{"127.0.0.1":{"application":"app"}}'), 'listeners no port') + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_php_basic.py b/test/test_php_basic.py index fa1484a3..d37f92f5 100644 --- a/test/test_php_basic.py +++ b/test/test_php_basic.py @@ -1,166 +1,166 @@ -import unittest
-import unit
-
-class TestUnitBasic(unit.TestUnitControl):
-
- def setUpClass():
- unit.TestUnit().check_modules('php')
-
- def test_php_get(self):
- resp = self.get()
- self.assertEqual(resp, {'listeners': {}, 'applications': {}}, 'empty')
- self.assertEqual(self.get('/listeners'), {}, 'empty listeners prefix')
- self.assertEqual(self.get('/applications'), {},
- 'empty applications prefix')
-
- self.put('/applications', """
- {
- "app": {
- "type": "php",
- "workers": 1,
- "root": "/app",
- "index": "index.php"
- }
- }
- """)
-
- resp = self.get()
-
- self.assertEqual(resp['listeners'], {}, 'php empty listeners')
- self.assertEqual(resp['applications'],
- {
- "app": {
- "type": "php",
- "workers": 1,
- "root": "/app",
- "index": "index.php"
- }
- },
- 'php applications')
-
- self.assertEqual(self.get('/applications'),
- {
- "app": {
- "type": "php",
- "workers": 1,
- "root": "/app",
- "index": "index.php"
- }
- },
- 'php applications prefix')
-
- self.assertEqual(self.get('/applications/app'),
- {
- "type": "php",
- "workers": 1,
- "root": "/app",
- "index": "index.php"
- },
- 'php applications prefix 2')
-
- self.assertEqual(self.get('/applications/app/type'), 'php',
- 'php applications type')
- self.assertEqual(self.get('/applications/app/workers'), 1,
- 'php applications workers')
-
- self.put('/listeners', '{"*:7080":{"application":"app"}}')
-
- self.assertEqual(self.get()['listeners'],
- {"*:7080":{"application":"app"}}, 'php listeners')
- self.assertEqual(self.get('/listeners'),
- {"*:7080":{"application":"app"}}, 'php listeners prefix')
- self.assertEqual(self.get('/listeners/*:7080'),
- {"application":"app"}, 'php listeners prefix 2')
- self.assertEqual(self.get('/listeners/*:7080/application'), 'app',
- 'php listeners application')
-
- def test_php_put(self):
- self.put('/', """
- {
- "listeners": {
- "*:7080": {
- "application": "app"
- }
- },
- "applications": {
- "app": {
- "type": "php",
- "workers": 1,
- "root": "/app",
- "index": "index.php"
- }
- }
- }
- """)
-
- resp = self.get()
-
- self.assertEqual(resp['listeners'], {"*:7080":{"application":"app"}},
- 'put listeners')
-
- self.assertEqual(resp['applications'],
- {
- "app": {
- "type": "php",
- "workers": 1,
- "root": "/app",
- "index": "index.php"
- }
- },
- 'put applications')
-
- self.put('/listeners', '{"*:7081":{"application":"app"}}')
- self.assertEqual(self.get('/listeners'),
- {"*:7081": {"application":"app"}}, 'put listeners prefix')
-
- self.put('/listeners/*:7082', '{"application":"app"}')
-
- self.assertEqual(self.get('/listeners'),
- {
- "*:7081": {
- "application": "app"
- },
- "*:7082": {
- "application": "app"
- }
- },
- 'put listeners prefix 3')
-
- self.put('/applications/app/workers', '30')
- self.assertEqual(self.get('/applications/app/workers'), 30,
- 'put applications workers')
-
- self.put('/applications/app/root', '"/www"')
- self.assertEqual(self.get('/applications/app/root'), '/www',
- 'put applications root')
-
- def test_php_delete(self):
- self.put('/', """
- {
- "listeners": {
- "*:7080": {
- "application": "app"
- }
- },
- "applications": {
- "app": {
- "type": "php",
- "workers": 1,
- "root": "/app",
- "index": "index.php"
- }
- }
- }
- """)
-
- self.assertIn('error', self.delete('/applications/app'),
- 'delete app before listener')
- self.assertIn('success', self.delete('/listeners/*:7080'),
- 'delete listener')
- self.assertIn('success', self.delete('/applications/app'),
- 'delete app after listener')
- self.assertIn('error', self.delete('/applications/app'),
- 'delete app again')
-
-if __name__ == '__main__':
- unittest.main()
+import unittest +import unit + +class TestUnitBasic(unit.TestUnitControl): + + def setUpClass(): + unit.TestUnit().check_modules('php') + + def test_php_get(self): + resp = self.get() + self.assertEqual(resp, {'listeners': {}, 'applications': {}}, 'empty') + self.assertEqual(self.get('/listeners'), {}, 'empty listeners prefix') + self.assertEqual(self.get('/applications'), {}, + 'empty applications prefix') + + self.put('/applications', """ + { + "app": { + "type": "php", + "workers": 1, + "root": "/app", + "index": "index.php" + } + } + """) + + resp = self.get() + + self.assertEqual(resp['listeners'], {}, 'php empty listeners') + self.assertEqual(resp['applications'], + { + "app": { + "type": "php", + "workers": 1, + "root": "/app", + "index": "index.php" + } + }, + 'php applications') + + self.assertEqual(self.get('/applications'), + { + "app": { + "type": "php", + "workers": 1, + "root": "/app", + "index": "index.php" + } + }, + 'php applications prefix') + + self.assertEqual(self.get('/applications/app'), + { + "type": "php", + "workers": 1, + "root": "/app", + "index": "index.php" + }, + 'php applications prefix 2') + + self.assertEqual(self.get('/applications/app/type'), 'php', + 'php applications type') + self.assertEqual(self.get('/applications/app/workers'), 1, + 'php applications workers') + + self.put('/listeners', '{"*:7080":{"application":"app"}}') + + self.assertEqual(self.get()['listeners'], + {"*:7080":{"application":"app"}}, 'php listeners') + self.assertEqual(self.get('/listeners'), + {"*:7080":{"application":"app"}}, 'php listeners prefix') + self.assertEqual(self.get('/listeners/*:7080'), + {"application":"app"}, 'php listeners prefix 2') + self.assertEqual(self.get('/listeners/*:7080/application'), 'app', + 'php listeners application') + + def test_php_put(self): + self.put('/', """ + { + "listeners": { + "*:7080": { + "application": "app" + } + }, + "applications": { + "app": { + "type": "php", + "workers": 1, + "root": "/app", + "index": "index.php" + } + } + } + """) + + resp = self.get() + + self.assertEqual(resp['listeners'], {"*:7080":{"application":"app"}}, + 'put listeners') + + self.assertEqual(resp['applications'], + { + "app": { + "type": "php", + "workers": 1, + "root": "/app", + "index": "index.php" + } + }, + 'put applications') + + self.put('/listeners', '{"*:7081":{"application":"app"}}') + self.assertEqual(self.get('/listeners'), + {"*:7081": {"application":"app"}}, 'put listeners prefix') + + self.put('/listeners/*:7082', '{"application":"app"}') + + self.assertEqual(self.get('/listeners'), + { + "*:7081": { + "application": "app" + }, + "*:7082": { + "application": "app" + } + }, + 'put listeners prefix 3') + + self.put('/applications/app/workers', '30') + self.assertEqual(self.get('/applications/app/workers'), 30, + 'put applications workers') + + self.put('/applications/app/root', '"/www"') + self.assertEqual(self.get('/applications/app/root'), '/www', + 'put applications root') + + def test_php_delete(self): + self.put('/', """ + { + "listeners": { + "*:7080": { + "application": "app" + } + }, + "applications": { + "app": { + "type": "php", + "workers": 1, + "root": "/app", + "index": "index.php" + } + } + } + """) + + self.assertIn('error', self.delete('/applications/app'), + 'delete app before listener') + self.assertIn('success', self.delete('/listeners/*:7080'), + 'delete listener') + self.assertIn('success', self.delete('/applications/app'), + 'delete app after listener') + self.assertIn('error', self.delete('/applications/app'), + 'delete app again') + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_python_application.py b/test/test_python_application.py index 5066624a..7014036b 100644 --- a/test/test_python_application.py +++ b/test/test_python_application.py @@ -1,90 +1,90 @@ -import unittest
-import unit
-
-class TestUnitApplication(unit.TestUnitControl):
-
- def setUpClass():
- u = unit.TestUnit()
-
- u.check_modules('python')
- u.check_version('0.4')
-
- def test_python_application(self):
- code, name = """
-
-def application(environ, start_response):
-
- content_length = int(environ.get('CONTENT_LENGTH', 0))
- body = bytes(environ['wsgi.input'].read(content_length))
-
- start_response('200 OK', [
- ('Content-Type', environ.get('CONTENT_TYPE')),
- ('Content-Length', str(len(body))),
- ('Request-Method', environ.get('REQUEST_METHOD')),
- ('Request-Uri', environ.get('REQUEST_URI')),
- ('Path-Info', environ.get('PATH_INFO')),
- ('Http-Host', environ.get('HTTP_HOST')),
- ('Remote-Addr', environ.get('REMOTE_ADDR')),
- ('Server-Name', environ.get('SERVER_NAME')),
- ('Server-Port', environ.get('SERVER_PORT')),
- ('Server-Protocol', environ.get('SERVER_PROTOCOL')),
- ('Custom-Header', environ.get('HTTP_CUSTOM_HEADER'))
- ])
- return [body]
-
-""", 'py_app'
-
- self.python_application(name, code)
-
- self.put('/', """
- {
- "listeners": {
- "*:7080": {
- "application": "app"
- }
- },
- "applications": {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "%s",
- "module": "wsgi"
- }
- }
- }
- """ % (self.testdir + '/' + name))
-
- body = 'Test body string.'
-
- r = unit.TestUnitHTTP.post(headers={
- 'Host': 'localhost',
- 'Content-Type': 'text/html',
- 'Custom-Header': 'blah'
- }, data=body)
-
- self.assertEqual(r.status_code, 200, 'status')
- self.assertEqual(r.headers['Content-Length'], str(len(body)),
- 'header content length')
- self.assertEqual(r.headers['Content-Type'], 'text/html',
- 'header content type')
- self.assertEqual(r.headers['Request-Method'], 'POST',
- 'header request method')
- self.assertEqual(r.headers['Request-Uri'], '/', 'header request uri')
- self.assertEqual(r.headers['Path-Info'], '/', 'header path info')
- self.assertEqual(r.headers['Http-Host'], 'localhost',
- 'header http host')
- self.assertEqual(r.headers['Remote-Addr'], '127.0.0.1',
- 'header remote addr')
-
- self.assertTry('assertEqual', 'header server port',
- r.headers['Server-Port'], '7080')
-
- self.assertEqual(r.headers['Server-Protocol'], 'HTTP/1.1',
- 'header server protocol')
- self.assertEqual(r.headers['Custom-Header'], 'blah',
- 'header custom header')
- self.assertEqual(r.content, str.encode(body), 'body')
-
-
-if __name__ == '__main__':
- unittest.main()
+import unittest +import unit + +class TestUnitApplication(unit.TestUnitControl): + + def setUpClass(): + u = unit.TestUnit() + + u.check_modules('python') + u.check_version('0.4') + + def test_python_application(self): + code, name = """ + +def application(environ, start_response): + + content_length = int(environ.get('CONTENT_LENGTH', 0)) + body = bytes(environ['wsgi.input'].read(content_length)) + + start_response('200 OK', [ + ('Content-Type', environ.get('CONTENT_TYPE')), + ('Content-Length', str(len(body))), + ('Request-Method', environ.get('REQUEST_METHOD')), + ('Request-Uri', environ.get('REQUEST_URI')), + ('Path-Info', environ.get('PATH_INFO')), + ('Http-Host', environ.get('HTTP_HOST')), + ('Remote-Addr', environ.get('REMOTE_ADDR')), + ('Server-Name', environ.get('SERVER_NAME')), + ('Server-Port', environ.get('SERVER_PORT')), + ('Server-Protocol', environ.get('SERVER_PROTOCOL')), + ('Custom-Header', environ.get('HTTP_CUSTOM_HEADER')) + ]) + return [body] + +""", 'py_app' + + self.python_application(name, code) + + self.put('/', """ + { + "listeners": { + "*:7080": { + "application": "app" + } + }, + "applications": { + "app": { + "type": "python", + "workers": 1, + "path": "%s", + "module": "wsgi" + } + } + } + """ % (self.testdir + '/' + name)) + + body = 'Test body string.' + + r = unit.TestUnitHTTP.post(headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': 'blah' + }, data=body) + + self.assertEqual(r.status_code, 200, 'status') + self.assertEqual(r.headers['Content-Length'], str(len(body)), + 'header content length') + self.assertEqual(r.headers['Content-Type'], 'text/html', + 'header content type') + self.assertEqual(r.headers['Request-Method'], 'POST', + 'header request method') + self.assertEqual(r.headers['Request-Uri'], '/', 'header request uri') + self.assertEqual(r.headers['Path-Info'], '/', 'header path info') + self.assertEqual(r.headers['Http-Host'], 'localhost', + 'header http host') + self.assertEqual(r.headers['Remote-Addr'], '127.0.0.1', + 'header remote addr') + + self.assertTry('assertEqual', 'header server port', + r.headers['Server-Port'], '7080') + + self.assertEqual(r.headers['Server-Protocol'], 'HTTP/1.1', + 'header server protocol') + self.assertEqual(r.headers['Custom-Header'], 'blah', + 'header custom header') + self.assertEqual(r.content, str.encode(body), 'body') + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_python_atexit.py b/test/test_python_atexit.py index 28b6f67b..be45d113 100644 --- a/test/test_python_atexit.py +++ b/test/test_python_atexit.py @@ -1,65 +1,65 @@ -import os
-import time
-import unittest
-import unit
-
-class TestUnitApplication(unit.TestUnitControl):
-
- def setUpClass():
- u = unit.TestUnit()
-
- u.check_modules('python')
- u.check_version('0.3')
-
- def test_python_application(self):
- code, name = """
-import atexit
-
-def create_file():
- open('%s', 'w')
-
-atexit.register(create_file)
-
-def application(env, start_response):
- start_response('200 OK', [('Content-Type','text/html')])
- return [b'body']
-
-""" % (self.testdir + '/atexit'), 'py_app'
-
- self.python_application(name, code)
-
- self.put('/', """
- {
- "listeners": {
- "*:7080": {
- "application": "app"
- }
- },
- "applications": {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "%s",
- "module": "wsgi"
- }
- }
- }
- """ % (self.testdir + '/' + name))
-
- unit.TestUnitHTTP.get()
-
- self.put('/', """
- {
- "listeners": {},
- "applications": {}
- }
- """)
-
- time.sleep(0.2)
-
- self.assertEqual(os.path.exists(self.testdir + '/atexit'), True,
- 'python atexit')
-
-
-if __name__ == '__main__':
- unittest.main()
+import os +import time +import unittest +import unit + +class TestUnitApplication(unit.TestUnitControl): + + def setUpClass(): + u = unit.TestUnit() + + u.check_modules('python') + u.check_version('0.3') + + def test_python_application(self): + code, name = """ +import atexit + +def create_file(): + open('%s', 'w') + +atexit.register(create_file) + +def application(env, start_response): + start_response('200 OK', [('Content-Type','text/html')]) + return [b'body'] + +""" % (self.testdir + '/atexit'), 'py_app' + + self.python_application(name, code) + + self.put('/', """ + { + "listeners": { + "*:7080": { + "application": "app" + } + }, + "applications": { + "app": { + "type": "python", + "workers": 1, + "path": "%s", + "module": "wsgi" + } + } + } + """ % (self.testdir + '/' + name)) + + unit.TestUnitHTTP.get() + + self.put('/', """ + { + "listeners": {}, + "applications": {} + } + """) + + time.sleep(0.2) + + self.assertEqual(os.path.exists(self.testdir + '/atexit'), True, + 'python atexit') + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_python_basic.py b/test/test_python_basic.py index a68f1bad..00460712 100644 --- a/test/test_python_basic.py +++ b/test/test_python_basic.py @@ -1,166 +1,166 @@ -import unittest
-import unit
-
-class TestUnitBasic(unit.TestUnitControl):
-
- def setUpClass():
- unit.TestUnit().check_modules('python')
-
- def test_python_get(self):
- resp = self.get()
- self.assertEqual(resp, {'listeners': {}, 'applications': {}}, 'empty')
- self.assertEqual(self.get('/listeners'), {}, 'empty listeners prefix')
- self.assertEqual(self.get('/applications'), {},
- 'empty applications prefix')
-
- self.put('/applications', """
- {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- """)
-
- resp = self.get()
-
- self.assertEqual(resp['listeners'], {}, 'python empty listeners')
- self.assertEqual(resp['applications'],
- {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- },
- 'python applications')
-
- self.assertEqual(self.get('/applications'),
- {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module":"wsgi"
- }
- },
- 'python applications prefix')
-
- self.assertEqual(self.get('/applications/app'),
- {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- },
- 'python applications prefix 2')
-
- self.assertEqual(self.get('/applications/app/type'), 'python',
- 'python applications type')
- self.assertEqual(self.get('/applications/app/workers'), 1,
- 'python applications workers')
-
- self.put('/listeners', '{"*:7080":{"application":"app"}}')
-
- self.assertEqual(self.get()['listeners'],
- {"*:7080":{"application":"app"}}, 'python listeners')
- self.assertEqual(self.get('/listeners'),
- {"*:7080":{"application":"app"}}, 'python listeners prefix')
- self.assertEqual(self.get('/listeners/*:7080'),
- {"application":"app"}, 'python listeners prefix 2')
- self.assertEqual(self.get('/listeners/*:7080/application'), 'app',
- 'python listeners application')
-
- def test_python_put(self):
- self.put('/', """
- {
- "listeners": {
- "*:7080": {
- "application": "app"
- }
- },
- "applications": {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- }
- """)
-
- resp = self.get()
-
- self.assertEqual(resp['listeners'], {"*:7080":{"application":"app"}},
- 'put listeners')
-
- self.assertEqual(resp['applications'],
- {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- },
- 'put applications')
-
- self.put('/listeners', '{"*:7081":{"application":"app"}}')
- self.assertEqual(self.get('/listeners'),
- {"*:7081": {"application":"app"}}, 'put listeners prefix')
-
- self.put('/listeners/*:7082', '{"application":"app"}')
-
- self.assertEqual(self.get('/listeners'),
- {
- "*:7081": {
- "application": "app"
- },
- "*:7082": {
- "application": "app"
- }
- },
- 'put listeners prefix 3')
-
- self.put('/applications/app/workers', '30')
- self.assertEqual(self.get('/applications/app/workers'), 30,
- 'put applications workers')
-
- self.put('/applications/app/path', '"/www"')
- self.assertEqual(self.get('/applications/app/path'), '/www',
- 'put applications path')
-
- def test_python_delete(self):
- self.put('/', """
- {
- "listeners": {
- "*:7080": {
- "application": "app"
- }
- },
- "applications": {
- "app": {
- "type": "python",
- "workers": 1,
- "path": "/app",
- "module": "wsgi"
- }
- }
- }
- """)
-
- self.assertIn('error', self.delete('/applications/app'),
- 'delete app before listener')
- self.assertIn('success', self.delete('/listeners/*:7080'),
- 'delete listener')
- self.assertIn('success', self.delete('/applications/app'),
- 'delete app after listener')
- self.assertIn('error', self.delete('/applications/app'),
- 'delete app again')
-
-if __name__ == '__main__':
- unittest.main()
+import unittest +import unit + +class TestUnitBasic(unit.TestUnitControl): + + def setUpClass(): + unit.TestUnit().check_modules('python') + + def test_python_get(self): + resp = self.get() + self.assertEqual(resp, {'listeners': {}, 'applications': {}}, 'empty') + self.assertEqual(self.get('/listeners'), {}, 'empty listeners prefix') + self.assertEqual(self.get('/applications'), {}, + 'empty applications prefix') + + self.put('/applications', """ + { + "app": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + """) + + resp = self.get() + + self.assertEqual(resp['listeners'], {}, 'python empty listeners') + self.assertEqual(resp['applications'], + { + "app": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + }, + 'python applications') + + self.assertEqual(self.get('/applications'), + { + "app": { + "type": "python", + "workers": 1, + "path": "/app", + "module":"wsgi" + } + }, + 'python applications prefix') + + self.assertEqual(self.get('/applications/app'), + { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + }, + 'python applications prefix 2') + + self.assertEqual(self.get('/applications/app/type'), 'python', + 'python applications type') + self.assertEqual(self.get('/applications/app/workers'), 1, + 'python applications workers') + + self.put('/listeners', '{"*:7080":{"application":"app"}}') + + self.assertEqual(self.get()['listeners'], + {"*:7080":{"application":"app"}}, 'python listeners') + self.assertEqual(self.get('/listeners'), + {"*:7080":{"application":"app"}}, 'python listeners prefix') + self.assertEqual(self.get('/listeners/*:7080'), + {"application":"app"}, 'python listeners prefix 2') + self.assertEqual(self.get('/listeners/*:7080/application'), 'app', + 'python listeners application') + + def test_python_put(self): + self.put('/', """ + { + "listeners": { + "*:7080": { + "application": "app" + } + }, + "applications": { + "app": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + } + """) + + resp = self.get() + + self.assertEqual(resp['listeners'], {"*:7080":{"application":"app"}}, + 'put listeners') + + self.assertEqual(resp['applications'], + { + "app": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + }, + 'put applications') + + self.put('/listeners', '{"*:7081":{"application":"app"}}') + self.assertEqual(self.get('/listeners'), + {"*:7081": {"application":"app"}}, 'put listeners prefix') + + self.put('/listeners/*:7082', '{"application":"app"}') + + self.assertEqual(self.get('/listeners'), + { + "*:7081": { + "application": "app" + }, + "*:7082": { + "application": "app" + } + }, + 'put listeners prefix 3') + + self.put('/applications/app/workers', '30') + self.assertEqual(self.get('/applications/app/workers'), 30, + 'put applications workers') + + self.put('/applications/app/path', '"/www"') + self.assertEqual(self.get('/applications/app/path'), '/www', + 'put applications path') + + def test_python_delete(self): + self.put('/', """ + { + "listeners": { + "*:7080": { + "application": "app" + } + }, + "applications": { + "app": { + "type": "python", + "workers": 1, + "path": "/app", + "module": "wsgi" + } + } + } + """) + + self.assertIn('error', self.delete('/applications/app'), + 'delete app before listener') + self.assertIn('success', self.delete('/listeners/*:7080'), + 'delete listener') + self.assertIn('success', self.delete('/applications/app'), + 'delete app after listener') + self.assertIn('error', self.delete('/applications/app'), + 'delete app again') + +if __name__ == '__main__': + unittest.main() diff --git a/test/unit.py b/test/unit.py index 86ff308d..87fbd029 100644 --- a/test/unit.py +++ b/test/unit.py @@ -1,236 +1,236 @@ -import os
-import re
-import sys
-import json
-import time
-import shutil
-import socket
-import tempfile
-import unittest
-from requests import Request, Session
-from subprocess import call
-from multiprocessing import Process
-
-class TestUnit(unittest.TestCase):
-
- pardir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
-
- def setUp(self):
- self._run()
-
- def tearDown(self):
- self._stop()
-
- if '--log' in sys.argv:
- with open(self.testdir + '/unit.log', 'r') as f:
- print(f.read())
-
- if '--leave' not in sys.argv:
- shutil.rmtree(self.testdir)
-
- def assertTry(self, assert_method, description, *args):
- try: getattr(self, assert_method)(*args, msg=description)
- except AssertionError: print('not yet: ' + description)
-
- def check_modules(self, *modules):
- self._run()
-
- for i in range(50):
- with open(self.testdir + '/unit.log', 'r') as f:
- log = f.read()
- m = re.search('controller started', log, re.M | re.S)
-
- if m is None:
- time.sleep(0.1)
- else:
- break
-
- if m is None:
- exit("Unit is writing log too long")
-
- missed_module = ''
- for module in modules:
- m = re.search('module: ' + module, log, re.M | re.S)
- if m is None:
- missed_module = module
- break
-
- self._stop()
- shutil.rmtree(self.testdir)
-
- if missed_module:
- raise unittest.SkipTest('Unit has no ' + missed_module + ' module')
-
- def check_version(self, version):
- with open(self.pardir + '/src/nxt_main.h' , 'r') as f:
- m = re.search('NXT_VERSION\s+"(\d+\.\d+)"', f.read(), re.M | re.S)
-
- current = m.group(1).split('.')
- need = version.split('.')
-
- for i in range(len(need)):
- if need[i] > current[i]:
- raise unittest.SkipTest('Unit too old')
-
- def _run(self):
- self.testdir = tempfile.mkdtemp(prefix='unit-test-')
-
- os.mkdir(self.testdir + '/state')
-
- print()
-
- def _run_unit():
- call([self.pardir + '/build/unitd',
- '--no-daemon',
- '--modules', self.pardir + '/build',
- '--state', self.testdir + '/state',
- '--pid', self.testdir + '/unit.pid',
- '--log', self.testdir + '/unit.log',
- '--control', 'unix:' + self.testdir + '/control.unit.sock'])
-
- self._p = Process(target=_run_unit)
- self._p.start()
-
- if not self._waitforfiles(self.testdir + '/unit.pid',
- self.testdir + '/unit.log', self.testdir + '/control.unit.sock'):
- exit("Could not start unit")
-
- def python_application(self, name, code):
- os.mkdir(self.testdir + '/' + name)
-
- with open(self.testdir + '/' + name + '/wsgi.py', 'w') as f:
- f.write(code)
-
- def _stop(self):
- with open(self.testdir + '/unit.pid', 'r') as f:
- pid = f.read().rstrip()
-
- call(['kill', pid])
-
- for i in range(50):
- if not os.path.exists(self.testdir + '/unit.pid'):
- break
- time.sleep(0.1)
-
- if os.path.exists(self.testdir + '/unit.pid'):
- exit("Could not terminate unit")
-
- self._p.join(timeout=1)
- self._terminate_process(self._p)
-
- def _terminate_process(self, process):
- if process.is_alive():
- process.terminate()
- process.join(timeout=5)
-
- if process.is_alive():
- exit("Could not terminate process " + process.pid)
-
- if process.exitcode:
- exit("Child process terminated with code " + str(process.exitcode))
-
- def _waitforfiles(self, *files):
- for i in range(50):
- wait = False
- ret = 0
-
- for f in files:
- if not os.path.exists(f):
- wait = True
- break
-
- if wait:
- time.sleep(0.1)
-
- else:
- ret = 1
- break
-
- return ret
-
-class TestUnitControl(TestUnit):
-
- # TODO socket reuse
- # TODO http client
-
- def http(self, req):
- with self._control_sock() as sock:
- sock.sendall(req)
-
- if '--verbose' in sys.argv:
- print('>>>', req, sep='\n')
-
- resp = self._recvall(sock)
-
- if '--verbose' in sys.argv:
- print('<<<', resp, sep='\n')
-
- return resp
-
- def get(self, path='/'):
- resp = self.http(('GET ' + path
- + ' HTTP/1.1\r\nHost: localhost\r\n\r\n').encode())
-
- return self._body_json(resp)
-
- def delete(self, path='/'):
- resp = self.http(('DELETE ' + path
- + ' HTTP/1.1\r\nHost: localhost\r\n\r\n').encode())
-
- return self._body_json(resp)
-
- def put(self, path='/', data=''):
- if isinstance(data, str):
- data = data.encode()
-
- resp = self.http(('PUT ' + path + ' HTTP/1.1\nHost: localhost\n'
- + 'Content-Length: ' + str(len(data))
- + '\r\n\r\n').encode() + data)
-
- return self._body_json(resp)
-
- def _control_sock(self):
- sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
- sock.connect(self.testdir + '/control.unit.sock')
- return sock
-
- def _recvall(self, sock, buff_size=4096):
- data = ''
- while True:
- part = sock.recv(buff_size).decode()
- data += part
- if len(part) < buff_size:
- break
-
- return data
-
- def _body_json(self, resp):
- m = re.search('.*?\x0d\x0a?\x0d\x0a?(.*)', resp, re.M | re.S)
- return json.loads(m.group(1))
-
-class TestUnitHTTP():
-
- @classmethod
- def http(self, method, **kwargs):
- host = '127.0.0.1:7080' if 'host' not in kwargs else kwargs['host']
- uri = '/' if 'uri' not in kwargs else kwargs['uri']
- sess = Session() if 'sess' not in kwargs else kwargs['sess']
- data = None if 'data' not in kwargs else kwargs['data']
- headers = None if 'headers' not in kwargs else kwargs['headers']
-
- req = Request(method, 'http://' + host + uri, data=data,
- headers=headers)
-
- r = sess.send(req.prepare())
-
- if 'keep' not in kwargs:
- sess.close()
- return r
-
- return (r, sess)
-
- def get(**kwargs):
- return TestUnitHTTP.http('GET', **kwargs)
-
- def post(**kwargs):
- return TestUnitHTTP.http('POST', **kwargs)
+import os +import re +import sys +import json +import time +import shutil +import socket +import tempfile +import unittest +from requests import Request, Session +from subprocess import call +from multiprocessing import Process + +class TestUnit(unittest.TestCase): + + pardir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)) + + def setUp(self): + self._run() + + def tearDown(self): + self._stop() + + if '--log' in sys.argv: + with open(self.testdir + '/unit.log', 'r') as f: + print(f.read()) + + if '--leave' not in sys.argv: + shutil.rmtree(self.testdir) + + def assertTry(self, assert_method, description, *args): + try: getattr(self, assert_method)(*args, msg=description) + except AssertionError: print('not yet: ' + description) + + def check_modules(self, *modules): + self._run() + + for i in range(50): + with open(self.testdir + '/unit.log', 'r') as f: + log = f.read() + m = re.search('controller started', log, re.M | re.S) + + if m is None: + time.sleep(0.1) + else: + break + + if m is None: + exit("Unit is writing log too long") + + missed_module = '' + for module in modules: + m = re.search('module: ' + module, log, re.M | re.S) + if m is None: + missed_module = module + break + + self._stop() + shutil.rmtree(self.testdir) + + if missed_module: + raise unittest.SkipTest('Unit has no ' + missed_module + ' module') + + def check_version(self, version): + with open(self.pardir + '/src/nxt_main.h' , 'r') as f: + m = re.search('NXT_VERSION\s+"(\d+\.\d+)"', f.read(), re.M | re.S) + + current = m.group(1).split('.') + need = version.split('.') + + for i in range(len(need)): + if need[i] > current[i]: + raise unittest.SkipTest('Unit too old') + + def _run(self): + self.testdir = tempfile.mkdtemp(prefix='unit-test-') + + os.mkdir(self.testdir + '/state') + + print() + + def _run_unit(): + call([self.pardir + '/build/unitd', + '--no-daemon', + '--modules', self.pardir + '/build', + '--state', self.testdir + '/state', + '--pid', self.testdir + '/unit.pid', + '--log', self.testdir + '/unit.log', + '--control', 'unix:' + self.testdir + '/control.unit.sock']) + + self._p = Process(target=_run_unit) + self._p.start() + + if not self._waitforfiles(self.testdir + '/unit.pid', + self.testdir + '/unit.log', self.testdir + '/control.unit.sock'): + exit("Could not start unit") + + def python_application(self, name, code): + os.mkdir(self.testdir + '/' + name) + + with open(self.testdir + '/' + name + '/wsgi.py', 'w') as f: + f.write(code) + + def _stop(self): + with open(self.testdir + '/unit.pid', 'r') as f: + pid = f.read().rstrip() + + call(['kill', pid]) + + for i in range(50): + if not os.path.exists(self.testdir + '/unit.pid'): + break + time.sleep(0.1) + + if os.path.exists(self.testdir + '/unit.pid'): + exit("Could not terminate unit") + + self._p.join(timeout=1) + self._terminate_process(self._p) + + def _terminate_process(self, process): + if process.is_alive(): + process.terminate() + process.join(timeout=5) + + if process.is_alive(): + exit("Could not terminate process " + process.pid) + + if process.exitcode: + exit("Child process terminated with code " + str(process.exitcode)) + + def _waitforfiles(self, *files): + for i in range(50): + wait = False + ret = 0 + + for f in files: + if not os.path.exists(f): + wait = True + break + + if wait: + time.sleep(0.1) + + else: + ret = 1 + break + + return ret + +class TestUnitControl(TestUnit): + + # TODO socket reuse + # TODO http client + + def http(self, req): + with self._control_sock() as sock: + sock.sendall(req) + + if '--verbose' in sys.argv: + print('>>>', req, sep='\n') + + resp = self._recvall(sock) + + if '--verbose' in sys.argv: + print('<<<', resp, sep='\n') + + return resp + + def get(self, path='/'): + resp = self.http(('GET ' + path + + ' HTTP/1.1\r\nHost: localhost\r\n\r\n').encode()) + + return self._body_json(resp) + + def delete(self, path='/'): + resp = self.http(('DELETE ' + path + + ' HTTP/1.1\r\nHost: localhost\r\n\r\n').encode()) + + return self._body_json(resp) + + def put(self, path='/', data=''): + if isinstance(data, str): + data = data.encode() + + resp = self.http(('PUT ' + path + ' HTTP/1.1\nHost: localhost\n' + + 'Content-Length: ' + str(len(data)) + + '\r\n\r\n').encode() + data) + + return self._body_json(resp) + + def _control_sock(self): + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(self.testdir + '/control.unit.sock') + return sock + + def _recvall(self, sock, buff_size=4096): + data = '' + while True: + part = sock.recv(buff_size).decode() + data += part + if len(part) < buff_size: + break + + return data + + def _body_json(self, resp): + m = re.search('.*?\x0d\x0a?\x0d\x0a?(.*)', resp, re.M | re.S) + return json.loads(m.group(1)) + +class TestUnitHTTP(): + + @classmethod + def http(self, method, **kwargs): + host = '127.0.0.1:7080' if 'host' not in kwargs else kwargs['host'] + uri = '/' if 'uri' not in kwargs else kwargs['uri'] + sess = Session() if 'sess' not in kwargs else kwargs['sess'] + data = None if 'data' not in kwargs else kwargs['data'] + headers = None if 'headers' not in kwargs else kwargs['headers'] + + req = Request(method, 'http://' + host + uri, data=data, + headers=headers) + + r = sess.send(req.prepare()) + + if 'keep' not in kwargs: + sess.close() + return r + + return (r, sess) + + def get(**kwargs): + return TestUnitHTTP.http('GET', **kwargs) + + def post(**kwargs): + return TestUnitHTTP.http('POST', **kwargs) |