Add more tests

ipykernel/iostream.py

@@ -264,7 +264,7 @@ def __getattr__(self, attr):
         super().__getattr__(attr)  # type:ignore[misc]
 
     def __setattr__(self, attr, value):
-        if attr == "io_thread" or (attr.startswith("__" and attr.endswith("__"))):
+        if attr == "io_thread" or (attr.startswith("__") and attr.endswith("__")):
             super().__setattr__(attr, value)
         else:
             warnings.warn(

ipykernel/tests/test_io.py

@@ -1,12 +1,13 @@
 """Test IO capturing functionality"""
 
 import io
+import warnings
 
 import pytest
 import zmq
 from jupyter_client.session import Session
 
-from ipykernel.iostream import IOPubThread, OutStream
+from ipykernel.iostream import MASTER, BackgroundSocket, IOPubThread, OutStream
 
 
 def test_io_api():
@@ -51,3 +52,57 @@ def test_io_isatty():
 
     stream = OutStream(session, thread, "stdout", isatty=True)
     assert stream.isatty()
+
+
+def test_io_thread():
+    ctx = zmq.Context()
+    pub = ctx.socket(zmq.PUB)
+    thread = IOPubThread(pub)
+    thread._setup_pipe_in()
+    msg = [thread._pipe_uuid, b"a"]
+    thread._handle_pipe_msg(msg)
+    ctx1, pipe = thread._setup_pipe_out()
+    pipe.close()
+    thread._pipe_in.close()
+    thread._check_mp_mode = lambda: MASTER
+    thread._really_send([b"hi"])
+    ctx1.destroy()
+    thread.close()
+    thread.close()
+    thread._really_send(None)
+
+
+def test_background_socket():
+    ctx = zmq.Context()
+    pub = ctx.socket(zmq.PUB)
+    thread = IOPubThread(pub)
+    sock = BackgroundSocket(thread)
+    assert sock.__class__ == BackgroundSocket
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        sock.linger = 101
+        assert thread.socket.linger == 101
+    assert sock.io_thread == thread
+    sock.send(b"hi")
+
+
+def test_outstream():
+    session = Session()
+    ctx = zmq.Context()
+    pub = ctx.socket(zmq.PUB)
+    thread = IOPubThread(pub)
+    thread.start()
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        stream = OutStream(session, pub, "stdout")
+        stream = OutStream(session, thread, "stdout", pipe=object())
+
+    stream = OutStream(session, thread, "stdout", isatty=True, echo=io.StringIO())
+    with pytest.raises(io.UnsupportedOperation):
+        stream.fileno()
+    stream._watch_pipe_fd()
+    stream.flush()
+    stream.write("hi")
+    stream.writelines(["ab", "cd"])
+    assert stream.writable

ipykernel/tests/test_zmq_shell.py

@@ -3,15 +3,24 @@
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
+import os
 import unittest
+import warnings
 from queue import Queue
 from threading import Thread
+from unittest.mock import MagicMock
 
+import pytest
 import zmq
 from jupyter_client.session import Session
 from traitlets import Int
 
-from ipykernel.zmqshell import ZMQDisplayPublisher
+from ipykernel.zmqshell import (
+    InteractiveShell,
+    KernelMagics,
+    ZMQDisplayPublisher,
+    ZMQInteractiveShell,
+)
 
 
 class NoReturnDisplayHook:
@@ -201,5 +210,49 @@ def test_unregister_hook(self):
         self.assertFalse(second)
 
 
+def test_magics(tmp_path):
+    context = zmq.Context()
+    socket = context.socket(zmq.PUB)
+    shell = InteractiveShell()
+    shell.user_ns["hi"] = 1
+    magics = KernelMagics(shell)
+
+    def find_edit_target(*args):
+        return str(tmp_path), 0, 1
+
+    tmp_file = tmp_path / "test.txt"
+    tmp_file.write_text("hi", "utf8")
+    magics._find_edit_target = find_edit_target
+    magics.edit("hi")
+    magics.clear([])
+    magics.less(str(tmp_file))
+    if os.name == "posix":
+        magics.man("ls")
+    magics.autosave("10")
+
+    socket.close()
+    context.destroy()
+
+
+def test_zmq_interactive_shell(kernel):
+    shell = ZMQInteractiveShell()
+
+    with pytest.raises(RuntimeError):
+        shell.enable_gui("tk")
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        shell.data_pub_class = MagicMock()
+        shell.data_pub
+    shell.kernel = kernel
+    shell.set_next_input("hi")
+    assert shell.get_parent() is None
+    if os.name == "posix":
+        shell.system_piped("ls")
+    else:
+        shell.system_piped("dir")
+    shell.ask_exit()
+
+
 if __name__ == "__main__":
     unittest.main()

ipykernel/zmqshell.py

@@ -285,7 +285,7 @@ def edit(self, parameter_s="", last_call=None):
         opts, args = self.parse_options(parameter_s, "prn:")
 
         try:
-            filename, lineno, _ = CodeMagics._find_edit_target(self.shell, args, opts, last_call)
+            filename, lineno, _ = self._find_edit_target(self.shell, args, opts, last_call)
         except MacroToEdit:
             # TODO: Implement macro editing over 2 processes.
             print("Macro editing not yet implemented in 2-process model.")
@@ -326,7 +326,8 @@ def less(self, arg_s):
         if arg_s.endswith(".py"):
             cont = self.shell.pycolorize(openpy.read_py_file(arg_s, skip_encoding_cookie=False))
         else:
-            cont = open(arg_s).read()
+            with open(arg_s) as fid:
+                cont = fid.read()
         page.page(cont)
 
     more = line_magic("more")(less)