Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
sagemath
GitHub Repository: sagemath/sagecell
Path: blob/master/tests/untrusted_kernel_manager_tests.py
447 views
1
import untrusted_kernel_manager
2
3
from misc import assert_is, assert_equal, assert_in, assert_not_in, assert_raises, assert_regexp_matches, assert_is_instance, assert_is_not_none, assert_greater, assert_len, assert_uuid, capture_output, Config
4
5
def test_init():
6
umkm = untrusted_kernel_manager.UntrustedMultiKernelManager("testing.log", '127.0.0.1', update_function=test_init)
7
assert_len(umkm._kernels, 0)
8
assert_equal(umkm.filename, "testing.log")
9
assert_is(hasattr(umkm, "fkm"), True)
10
11
class TestUntrustedMultiKernelManager(object):
12
def setup(self):
13
self.a = untrusted_kernel_manager.UntrustedMultiKernelManager("/dev/null", '127.0.0.1')
14
def teardown(self):
15
for i in list(self.a._kernels):
16
self.a.kill_kernel(i)
17
18
def test_start_kernel_success(self):
19
y = self.a.start_kernel()
20
assert_is_instance(y, dict)
21
assert_len(y, 2)
22
assert_in("kernel_id", y)
23
assert_uuid(y["kernel_id"])
24
assert_in(y["kernel_id"], self.a._kernels)
25
assert_in("connection", y)
26
assert_len(y["connection"], 6)
27
for s in ("stdin_port", "hb_port", "shell_port", "iopub_port"):
28
assert_in(s, y["connection"])
29
assert_len(str(y["connection"][s]), 5)
30
assert_in("ip", y["connection"])
31
assert_equal(y["connection"]["ip"], "127.0.0.1")
32
assert_in("key", y["connection"])
33
assert_uuid(y["connection"]["key"])
34
35
def test_kill_kernel_success(self): # depends on start_kernel
36
y = self.a.start_kernel()
37
kernel_id = y["kernel_id"]
38
assert_in(kernel_id, self.a._kernels)
39
40
retval = self.a.kill_kernel(kernel_id)
41
assert_is(retval, True)
42
assert_not_in(kernel_id, self.a._kernels)
43
44
def test_kill_kernel_invalid_kernel_id(self):
45
retval = self.a.kill_kernel(44)
46
assert_is(retval, False)
47
48
def test_purge_kernels_success(self): # depends on start_kernel
49
for i in xrange(5):
50
self.a.start_kernel()
51
52
retval = self.a.purge_kernels()
53
assert_equal(retval, [])
54
55
def test_purge_kernels_with_failures(self): # depends on start_kernel
56
for i in xrange(5):
57
self.a.start_kernel()
58
self.a._kernels.add(55)
59
self.a._kernels.add(66)
60
61
retval = self.a.purge_kernels()
62
assert_in(55, retval)
63
assert_in(66, retval)
64
assert_len(retval, 2)
65
66