Reformat files that woudl only touch a single line.

IPython/core/async_helpers.py

@@ -1,156 +1,155 b''
 """
 Async helper function that are invalid syntax on Python 3.5 and below.
 
 This code is best effort, and may have edge cases not behaving as expected. In
 particular it contain a number of heuristics to detect whether code is
 effectively async and need to run in an event loop or not.
 
 Some constructs (like top-level `return`, or `yield`) are taken care of
 explicitly to actually raise a SyntaxError and stay as close as possible to
 Python semantics.
 """
 
-
 import ast
 import asyncio
 import inspect
 from functools import wraps
 
 _asyncio_event_loop = None
 
 
 def get_asyncio_loop():
     """asyncio has deprecated get_event_loop
 
     Replicate it here, with our desired semantics:
 
     - always returns a valid, not-closed loop
     - not thread-local like asyncio's,
       because we only want one loop for IPython
     - if called from inside a coroutine (e.g. in ipykernel),
       return the running loop
 
     .. versionadded:: 8.0
     """
     try:
         return asyncio.get_running_loop()
     except RuntimeError:
         # not inside a coroutine,
         # track our own global
         pass
 
     # not thread-local like asyncio's,
     # because we only track one event loop to run for IPython itself,
     # always in the main thread.
     global _asyncio_event_loop
     if _asyncio_event_loop is None or _asyncio_event_loop.is_closed():
         _asyncio_event_loop = asyncio.new_event_loop()
     return _asyncio_event_loop
 
 
 class _AsyncIORunner:
     def __call__(self, coro):
         """
         Handler for asyncio autoawait
         """
         return get_asyncio_loop().run_until_complete(coro)
 
     def __str__(self):
         return "asyncio"
 
 
 _asyncio_runner = _AsyncIORunner()
 
 
 class _AsyncIOProxy:
     """Proxy-object for an asyncio
 
     Any coroutine methods will be wrapped in event_loop.run_
     """
 
     def __init__(self, obj, event_loop):
         self._obj = obj
         self._event_loop = event_loop
 
     def __repr__(self):
         return f"<_AsyncIOProxy({self._obj!r})>"
 
     def __getattr__(self, key):
         attr = getattr(self._obj, key)
         if inspect.iscoroutinefunction(attr):
             # if it's a coroutine method,
             # return a threadsafe wrapper onto the _current_ asyncio loop
             @wraps(attr)
             def _wrapped(*args, **kwargs):
                 concurrent_future = asyncio.run_coroutine_threadsafe(
                     attr(*args, **kwargs), self._event_loop
                 )
                 return asyncio.wrap_future(concurrent_future)
 
             return _wrapped
         else:
             return attr
 
     def __dir__(self):
         return dir(self._obj)
 
 
 def _curio_runner(coroutine):
     """
     handler for curio autoawait
     """
     import curio
 
     return curio.run(coroutine)
 
 
 def _trio_runner(async_fn):
     import trio
 
     async def loc(coro):
         """
         We need the dummy no-op async def to protect from
         trio's internal. See https://github.com/python-trio/trio/issues/89
         """
         return await coro
 
     return trio.run(loc, async_fn)
 
 
 def _pseudo_sync_runner(coro):
     """
     A runner that does not really allow async execution, and just advance the coroutine.
 
     See discussion in https://github.com/python-trio/trio/issues/608,
 
     Credit to Nathaniel Smith
     """
     try:
         coro.send(None)
     except StopIteration as exc:
         return exc.value
     else:
         # TODO: do not raise but return an execution result with the right info.
         raise RuntimeError(
             "{coro_name!r} needs a real async loop".format(coro_name=coro.__name__)
         )
 
 
 def _should_be_async(cell: str) -> bool:
     """Detect if a block of code need to be wrapped in an `async def`
 
     Attempt to parse the block of code, it it compile we're fine.
     Otherwise we  wrap if and try to compile.
 
     If it works, assume it should be async. Otherwise Return False.
 
     Not handled yet: If the block of code has a return statement as the top
     level, it will be seen as async. This is a know limitation.
     """
     try:
         code = compile(
             cell, "<>", "exec", flags=getattr(ast, "PyCF_ALLOW_TOP_LEVEL_AWAIT", 0x0)
         )
         return inspect.CO_COROUTINE & code.co_flags == inspect.CO_COROUTINE
     except (SyntaxError, MemoryError):
         return False

