Commit cdf83ddc authored by BAIRE Anthony's avatar BAIRE Anthony
Browse files

test common functions

parent a8ef83a6
......@@ -5,6 +5,7 @@ import asyncio
from contextlib import contextmanager, suppress
import logging
import re
import socket
import threading
import time
import unittest
......@@ -20,6 +21,7 @@ REGISTRY = "localhost:8002/allgo/dev/test"
ENV = "test"
MYSQL_HOST = "dev-mysql"
DOCKER_HOST = "unix:///run/docker.sock"
PORT = 6789
S = SandboxState
V= VersionState
......@@ -29,6 +31,28 @@ log = logging.getLogger("test_controller")
BAD_SANDBOX_NAMES = "", ".", "..", "foo/bar", "../blah", "/root", "foo/"
@contextmanager
def lock_loop(loop, *, timeout=1):
cond = threading.Condition()
def lock():
with cond:
cond.notify()
log.debug("loop locked")
cond.wait()
log.debug("loop unlocked")
with cond:
loop.call_soon_threadsafe(lock)
if not cond.wait(timeout):
raise Exception("timeout")
try:
yield
finally:
with cond:
cond.notify()
class DockerError(docker.errors.APIError):
def __init__(self, *k):
Exception.__init__(self, *k)
......@@ -46,7 +70,7 @@ class ControllerTestCase(unittest.TestCase):
sandbox_host = DOCKER_HOST,
swarm_host = DOCKER_HOST,
mysql_host = MYSQL_HOST,
port = 6789,
port = PORT,
registry = REGISTRY,
env = ENV,
datastore_path = "/data/dev/cache/test/datastore",
......@@ -141,6 +165,8 @@ class ControllerTestCase(unittest.TestCase):
while not m.called and self._thread.is_alive():
time.sleep(0.01)
def lock_loop(self, **kw):
return lock_loop(self._loop, **kw)
def tearDown(self):
self._loop.call_soon_threadsafe(self.ctrl.shutdown)
......@@ -258,15 +284,65 @@ class ControllerTestCase(unittest.TestCase):
@with_db
def test_webapp_query(self, ses, app):
info = self.ctrl.query_webapp(app.id)
self.assertIsInstance(info, controller.WebappInfo)
self.assertIs(info.ctrl, self.ctrl)
self.assertEqual(info.docker_name, "test-app")
self.assertEqual(info.sandbox_state, S.idle)
self.assertIsNone(info.version)
self.assertEqual(info.image, REGISTRY + "/webapp/test-app")
self.assertEqual(info.sandbox_name, ENV + "-sandbox-test-app")
def test_common_functions(self, ses, app):
ctrl = self.ctrl
self.assertEqual(ctrl.gen_sandbox_name(app), ENV + "-sandbox-test-app")
self.assertEqual(ctrl.gen_image_name(app), REGISTRY + "/webapp/test-app")
# session is a thread-local session object
sid = id(ctrl.session)
for i in range(10):
self.assertEqual(id(ctrl.session), sid)
def func():
sid2 = id(ctrl.session)
self.assertNotEqual(sid2, sid)
for i in range(10):
self.assertEqual(id(ctrl.session), sid2)
th = threading.Thread(target = func)
th.start()
th.join()
def test_check_host_path(self):
good = lambda *k: self.assertIsNone(self.ctrl.check_host_path(*k))
bad = lambda reg, *k: self.assertRaisesRegex(controller.Error, reg, self.ctrl.check_host_path, *k)
good( "isfile", "/bin/ls")
bad("path .* is not canonical", "isfile", "/bin//ls")
bad("path .* is not canonical", "isfile", "/etc/../bin/ls")
bad("path .* is not absolute", "isfile", "bin/ls")
bad("path .* not found", "isdir", "/bin/ls")
good( "isdir", "/proc/1/fd")
bad("path .* contains a symbolic link", "isdir", "/proc/self/fd")
def test_sock_callback(self):
def connect(nb):
for s in range(nb):
s = socket.socket()
s.connect(("127.0.0.1", PORT))
s.close()
time.sleep(.01)
with mock.patch.object(self.ctrl, "check_db") as m:
with self.lock_loop():
connect(10)
time.sleep(.1)
m.assert_called_once_with()
m.reset_mock()
with self.lock_loop():
connect(10)
time.sleep(.1)
m.reset_mock()
connect(10)
self.assertGreater(m.call_count, 1)
@with_db
def test_sandbox_start(self, ses, app):
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment