test_autoreload.py
555 lines
| 16.8 KiB
| text/x-python
|
PythonLexer
|
r6951 | """Tests for autoreload extension. | ||
""" | ||||
|
r26242 | # ----------------------------------------------------------------------------- | ||
|
r6951 | # 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. | ||||
|
r26242 | # ----------------------------------------------------------------------------- | ||
|
r6951 | |||
|
r26242 | # ----------------------------------------------------------------------------- | ||
|
r6951 | # Imports | ||
|
r26242 | # ----------------------------------------------------------------------------- | ||
|
r6951 | |||
|
r4842 | import os | ||
import sys | ||||
import tempfile | ||||
|
r23397 | import textwrap | ||
|
r4842 | import shutil | ||
import random | ||||
import time | ||||
|
r23079 | from io import StringIO | ||
|
r4842 | |||
import nose.tools as nt | ||||
|
r4904 | import IPython.testing.tools as tt | ||
|
r4842 | |||
|
r25113 | from unittest import TestCase | ||
|
r8197 | from IPython.extensions.autoreload import AutoreloadMagics | ||
|
r15607 | from IPython.core.events import EventManager, pre_run_cell | ||
|
r4842 | |||
|
r26242 | # ----------------------------------------------------------------------------- | ||
|
r4842 | # Test fixture | ||
|
r26242 | # ----------------------------------------------------------------------------- | ||
|
r4842 | |||
|
r6951 | noop = lambda *a, **kw: None | ||
|
r13237 | |||
|
r26242 | class FakeShell: | ||
|
r4842 | def __init__(self): | ||
self.ns = {} | ||||
|
r25090 | self.user_ns = self.ns | ||
|
r24976 | self.user_ns_hidden = {} | ||
|
r26242 | self.events = EventManager(self, {"pre_run_cell", pre_run_cell}) | ||
|
r8197 | self.auto_magics = AutoreloadMagics(shell=self) | ||
|
r26242 | self.events.register("pre_run_cell", self.auto_magics.pre_run_cell) | ||
|
r6951 | |||
register_magics = set_hook = noop | ||||
|
r4842 | |||
def run_code(self, code): | ||||
|
r26242 | self.events.trigger("pre_run_cell") | ||
|
r25090 | exec(code, self.user_ns) | ||
|
r15682 | self.auto_magics.post_execute_hook() | ||
|
r4842 | |||
def push(self, items): | ||||
self.ns.update(items) | ||||
def magic_autoreload(self, parameter): | ||||
|
r8197 | self.auto_magics.autoreload(parameter) | ||
|
r4842 | |||
def magic_aimport(self, parameter, stream=None): | ||||
|
r8197 | self.auto_magics.aimport(parameter, stream=stream) | ||
|
r15682 | self.auto_magics.post_execute_hook() | ||
|
r4842 | |||
|
r25113 | class Fixture(TestCase): | ||
|
r4842 | """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): | ||||
|
r26242 | module_name = "tmpmod_" + "".join(random.sample(self.filename_chars, 20)) | ||
|
r4842 | 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. | ||||
""" | ||||
|
r25090 | content = textwrap.dedent(content) | ||
|
r4842 | # Sleep one second + eps | ||
time.sleep(1.05) | ||||
# Write | ||||
|
r26242 | with open(filename, "w") as f: | ||
|
r4842 | f.write(content) | ||
def new_module(self, code): | ||||
|
r25090 | code = textwrap.dedent(code) | ||
|
r4842 | mod_name, mod_fn = self.get_module() | ||
|
r26242 | with open(mod_fn, "w") as f: | ||
|
r4842 | f.write(code) | ||
return mod_name, mod_fn | ||||
|
r26242 | |||
# ----------------------------------------------------------------------------- | ||||
|
r4842 | # Test automatic reloading | ||
|
r26242 | # ----------------------------------------------------------------------------- | ||
|
r4842 | |||
|
r25090 | 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 | ||||
|
r23397 | |||
|
r26242 | class TestAutoreload(Fixture): | ||
|
r23397 | def test_reload_enums(self): | ||
|
r26242 | mod_name, mod_fn = self.new_module( | ||
textwrap.dedent( | ||||
""" | ||||
|
r23397 | from enum import Enum | ||
class MyEnum(Enum): | ||||
A = 'A' | ||||
B = 'B' | ||||
|
r26242 | """ | ||
) | ||||
) | ||||
|
r23397 | self.shell.magic_autoreload("2") | ||
self.shell.magic_aimport(mod_name) | ||||
|
r26242 | self.write_file( | ||
mod_fn, | ||||
textwrap.dedent( | ||||
""" | ||||
|
r23397 | from enum import Enum | ||
class MyEnum(Enum): | ||||
A = 'A' | ||||
B = 'B' | ||||
C = 'C' | ||||
|
r26242 | """ | ||
), | ||||
) | ||||
with tt.AssertNotPrints( | ||||
("[autoreload of %s failed:" % mod_name), channel="stderr" | ||||
): | ||||
|
r23397 | self.shell.run_code("pass") # trigger another reload | ||
|
r25090 | 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)") | ||||
|
r24515 | def test_reload_class_attributes(self): | ||
self.shell.magic_autoreload("2") | ||||
|
r26242 | mod_name, mod_fn = self.new_module( | ||
textwrap.dedent( | ||||
""" | ||||
|
r24515 | 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 | ||||
|
r24516 | """ | ||
) | ||||
) | ||||
|
r24515 | self.shell.run_code("from %s import MyClass" % mod_name) | ||
|
r24516 | self.shell.run_code("first = MyClass(5)") | ||
self.shell.run_code("first.square()") | ||||
|
r24515 | with nt.assert_raises(AttributeError): | ||
|
r24516 | self.shell.run_code("first.cube()") | ||
|
r24515 | with nt.assert_raises(AttributeError): | ||
|
r24516 | self.shell.run_code("first.power(5)") | ||
self.shell.run_code("first.b") | ||||
|
r24515 | with nt.assert_raises(AttributeError): | ||
|
r24516 | self.shell.run_code("first.toto") | ||
|
r24515 | |||
|
r24516 | # remove square, add power | ||
|
r24515 | |||
|
r24516 | self.write_file( | ||
mod_fn, | ||||
textwrap.dedent( | ||||
""" | ||||
|
r24515 | 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 | ||||
|
r24516 | """ | ||
), | ||||
) | ||||
self.shell.run_code("second = MyClass(5)") | ||||
|
r26242 | for object_name in {"first", "second"}: | ||
|
r26240 | self.shell.run_code(f"{object_name}.power(5)") | ||
|
r24516 | with nt.assert_raises(AttributeError): | ||
|
r26240 | self.shell.run_code(f"{object_name}.cube()") | ||
|
r24516 | with nt.assert_raises(AttributeError): | ||
|
r26240 | self.shell.run_code(f"{object_name}.square()") | ||
self.shell.run_code(f"{object_name}.b") | ||||
self.shell.run_code(f"{object_name}.a") | ||||
|
r24516 | with nt.assert_raises(AttributeError): | ||
|
r26240 | self.shell.run_code(f"{object_name}.toto") | ||
|
r23397 | |||
|
r26238 | def test_autoload_newly_added_objects(self): | ||
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 nt.assert_raises(NameError): | ||||
|
r26242 | self.shell.run_code("func2()") | ||
|
r26238 | with nt.assert_raises(NameError): | ||
|
r26242 | self.shell.run_code("t = Test()") | ||
|
r26238 | with nt.assert_raises(NameError): | ||
|
r26242 | self.shell.run_code("number") | ||
|
r26238 | |||
# ----------- 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 | ||||
|
r26242 | self.shell.run_code("number") | ||
|
r26238 | # test the enumerations gets loaded succesfully | ||
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") | ||||
self.shell.run_code("assert TestEnum.B.value == 'added'") | ||||
# ----------- TEST IMPORT FROM MODULE -------------------------- | ||||
|
r26242 | new_mod_code = """ | ||
|
r26238 | from enum import Enum | ||
class Ext(Enum): | ||||
A = 'ext' | ||||
def ext_func(): | ||||
return 'ext' | ||||
class ExtTest: | ||||
def meth(self): | ||||
return 'ext' | ||||
ext_int = 2 | ||||
|
r26242 | """ | ||
|
r26238 | new_mod_name, new_mod_fn = self.new_module(textwrap.dedent(new_mod_code)) | ||
|
r26242 | current_mod_code = f""" | ||
|
r26238 | from {new_mod_name} import * | ||
|
r26242 | """ | ||
|
r26238 | 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") | ||||
|
r4842 | def _check_smoketest(self, use_aimport=True): | ||
""" | ||||
Functional test for the automatic reloader using either | ||||
'%autoreload 1' or '%autoreload 2' | ||||
""" | ||||
|
r26242 | mod_name, mod_fn = self.new_module( | ||
""" | ||||
|
r4842 | 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 | ||||
|
r26242 | """ | ||
) | ||||
|
r4842 | |||
# | ||||
# 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) | ||||
|
r15682 | nt.assert_in(("Modules to reload:\n%s" % mod_name), stream.getvalue()) | ||
|
r4842 | |||
|
r15682 | with nt.assert_raises(ImportError): | ||
self.shell.magic_aimport("tmpmod_as318989e89ds") | ||||
|
r4842 | else: | ||
self.shell.magic_autoreload("2") | ||||
self.shell.run_code("import %s" % mod_name) | ||||
stream = StringIO() | ||||
self.shell.magic_aimport("", stream=stream) | ||||
|
r26242 | nt.assert_true( | ||
"Modules to reload:\nall-except-skipped" in stream.getvalue() | ||||
) | ||||
|
r6947 | nt.assert_in(mod_name, self.shell.ns) | ||
|
r4842 | |||
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(): | ||||
nt.assert_equal(mod.x, 9) | ||||
nt.assert_equal(mod.z, 123) | ||||
nt.assert_equal(old_foo(0), 3) | ||||
nt.assert_equal(mod.foo(0), 3) | ||||
obj = mod.Baz(9) | ||||
nt.assert_equal(old_obj.bar(1), 10) | ||||
nt.assert_equal(obj.bar(1), 10) | ||||
nt.assert_equal(obj.quux, 42) | ||||
nt.assert_equal(obj.zzz(), 99) | ||||
obj2 = mod.Bar() | ||||
nt.assert_equal(old_obj2.foo(), 1) | ||||
nt.assert_equal(obj2.foo(), 1) | ||||
check_module_contents() | ||||
# | ||||
# Simulate a failed reload: no reload should occur and exactly | ||||
# one error message should be printed | ||||
# | ||||
|
r26242 | self.write_file( | ||
mod_fn, | ||||
""" | ||||
|
r4842 | a syntax error | ||
|
r26242 | """, | ||
) | ||||
|
r4842 | |||
|
r26242 | 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 | ||||
|
r4904 | check_module_contents() | ||
|
r4842 | |||
# | ||||
# Rewrite module (this time reload should succeed) | ||||
# | ||||
|
r26242 | self.write_file( | ||
mod_fn, | ||||
""" | ||||
|
r4842 | 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 | ||||
|
r26242 | """, | ||
) | ||||
|
r4842 | |||
def check_module_contents(): | ||||
nt.assert_equal(mod.x, 10) | ||||
|
r26242 | nt.assert_false(hasattr(mod, "z")) | ||
|
r4842 | |||
|
r26242 | nt.assert_equal(old_foo(0), 4) # superreload magic! | ||
|
r4842 | nt.assert_equal(mod.foo(0), 4) | ||
obj = mod.Baz(9) | ||||
|
r26242 | nt.assert_equal(old_obj.bar(1), 11) # superreload magic! | ||
|
r4842 | nt.assert_equal(obj.bar(1), 11) | ||
nt.assert_equal(old_obj.quux, 43) | ||||
nt.assert_equal(obj.quux, 43) | ||||
|
r26242 | nt.assert_false(hasattr(old_obj, "zzz")) | ||
nt.assert_false(hasattr(obj, "zzz")) | ||||
|
r4842 | |||
obj2 = mod.Bar() | ||||
nt.assert_equal(old_obj2.foo(), 2) | ||||
nt.assert_equal(obj2.foo(), 2) | ||||
|
r26242 | self.shell.run_code("pass") # trigger reload | ||
|
r4842 | check_module_contents() | ||
# | ||||
# Another failure case: deleted file (shouldn't reload) | ||||
# | ||||
os.unlink(mod_fn) | ||||
|
r26242 | self.shell.run_code("pass") # trigger reload | ||
|
r4842 | 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) | ||||
|
r26242 | nt.assert_true(("Modules to skip:\n%s" % mod_name) in stream.getvalue()) | ||
|
r4842 | |||
# This should succeed, although no such module exists | ||||
self.shell.magic_aimport("-tmpmod_as318989e89ds") | ||||
else: | ||||
self.shell.magic_autoreload("0") | ||||
|
r26242 | self.write_file( | ||
mod_fn, | ||||
""" | ||||
|
r4842 | x = -99 | ||
|
r26242 | """, | ||
) | ||||
|
r4842 | |||
|
r26242 | self.shell.run_code("pass") # trigger reload | ||
|
r4842 | 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("") | ||||
|
r26242 | self.shell.run_code("pass") # trigger reload | ||
|
r4842 | nt.assert_equal(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) | ||||