IPython/core/tests/test_autocall.py

@@ -1,67 +1,68 b''
 """These kinds of tests are less than ideal, but at least they run.
 
 This was an old test that was being run interactively in the top-level tests/
 directory, which we are removing.  For now putting this here ensures at least
 we do run the test, though ultimately this functionality should all be tested
 with better-isolated tests that don't rely on the global instance in iptest.
 """
+
 from IPython.core.splitinput import LineInfo
 from IPython.core.prefilter import AutocallChecker
 
 
 def doctest_autocall():
     """
     In [1]: def f1(a,b,c):
        ...:     return a+b+c
        ...:
 
     In [2]: def f2(a):
        ...:     return a + a
        ...:
 
     In [3]: def r(x):
        ...:     return True
        ...:
 
     In [4]: ;f2 a b c
     Out[4]: 'a b ca b c'
 
     In [5]: assert _ == "a b ca b c"
 
     In [6]: ,f1 a b c
     Out[6]: 'abc'
 
     In [7]: assert _ == 'abc'
 
     In [8]: print(_)
     abc
 
     In [9]: /f1 1,2,3
     Out[9]: 6
 
     In [10]: assert _ == 6
 
     In [11]: /f2 4
     Out[11]: 8
 
     In [12]: assert _ == 8
 
     In [12]: del f1, f2
 
     In [13]: ,r a
     Out[13]: True
 
     In [14]: assert _ == True
 
     In [15]: r'a'
     Out[15]: 'a'
 
     In [16]: assert _ == 'a'
     """
 
 
 def test_autocall_should_ignore_raw_strings():
     line_info = LineInfo("r'a'")
     pm = ip.prefilter_manager
     ac = AutocallChecker(shell=pm.shell, prefilter_manager=pm, config=pm.config)
     assert ac.check(line_info) is None

IPython/extensions/tests/test_autoreload.py

@@ -1,711 +1,712 b''
 """Tests for autoreload extension.
 """
