Gecko [ lorencats.com ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	998 B	-rw-r--r--
_asyncio.py	363 B	-rw-r--r--
test_connect.py	1.92 KB	-rw-r--r--
test_embed_kernel.py	4.53 KB	-rw-r--r--
test_eventloop.py	971 B	-rw-r--r--
test_io.py	1.06 KB	-rw-r--r--
test_jsonutil.py	3.18 KB	-rw-r--r--
test_kernel.py	9.38 KB	-rw-r--r--
test_kernelspec.py	3.63 KB	-rw-r--r--
test_message_spec.py	14.27 KB	-rw-r--r--
test_pickleutil.py	1.17 KB	-rw-r--r--
test_serialize.py	5.44 KB	-rw-r--r--
test_start_kernel.py	1.73 KB	-rw-r--r--
test_zmq_shell.py	5.75 KB	-rw-r--r--
utils.py	4.66 KB	-rw-r--r--

Code Editor : test_connect.py

"""Tests for kernel connection utilities"""

# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.

import json
import os

from traitlets.config import Config
from ipython_genutils.tempdir import TemporaryDirectory, TemporaryWorkingDirectory
from ipython_genutils.py3compat import str_to_bytes
from ipykernel import connect
from ipykernel.kernelapp import IPKernelApp

sample_info = dict(ip='1.2.3.4', transport='ipc',
        shell_port=1, hb_port=2, iopub_port=3, stdin_port=4, control_port=5,
        key=b'abc123', signature_scheme='hmac-md5',
    )

class DummyKernelApp(IPKernelApp):
    def initialize(self, argv=[]):
        self.init_profile_dir()
        self.init_connection_file()

def test_get_connection_file():
    cfg = Config()
    with TemporaryWorkingDirectory() as d:
        cfg.ProfileDir.location = d
        cf = 'kernel.json'
        app = DummyKernelApp(config=cfg, connection_file=cf)
        app.initialize()

profile_cf = os.path.join(app.connection_dir, cf)
        assert profile_cf == app.abs_connection_file
        with open(profile_cf, 'w') as f:
            f.write("{}")
        assert os.path.exists(profile_cf)
        assert connect.get_connection_file(app) == profile_cf

app.connection_file = cf
        assert connect.get_connection_file(app) == profile_cf

def test_get_connection_info():
    with TemporaryDirectory() as d:
        cf = os.path.join(d, 'kernel.json')
        connect.write_connection_file(cf, **sample_info)
        json_info = connect.get_connection_info(cf)
        info = connect.get_connection_info(cf, unpack=True)
    assert isinstance(json_info, str)

sub_info = {k:v for k,v in info.items() if k in sample_info}
    assert sub_info == sample_info

info2 = json.loads(json_info)
    info2['key'] = str_to_bytes(info2['key'])
    sub_info2 = {k:v for k,v in info.items() if k in sample_info}
    assert sub_info2 == sample_info

${this.title}

Code Editor : test_connect.py