Drop ipython_genutils requirement

* use `from traitlets.utils import filefind`
* `ensure_dir_exists` -> `os.makedirs(..., exist_ok=True)`
* `buffer_to_bytes` - use in-place copy
* `TemporaryDirectory` - use from `tempfile`
* `TemporaryWorkingDirectory` - make private copy
* `str_to_bytes` - use `str.encode("utf-8")` for test cases

Author: penguinolog

ipykernel/connect.py

@@ -7,7 +7,6 @@
 import sys
 from subprocess import Popen, PIPE
 
-
 import jupyter_client
 from jupyter_client import write_connection_file
 
@@ -21,7 +20,7 @@ def get_connection_file(app=None):
     app : IPKernelApp instance [optional]
         If unspecified, the currently running app will be used
     """
-    from ipython_genutils.path import filefind
+    from traitlets.utils import filefind
     if app is None:
         from ipykernel.kernelapp import IPKernelApp
         if not IPKernelApp.initialized():

ipykernel/kernelapp.py

@@ -26,11 +26,11 @@
 from IPython.core.shellapp import (
     InteractiveShellApp, shell_flags, shell_aliases
 )
-from ipython_genutils.path import filefind, ensure_dir_exists
 from traitlets import (
     Any, Instance, Dict, Unicode, Integer, Bool, DottedObjectName, Type, default
 )
 from traitlets.utils.importstring import import_item
+from traitlets.utils import filefind
 from jupyter_core.paths import jupyter_runtime_dir
 from jupyter_client import write_connection_file
 from jupyter_client.connect import ConnectionFileMixin
@@ -260,7 +260,7 @@ def init_connection_file(self):
         except IOError:
             self.log.debug("Connection file not found: %s", self.connection_file)
             # This means I own it, and I'll create it in this directory:
-            ensure_dir_exists(os.path.dirname(self.abs_connection_file), 0o700)
+            os.makedirs(os.path.dirname(self.abs_connection_file), mode=0o700, exist_ok=True)
             # Also, I will clean it up:
             atexit.register(self.cleanup_connection_file)
             return

ipykernel/pickleutil.py

@@ -2,8 +2,9 @@
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
-
+import typing
 import warnings
+
 warnings.warn("ipykernel.pickleutil is deprecated. It has moved to ipyparallel.",
     DeprecationWarning,
     stacklevel=2
@@ -15,7 +16,6 @@
 from types import FunctionType
 
 from traitlets.utils.importstring import import_item
-from ipython_genutils.py3compat import buffer_to_bytes
 
 # This registers a hook when it's imported
 from ipyparallel.serialize import codeutil  # noqa F401
@@ -279,11 +279,18 @@ def get_object(self, g=None):
 
 
 class CannedBytes(CannedObject):
-    wrap = staticmethod(buffer_to_bytes)
+    @staticmethod
+    def wrap(buf: typing.Union[memoryview, bytes, typing.SupportsBytes]) -> bytes:
+        """Cast a buffer or memoryview object to bytes"""
+        if isinstance(buf, memoryview):
+            return buf.tobytes()
+        if not isinstance(buf, bytes):
+            return bytes(buf)
+        return buf
 
     def __init__(self, obj):
         self.buffers = [obj]
-    
+
     def get_object(self, g=None):
         data = self.buffers[0]
         return self.wrap(data)

ipykernel/tests/test_connect.py

@@ -6,22 +6,30 @@
 import errno
 import json
 import os
+from tempfile import TemporaryDirectory
 from unittest.mock import patch
 
 import pytest
 import zmq
 
 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
 
+from .utils import TemporaryWorkingDirectory
 
-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',
-    )
+
+sample_info = {
+    '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):
@@ -60,12 +68,12 @@ def test_get_connection_info():
         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}
+    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}
+    info2['key'] = info2['key'].encode("utf-8")
+    sub_info2 = {k: v for k, v in info.items() if k in sample_info}
     assert sub_info2 == sample_info
 
 

ipykernel/tests/test_embed_kernel.py

@@ -14,7 +14,6 @@
 
 from jupyter_client import BlockingKernelClient
 from jupyter_core import paths
-from ipython_genutils import py3compat
 
 
 SETUP_TIMEOUT = 60
@@ -41,7 +40,7 @@ def connection_file_ready(connection_file):
         except ValueError:
             return False
 
-    kernel = Popen([sys.executable, '-c', cmd], stdout=PIPE, stderr=PIPE)
+    kernel = Popen([sys.executable, '-c', cmd], stdout=PIPE, stderr=PIPE, encoding="utf-8")
     try:
         connection_file = os.path.join(
             paths.jupyter_runtime_dir(),
@@ -58,8 +57,7 @@ def connection_file_ready(connection_file):
         time.sleep(0.1)
 
         if kernel.poll() is not None:
-            o,e = kernel.communicate()
-            e = py3compat.cast_unicode(e)
+            o, e = kernel.communicate()
             raise IOError("Kernel failed to start:\n%s" % e)
 
         if not os.path.exists(connection_file):

ipykernel/tests/test_kernel.py

@@ -8,6 +8,7 @@
 import os.path
 import sys
 import time
+from tempfile import TemporaryDirectory
 
 from flaky import flaky
 import pytest
@@ -16,7 +17,6 @@
 from IPython.testing import decorators as dec, tools as tt
 import IPython
 from IPython.paths import locate_profile
-from ipython_genutils.tempdir import TemporaryDirectory
 
 from .utils import (
     new_kernel, kernel, TIMEOUT, assemble_output, execute,

ipykernel/tests/utils.py

@@ -4,7 +4,9 @@
 # Distributed under the terms of the Modified BSD License.
 
 import atexit
+import os
 import sys
+from tempfile import TemporaryDirectory
 from time import time
 
 from contextlib import contextmanager
@@ -180,3 +182,22 @@ def wait_for_idle(kc):
         content = msg['content']
         if msg_type == 'status' and content['execution_state'] == 'idle':
             break
+
+
+class TemporaryWorkingDirectory(TemporaryDirectory):
+    """
+    Creates a temporary directory and sets the cwd to that directory.
+    Automatically reverts to previous cwd upon cleanup.
+    Usage example:
+
+        with TemporaryWorkingDirectory() as tmpdir:
+            ...
+    """
+    def __enter__(self):
+        self.old_wd = os.getcwd()
+        os.chdir(self.name)
+        return super().__enter__()
+
+    def __exit__(self, exc, value, tb):
+        os.chdir(self.old_wd)
+        return super().__exit__(exc, value, tb)

setup.py

@@ -62,7 +62,6 @@ def run(self):
     keywords=['Interactive', 'Interpreter', 'Shell', 'Web'],
     python_requires='>=3.7',
     install_requires=[
-        "ipython_genutils",
         'importlib-metadata<5;python_version<"3.8.0"',
         'argcomplete>=1.12.3;python_version<"3.8.0"',
         'debugpy>=1.0.0,<2.0',