+
 # -----------------------------------------------------------------------------
 #  Copyright (c) 2012 IPython Development Team.
 #
 #  Distributed under the terms of the Modified BSD License.
 #
 #  The full license is in the file COPYING.txt, distributed with this software.
 # -----------------------------------------------------------------------------
 
 # -----------------------------------------------------------------------------
 # Imports
 # -----------------------------------------------------------------------------
 
 import os
 import platform
 import pytest
 import sys
 import tempfile
 import textwrap
 import shutil
 import random
 import time
 import traceback
 from io import StringIO
 from dataclasses import dataclass
 
 import IPython.testing.tools as tt
 
 from unittest import TestCase
 
 from IPython.extensions.autoreload import AutoreloadMagics
 from IPython.core.events import EventManager, pre_run_cell
 from IPython.testing.decorators import skipif_not_numpy
 from IPython.core.interactiveshell import ExecutionInfo
 
 if platform.python_implementation() == "PyPy":
     pytest.skip(
         "Current autoreload implementation is extremely slow on PyPy",
         allow_module_level=True,
     )
 
 # -----------------------------------------------------------------------------
 # Test fixture
 # -----------------------------------------------------------------------------
 
 noop = lambda *a, **kw: None
 
 
 class FakeShell:
     def __init__(self):
         self.ns = {}
         self.user_ns = self.ns
         self.user_ns_hidden = {}
         self.events = EventManager(self, {"pre_run_cell", pre_run_cell})
         self.auto_magics = AutoreloadMagics(shell=self)
         self.events.register("pre_run_cell", self.auto_magics.pre_run_cell)
 
     register_magics = set_hook = noop
 
     def showtraceback(
         self,
         exc_tuple=None,
         filename=None,
         tb_offset=None,
         exception_only=False,
         running_compiled_code=False,
     ):
         traceback.print_exc()
 
     def run_code(self, code):
         self.events.trigger(
             "pre_run_cell",
             ExecutionInfo(
                 raw_cell="",
                 store_history=False,
                 silent=False,
                 shell_futures=False,
                 cell_id=None,
             ),
         )
         exec(code, self.user_ns)
         self.auto_magics.post_execute_hook()
 
     def push(self, items):
         self.ns.update(items)
 
     def magic_autoreload(self, parameter):
         self.auto_magics.autoreload(parameter)
 
     def magic_aimport(self, parameter, stream=None):
         self.auto_magics.aimport(parameter, stream=stream)
         self.auto_magics.post_execute_hook()
 
 
 class Fixture(TestCase):
     """Fixture for creating test module files"""
 
     test_dir = None
     old_sys_path = None
     filename_chars = "abcdefghijklmopqrstuvwxyz0123456789"
 
     def setUp(self):
         self.test_dir = tempfile.mkdtemp()
         self.old_sys_path = list(sys.path)
         sys.path.insert(0, self.test_dir)
         self.shell = FakeShell()
 
     def tearDown(self):
         shutil.rmtree(self.test_dir)
         sys.path = self.old_sys_path
 
         self.test_dir = None
         self.old_sys_path = None
         self.shell = None
 
     def get_module(self):
         module_name = "tmpmod_" + "".join(random.sample(self.filename_chars, 20))
         if module_name in sys.modules:
             del sys.modules[module_name]
         file_name = os.path.join(self.test_dir, module_name + ".py")
         return module_name, file_name
 
     def write_file(self, filename, content):
         """
         Write a file, and force a timestamp difference of at least one second
 
         Notes
         -----
         Python's .pyc files record the timestamp of their compilation
         with a time resolution of one second.
 
         Therefore, we need to force a timestamp difference between .py
         and .pyc, without having the .py file be timestamped in the
         future, and without changing the timestamp of the .pyc file
         (because that is stored in the file).  The only reliable way
         to achieve this seems to be to sleep.
         """
         content = textwrap.dedent(content)
         # Sleep one second + eps
         time.sleep(1.05)
 
         # Write
         with open(filename, "w", encoding="utf-8") as f:
             f.write(content)
 
     def new_module(self, code):
         code = textwrap.dedent(code)
         mod_name, mod_fn = self.get_module()
         with open(mod_fn, "w", encoding="utf-8") as f:
             f.write(code)
         return mod_name, mod_fn
 
 
 # -----------------------------------------------------------------------------
 # Test automatic reloading
 # -----------------------------------------------------------------------------
 
 
 def pickle_get_current_class(obj):
     """
     Original issue comes from pickle; hence the name.
     """
     name = obj.__class__.__name__
     module_name = getattr(obj, "__module__", None)
     obj2 = sys.modules[module_name]
     for subpath in name.split("."):
         obj2 = getattr(obj2, subpath)
     return obj2
 
 
 class TestAutoreload(Fixture):
     def test_reload_enums(self):
         mod_name, mod_fn = self.new_module(
             textwrap.dedent(
                 """
                                 from enum import Enum
                                 class MyEnum(Enum):
                                     A = 'A'
                                     B = 'B'
                             """
             )
         )
         self.shell.magic_autoreload("2")
         self.shell.magic_aimport(mod_name)
         self.write_file(
             mod_fn,
             textwrap.dedent(
                 """
                                 from enum import Enum
                                 class MyEnum(Enum):
                                     A = 'A'
                                     B = 'B'
                                     C = 'C'
                             """
             ),
         )
         with tt.AssertNotPrints(
             ("[autoreload of %s failed:" % mod_name), channel="stderr"
         ):
             self.shell.run_code("pass")  # trigger another reload
 
     def test_reload_class_type(self):
         self.shell.magic_autoreload("2")
         mod_name, mod_fn = self.new_module(
             """
             class Test():
                 def meth(self):
                     return "old"
         """
         )
         assert "test" not in self.shell.ns
         assert "result" not in self.shell.ns
 
         self.shell.run_code("from %s import Test" % mod_name)
         self.shell.run_code("test = Test()")
 
         self.write_file(
             mod_fn,
             """
             class Test():
                 def meth(self):
                     return "new"
         """,
         )
 
         test_object = self.shell.ns["test"]
 
         # important to trigger autoreload logic !
         self.shell.run_code("pass")
 
         test_class = pickle_get_current_class(test_object)
         assert isinstance(test_object, test_class)
 
         # extra check.
         self.shell.run_code("import pickle")
         self.shell.run_code("p = pickle.dumps(test)")
 
     def test_reload_class_attributes(self):
         self.shell.magic_autoreload("2")
         mod_name, mod_fn = self.new_module(
             textwrap.dedent(
                 """
                                 class MyClass:
 
                                     def __init__(self, a=10):
                                         self.a = a
                                         self.b = 22 
                                         # self.toto = 33
 
                                     def square(self):
                                         print('compute square')
                                         return self.a*self.a
                             """
             )
         )
         self.shell.run_code("from %s import MyClass" % mod_name)
         self.shell.run_code("first = MyClass(5)")
         self.shell.run_code("first.square()")
         with self.assertRaises(AttributeError):
             self.shell.run_code("first.cube()")
         with self.assertRaises(AttributeError):
             self.shell.run_code("first.power(5)")
         self.shell.run_code("first.b")
         with self.assertRaises(AttributeError):
             self.shell.run_code("first.toto")
 
         # remove square, add power
 
         self.write_file(
             mod_fn,
             textwrap.dedent(
                 """
                             class MyClass:
 
                                 def __init__(self, a=10):
                                     self.a = a
                                     self.b = 11
 
                                 def power(self, p):
                                     print('compute power '+str(p))
                                     return self.a**p
                             """
             ),
         )
 
         self.shell.run_code("second = MyClass(5)")
 
         for object_name in {"first", "second"}:
             self.shell.run_code(f"{object_name}.power(5)")
             with self.assertRaises(AttributeError):
                 self.shell.run_code(f"{object_name}.cube()")
             with self.assertRaises(AttributeError):
                 self.shell.run_code(f"{object_name}.square()")
             self.shell.run_code(f"{object_name}.b")
             self.shell.run_code(f"{object_name}.a")
             with self.assertRaises(AttributeError):
                 self.shell.run_code(f"{object_name}.toto")
 
     @skipif_not_numpy
     def test_comparing_numpy_structures(self):
         self.shell.magic_autoreload("2")
         self.shell.run_code("1+1")
         mod_name, mod_fn = self.new_module(
             textwrap.dedent(
                 """
                                 import numpy as np
                                 class MyClass:
                                     a = (np.array((.1, .2)),
                                          np.array((.2, .3)))
                             """
             )
         )
         self.shell.run_code("from %s import MyClass" % mod_name)
         self.shell.run_code("first = MyClass()")
 
         # change property `a`
         self.write_file(
             mod_fn,
             textwrap.dedent(
                 """
                                 import numpy as np
                                 class MyClass:
                                     a = (np.array((.3, .4)),
                                          np.array((.5, .6)))
                             """
             ),
         )
 
         with tt.AssertNotPrints(
             ("[autoreload of %s failed:" % mod_name), channel="stderr"
         ):
             self.shell.run_code("pass")  # trigger another reload
 
     def test_autoload_newly_added_objects(self):
         # All of these fail with %autoreload 2
         self.shell.magic_autoreload("3")
         mod_code = """
         def func1(): pass
         """
         mod_name, mod_fn = self.new_module(textwrap.dedent(mod_code))
         self.shell.run_code(f"from {mod_name} import *")
         self.shell.run_code("func1()")
         with self.assertRaises(NameError):
             self.shell.run_code("func2()")
         with self.assertRaises(NameError):
             self.shell.run_code("t = Test()")
         with self.assertRaises(NameError):
             self.shell.run_code("number")
 
         # ----------- TEST NEW OBJ LOADED --------------------------
 
         new_code = """
         def func1(): pass
         def func2(): pass
         class Test: pass
         number = 0
         from enum import Enum
         class TestEnum(Enum):
             A = 'a'
         """
         self.write_file(mod_fn, textwrap.dedent(new_code))
 
         # test function now exists in shell's namespace namespace
         self.shell.run_code("func2()")
         # test function now exists in module's dict
         self.shell.run_code(f"import sys; sys.modules['{mod_name}'].func2()")
         # test class now exists
         self.shell.run_code("t = Test()")
         # test global built-in var now exists
         self.shell.run_code("number")
         # test the enumerations gets loaded successfully
         self.shell.run_code("TestEnum.A")
 
         # ----------- TEST NEW OBJ CAN BE CHANGED --------------------
 
         new_code = """
         def func1(): return 'changed'
         def func2(): return 'changed'
         class Test:
             def new_func(self):
                 return 'changed'
         number = 1
         from enum import Enum
         class TestEnum(Enum):
             A = 'a'
             B = 'added'
         """
         self.write_file(mod_fn, textwrap.dedent(new_code))
         self.shell.run_code("assert func1() == 'changed'")
         self.shell.run_code("assert func2() == 'changed'")
         self.shell.run_code("t = Test(); assert t.new_func() == 'changed'")
         self.shell.run_code("assert number == 1")
         if sys.version_info < (3, 12):
             self.shell.run_code("assert TestEnum.B.value == 'added'")
 
         # ----------- TEST IMPORT FROM MODULE --------------------------
 
         new_mod_code = """
         from enum import Enum
         class Ext(Enum):
             A = 'ext'
         def ext_func():
             return 'ext'
         class ExtTest:
             def meth(self):
                 return 'ext'
         ext_int = 2
         """
         new_mod_name, new_mod_fn = self.new_module(textwrap.dedent(new_mod_code))
         current_mod_code = f"""
         from {new_mod_name} import *
         """
         self.write_file(mod_fn, textwrap.dedent(current_mod_code))
         self.shell.run_code("assert Ext.A.value == 'ext'")
         self.shell.run_code("assert ext_func() == 'ext'")
         self.shell.run_code("t = ExtTest(); assert t.meth() == 'ext'")
         self.shell.run_code("assert ext_int == 2")
 
     def test_verbose_names(self):
         # Asserts correspondense between original mode names and their verbose equivalents.
         @dataclass
         class AutoreloadSettings:
             check_all: bool
             enabled: bool
             autoload_obj: bool
 
         def gather_settings(mode):
             self.shell.magic_autoreload(mode)
             module_reloader = self.shell.auto_magics._reloader
             return AutoreloadSettings(
                 module_reloader.check_all,
                 module_reloader.enabled,
                 module_reloader.autoload_obj,
             )
 
         assert gather_settings("0") == gather_settings("off")
         assert gather_settings("0") == gather_settings("OFF")  # Case insensitive
         assert gather_settings("1") == gather_settings("explicit")
         assert gather_settings("2") == gather_settings("all")
         assert gather_settings("3") == gather_settings("complete")
 
         # And an invalid mode name raises an exception.
         with self.assertRaises(ValueError):
             self.shell.magic_autoreload("4")
 
     def test_aimport_parsing(self):
         # Modules can be included or excluded all in one line.
         module_reloader = self.shell.auto_magics._reloader
         self.shell.magic_aimport("os")  # import and mark `os` for auto-reload.
         assert module_reloader.modules["os"] is True
         assert "os" not in module_reloader.skip_modules.keys()
 
         self.shell.magic_aimport("-math")  # forbid autoreloading of `math`
         assert module_reloader.skip_modules["math"] is True
         assert "math" not in module_reloader.modules.keys()
 
         self.shell.magic_aimport(
             "-os, math"
         )  # Can do this all in one line; wasn't possible before.
         assert module_reloader.modules["math"] is True
         assert "math" not in module_reloader.skip_modules.keys()
         assert module_reloader.skip_modules["os"] is True
         assert "os" not in module_reloader.modules.keys()
 
     def test_autoreload_output(self):
         self.shell.magic_autoreload("complete")
         mod_code = """
         def func1(): pass
         """
         mod_name, mod_fn = self.new_module(mod_code)
         self.shell.run_code(f"import {mod_name}")
         with tt.AssertPrints("", channel="stdout"):  # no output; this is default
             self.shell.run_code("pass")
 
         self.shell.magic_autoreload("complete --print")
         self.write_file(mod_fn, mod_code)  # "modify" the module
         with tt.AssertPrints(
             f"Reloading '{mod_name}'.", channel="stdout"
         ):  # see something printed out
             self.shell.run_code("pass")
 
         self.shell.magic_autoreload("complete -p")
         self.write_file(mod_fn, mod_code)  # "modify" the module
         with tt.AssertPrints(
             f"Reloading '{mod_name}'.", channel="stdout"
         ):  # see something printed out
             self.shell.run_code("pass")
 
         self.shell.magic_autoreload("complete --print --log")
         self.write_file(mod_fn, mod_code)  # "modify" the module
         with tt.AssertPrints(
             f"Reloading '{mod_name}'.", channel="stdout"
         ):  # see something printed out
             self.shell.run_code("pass")
 
         self.shell.magic_autoreload("complete --print --log")
         self.write_file(mod_fn, mod_code)  # "modify" the module
         with self.assertLogs(logger="autoreload") as lo:  # see something printed out
             self.shell.run_code("pass")
         assert lo.output == [f"INFO:autoreload:Reloading '{mod_name}'."]
 
         self.shell.magic_autoreload("complete -l")
         self.write_file(mod_fn, mod_code)  # "modify" the module
         with self.assertLogs(logger="autoreload") as lo:  # see something printed out
             self.shell.run_code("pass")
         assert lo.output == [f"INFO:autoreload:Reloading '{mod_name}'."]
 
     def _check_smoketest(self, use_aimport=True):
         """
         Functional test for the automatic reloader using either
         '%autoreload 1' or '%autoreload 2'
         """
 
         mod_name, mod_fn = self.new_module(
             """
 x = 9
 
 z = 123  # this item will be deleted
 
 def foo(y):
     return y + 3
 
 class Baz(object):
     def __init__(self, x):
         self.x = x
     def bar(self, y):
         return self.x + y
     @property
     def quux(self):
         return 42
     def zzz(self):
         '''This method will be deleted below'''
         return 99
 
 class Bar:    # old-style class: weakref doesn't work for it on Python < 2.7
     def foo(self):
         return 1
 """
         )
 
         #
         # Import module, and mark for reloading
         #
         if use_aimport:
             self.shell.magic_autoreload("1")
             self.shell.magic_aimport(mod_name)
             stream = StringIO()
             self.shell.magic_aimport("", stream=stream)
             self.assertIn(("Modules to reload:\n%s" % mod_name), stream.getvalue())
 
             with self.assertRaises(ImportError):
                 self.shell.magic_aimport("tmpmod_as318989e89ds")
         else:
             self.shell.magic_autoreload("2")
             self.shell.run_code("import %s" % mod_name)
             stream = StringIO()
             self.shell.magic_aimport("", stream=stream)
             self.assertTrue(
                 "Modules to reload:\nall-except-skipped" in stream.getvalue()
             )
         self.assertIn(mod_name, self.shell.ns)
 
         mod = sys.modules[mod_name]
 
         #
         # Test module contents
         #
         old_foo = mod.foo
         old_obj = mod.Baz(9)
         old_obj2 = mod.Bar()
 
         def check_module_contents():
             self.assertEqual(mod.x, 9)
             self.assertEqual(mod.z, 123)
 
             self.assertEqual(old_foo(0), 3)
             self.assertEqual(mod.foo(0), 3)
 
             obj = mod.Baz(9)
             self.assertEqual(old_obj.bar(1), 10)
             self.assertEqual(obj.bar(1), 10)
             self.assertEqual(obj.quux, 42)
             self.assertEqual(obj.zzz(), 99)
 
             obj2 = mod.Bar()
             self.assertEqual(old_obj2.foo(), 1)
             self.assertEqual(obj2.foo(), 1)
 
         check_module_contents()
 
         #
         # Simulate a failed reload: no reload should occur and exactly
         # one error message should be printed
         #
         self.write_file(
             mod_fn,
             """
 a syntax error
 """,
         )
 
         with tt.AssertPrints(
             ("[autoreload of %s failed:" % mod_name), channel="stderr"
         ):
             self.shell.run_code("pass")  # trigger reload
         with tt.AssertNotPrints(
             ("[autoreload of %s failed:" % mod_name), channel="stderr"
         ):
             self.shell.run_code("pass")  # trigger another reload
         check_module_contents()
 
         #
         # Rewrite module (this time reload should succeed)
         #
         self.write_file(
             mod_fn,
             """
 x = 10
 
 def foo(y):
     return y + 4
 
 class Baz(object):
     def __init__(self, x):
         self.x = x
     def bar(self, y):
         return self.x + y + 1
     @property
     def quux(self):
         return 43
 
 class Bar:    # old-style class
     def foo(self):
         return 2
 """,
         )
 
         def check_module_contents():
             self.assertEqual(mod.x, 10)
             self.assertFalse(hasattr(mod, "z"))
 
             self.assertEqual(old_foo(0), 4)  # superreload magic!
             self.assertEqual(mod.foo(0), 4)
 
             obj = mod.Baz(9)
             self.assertEqual(old_obj.bar(1), 11)  # superreload magic!
             self.assertEqual(obj.bar(1), 11)
 
             self.assertEqual(old_obj.quux, 43)
             self.assertEqual(obj.quux, 43)
 
             self.assertFalse(hasattr(old_obj, "zzz"))
             self.assertFalse(hasattr(obj, "zzz"))
 
             obj2 = mod.Bar()
             self.assertEqual(old_obj2.foo(), 2)
             self.assertEqual(obj2.foo(), 2)
 
         self.shell.run_code("pass")  # trigger reload
         check_module_contents()
 
         #
         # Another failure case: deleted file (shouldn't reload)
         #
         os.unlink(mod_fn)
 
         self.shell.run_code("pass")  # trigger reload
         check_module_contents()
 
         #
         # Disable autoreload and rewrite module: no reload should occur
         #
         if use_aimport:
             self.shell.magic_aimport("-" + mod_name)
             stream = StringIO()
             self.shell.magic_aimport("", stream=stream)
             self.assertTrue(("Modules to skip:\n%s" % mod_name) in stream.getvalue())
 
             # This should succeed, although no such module exists
             self.shell.magic_aimport("-tmpmod_as318989e89ds")
         else:
             self.shell.magic_autoreload("0")
 
         self.write_file(
             mod_fn,
             """
 x = -99
 """,
         )
 
         self.shell.run_code("pass")  # trigger reload
         self.shell.run_code("pass")
         check_module_contents()
 
         #
         # Re-enable autoreload: reload should now occur
         #
         if use_aimport:
             self.shell.magic_aimport(mod_name)
         else:
             self.shell.magic_autoreload("")
 
         self.shell.run_code("pass")  # trigger reload
         self.assertEqual(mod.x, -99)
 
     def test_smoketest_aimport(self):
         self._check_smoketest(use_aimport=True)
 
     def test_smoketest_autoreload(self):
         self._check_smoketest(use_aimport=False)

IPython/terminal/pt_inputhooks/asyncio.py

@@ -1,62 +1,63 b''
 """
 Inputhook for running the original asyncio event loop while we're waiting for
 input.
 
 By default, in IPython, we run the prompt with a different asyncio event loop,
 because otherwise we risk that people are freezing the prompt by scheduling bad
 coroutines. E.g., a coroutine that does a while/true and never yield back
 control to the loop. We can't cancel that.
 
 However, sometimes we want the asyncio loop to keep running while waiting for
 a prompt.
 
 The following example will print the numbers from 1 to 10 above the prompt,
 while we are waiting for input. (This works also because we use
 prompt_toolkit`s `patch_stdout`)::
 
     In [1]: import asyncio
 
     In [2]: %gui asyncio
 
     In [3]: async def f():
        ...:     for i in range(10):
        ...:         await asyncio.sleep(1)
        ...:         print(i)
 
 
     In [4]: asyncio.ensure_future(f())
 
 """
+
 from prompt_toolkit import __version__ as ptk_version
 
 from IPython.core.async_helpers import get_asyncio_loop
 
 PTK3 = ptk_version.startswith("3.")
 
 
 def inputhook(context):
     """
     Inputhook for asyncio event loop integration.
     """
     # For prompt_toolkit 3.0, this input hook literally doesn't do anything.
     # The event loop integration here is implemented in `interactiveshell.py`
     # by running the prompt itself in the current asyncio loop. The main reason
     # for this is that nesting asyncio event loops is unreliable.
     if PTK3:
         return
 
     # For prompt_toolkit 2.0, we can run the current asyncio event loop,
     # because prompt_toolkit 2.0 uses a different event loop internally.
 
     # get the persistent asyncio event loop
     loop = get_asyncio_loop()
 
     def stop():
         loop.stop()
 
     fileno = context.fileno()
     loop.add_reader(fileno, stop)
     try:
         loop.run_forever()
     finally:
         loop.remove_reader(fileno)

IPython/utils/_process_emscripten.py

@@ -1,23 +1,22 b''
 """Emscripten-specific implementation of process utilities.
 
 This file is only meant to be imported by process.py, not by end-users.
 """
 
-
 from ._process_common import arg_split
 
 
 def system(cmd):
     raise OSError("Not available")
 
 
 def getoutput(cmd):
     raise OSError("Not available")
 
 
 def check_pid(cmd):
     raise OSError("Not available")
 
 
 # `arg_split` is still used by magics regardless of whether we are on a posix/windows/emscipten
 __all__ = ["system", "getoutput", "check_pid", "arg_split"]