Skip to content

Commit

Permalink
Make tests with temporary modules more robust
Browse files Browse the repository at this point in the history 
- Replace import by importlib.util.spec_from_file_location()
- Invalidate module caches after creating modules
- Improve single-use module names
  • Loading branch information
@dlu-ch
dlu-ch committed Dec 28, 2021
1 parent 1541c7a commit 930fc72
Show file tree
Hide file tree
Showing 3 changed files with 99 additions and 44 deletions.
83 changes: 59 additions & 24 deletions test/dlb/0/test_ex_tool_definition.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import time
import tempfile
import zipfile
import importlib.util
import unittest


Expand Down Expand Up @@ -710,7 +711,7 @@ class ToolDefinitionAmbiguityTest(testenv.TemporaryDirectoryTestCase):

# noinspection PyAbstractClass
def test_location_of_tools_are_correct(self):
lineno = 713 # of this line
lineno = 714 # of this line

class A(dlb.ex.Tool):
pass
Expand All @@ -726,6 +727,9 @@ class C(A):
self.assertEqual(C.definition_location, (os.path.realpath(__file__), None, lineno + 2 + 3 + 3))

def test_location_in_zip_archive_package_is_correct(self):
module_name = 'single_use_module1'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

with tempfile.TemporaryDirectory() as tmp_dir_path:
with tempfile.TemporaryDirectory() as content_tmp_dir_path:
open(os.path.join(content_tmp_dir_path, '__init__.py'), 'w').close()
Expand All @@ -737,19 +741,24 @@ def test_location_in_zip_archive_package_is_correct(self):

zip_file_path = os.path.join(tmp_dir_path, 'abc.zip')
with zipfile.ZipFile(zip_file_path, 'w') as z:
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname='u1/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname='u1/v.py')
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname=f'{module_name}/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname=f'{module_name}/v.py')

importlib.invalidate_caches()
sys.path.insert(0, zip_file_path)
try:
# noinspection PyUnresolvedReferences
import u1.v
import single_use_module1.v
finally:
del sys.path[0]

self.assertEqual(u1.v.A.definition_location, (os.path.realpath(zip_file_path), os.path.join('u1', 'v.py'), 2))
self.assertEqual((os.path.realpath(zip_file_path), os.path.join(module_name, 'v.py'), 2),
single_use_module1.v.A.definition_location)

def test_fails_for_zip_without_zip_suffix(self):
module_name = 'single_use_module2'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

with tempfile.TemporaryDirectory() as tmp_dir_path:
with tempfile.TemporaryDirectory() as content_tmp_dir_path:
open(os.path.join(content_tmp_dir_path, '__init__.py'), 'w').close()
Expand All @@ -761,47 +770,53 @@ def test_fails_for_zip_without_zip_suffix(self):

zip_file_path = os.path.join(tmp_dir_path, 'abc.zi')
with zipfile.ZipFile(zip_file_path, 'w') as z:
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname='u2/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname='u2/v.py')
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname=f'{module_name}/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname=f'{module_name}/v.py')

importlib.invalidate_caches()
sys.path.insert(0, zip_file_path)
try:
# noinspection PyUnresolvedReferences
with self.assertRaises(dlb.ex.DefinitionAmbiguityError) as cm:
import u2.v
import single_use_module2.v
finally:
del sys.path[0]

msg = (
"invalid tool definition: location of definition is unknown\n"
" | class: <class 'u2.v.A'>\n"
" | define the class in a regular file or in a zip archive ending in '.zip'\n"
" | note also the significance of upper and lower case of module search paths "
"on case-insensitive filesystems"
f"invalid tool definition: location of definition is unknown\n"
f" | class: <class '{module_name}.v.A'>\n"
f" | define the class in a regular file or in a zip archive ending in '.zip'\n"
f" | note also the significance of upper and lower case of module search paths "
f"on case-insensitive filesystems"
)
self.assertEqual(msg, str(cm.exception))

def test_location_in_zip_archive_module_is_correct(self):
module_name = 'single_use_module3'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

with tempfile.TemporaryDirectory() as tmp_dir_path:
with tempfile.TemporaryDirectory() as content_tmp_dir_path:
with open(os.path.join(content_tmp_dir_path, 'u3.py'), 'w') as f:
with open(os.path.join(content_tmp_dir_path, f'{module_name}.py'), 'w') as f:
f.write(
'import dlb.ex\n'
'class A(dlb.ex.Tool): pass'
)

zip_file_path = os.path.join(tmp_dir_path, 'abc.zip')
with zipfile.ZipFile(zip_file_path, 'w') as z:
z.write(os.path.join(content_tmp_dir_path, 'u3.py'), arcname='u3.py')
z.write(os.path.join(content_tmp_dir_path, f'{module_name}.py'), arcname=f'{module_name}.py')

importlib.invalidate_caches()
sys.path.insert(0, zip_file_path)
try:
# noinspection PyUnresolvedReferences
import u3
import single_use_module3
finally:
del sys.path[0]

self.assertEqual(u3.A.definition_location, (os.path.realpath(zip_file_path), 'u3.py', 2))
self.assertEqual((os.path.realpath(zip_file_path), f'{module_name}.py', 2),
single_use_module3.A.definition_location)

# noinspection PyAbstractClass
def test_definition_location_is_readonly(self):
Expand All @@ -816,13 +831,20 @@ class A(dlb.ex.Tool):
self.assertEqual(A.definition_location[0], os.path.realpath(__file__))

def test_fails_on_import_with_relative_search_path(self):
with open(os.path.join('z.py'), 'x') as f:
module_name = 'single_use_module1'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

module_file_name = f'{module_name}.py'

with open(os.path.join(module_file_name), 'x') as f:
f.write(
'import dlb.ex\n'
'class A(dlb.ex.Tool): pass\n'
)


sys.path.insert(0, '.') # !
importlib.invalidate_caches()
try:
regex = (
r"(?m)\A"
Expand All @@ -833,8 +855,9 @@ def test_fails_on_import_with_relative_search_path(self):
r"when the defining module is imported\Z"
)
with self.assertRaisesRegex(dlb.ex.DefinitionAmbiguityError, regex):
# noinspection PyUnresolvedReferences
import z # needs a name different from the already loaded modules
spec = importlib.util.spec_from_file_location(module_name, module_file_name)
spec.loader.exec_module(importlib.util.module_from_spec(spec))
# for some reason 'import ...' works differently on Travis CI than local, so avoid it
finally:
del sys.path[0]

Expand Down Expand Up @@ -1124,6 +1147,9 @@ class A(dlb.ex.Tool):
self.assertTrue(p in info.definition_paths, [p, info.definition_paths])

def test_path_of_tools_defined_in_managed_tree_are_correct(self):
module_name = 'w' # TODO rename
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

os.mkdir('a')
open(os.path.join('a/__init__.py'), 'x').close()
with open(os.path.join('a/u.py'), 'x') as f:
Expand All @@ -1139,17 +1165,18 @@ def test_path_of_tools_defined_in_managed_tree_are_correct(self):
'class D: pass\n'
)

with open(os.path.join('w.py'), 'x') as f:
with open(os.path.join(f'{module_name}.py'), 'x') as f:
f.write(
'import a.u\n'
'import v\n'
'class E(a.u.C, v.D): pass\n'
)

importlib.invalidate_caches()
sys.path.insert(0, os.getcwd())
try:
# noinspection PyUnresolvedReferences
import w # needs a name different from the already loaded modules
import w
finally:
del sys.path[0]

Expand All @@ -1175,16 +1202,24 @@ def test_path_of_tools_defined_in_managed_tree_are_correct(self):
self.assertEqual(info1, info2)

def test_definition_fails_in_import_with_relative_search_path(self):
with open(os.path.join('z.py'), 'x') as f:
module_name = 'single_use_module_2'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

module_file_name = f'{module_name}.py'

with open(os.path.join(module_file_name), 'x') as f:
f.write(
'import dlb.ex\n'
'class A(dlb.ex.Tool): pass\n'
)

sys.path.insert(0, '.') # !
importlib.invalidate_caches()
try:
with self.assertRaises(dlb.ex.DefinitionAmbiguityError):
import z # needs a name different from the already loaded modules
spec = importlib.util.spec_from_file_location(module_name, module_file_name)
spec.loader.exec_module(importlib.util.module_from_spec(spec))
# for some reason 'import ...' works differently on Travis CI than local, so avoid it
finally:
del sys.path[0]

Expand Down
21 changes: 16 additions & 5 deletions test/dlb/0/test_ex_tool_run.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,13 @@
import dlb.fs
import dlb.ex
import sys
import re
import os.path
import marshal
import tempfile
import zipfile
import io
import importlib
import unittest


Expand Down Expand Up @@ -940,6 +942,10 @@ def test_inaccessible_dependency_causes_redo(self):
class RedoIfDefinitionChangedTest(testenv.TemporaryWorkingDirectoryTestCase):

def test_redo_if_source_has_changed(self):
module_name = 'single_use_module4'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

zip_file_name = 'abc.zip'

with tempfile.TemporaryDirectory() as content_tmp_dir_path:
open(os.path.join(content_tmp_dir_path, '__init__.py'), 'w').close()
Expand All @@ -953,17 +959,18 @@ def test_redo_if_source_has_changed(self):

zip_file_path = os.path.abspath('abc.zip')
with zipfile.ZipFile(zip_file_path, 'w') as z:
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname='u4/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname='u4/v.py')
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname=f'{module_name}/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname=f'{module_name}/v.py')

importlib.invalidate_caches()
sys.path.insert(0, zip_file_path)
try:
# noinspection PyUnresolvedReferences
import u4.v
import single_use_module4.v
finally:
del sys.path[0]

t = u4.v.A()
t = single_use_module4.v.A()

with dlb.ex.Context():
output = io.StringIO()
Expand All @@ -981,7 +988,11 @@ def test_redo_if_source_has_changed(self):
output = io.StringIO()
dlb.di.set_output_file(output)
self.assertTrue(t.start())
regex = r"\b()redo necessary because of filesystem object: 'abc.zip' \n"
regex = (
r'\b()redo necessary because of filesystem object: '
f'{re.escape(repr(zip_file_name))} '
r'\n'
)
self.assertRegex(output.getvalue(), regex)


Expand Down
39 changes: 24 additions & 15 deletions test/dlb/3/test_ex_tool_definition.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import tempfile
import zipfile
import inspect
import importlib
import unittest

import dlb.ex
Expand All @@ -20,6 +21,9 @@ class ThisIsAUnitTest(unittest.TestCase):
class ToolDefinitionAmbiguityTest(testenv.TemporaryDirectoryTestCase):

def test_fails_for_non_existing_source_file(self):
module_name = 'single_use_module1'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

with tempfile.TemporaryDirectory() as tmp_dir_path:
with tempfile.TemporaryDirectory() as content_tmp_dir_path:
open(os.path.join(content_tmp_dir_path, '__init__.py'), 'w').close()
Expand All @@ -31,35 +35,38 @@ def test_fails_for_non_existing_source_file(self):

zip_file_path = os.path.join(tmp_dir_path, 'abc.zip')
with zipfile.ZipFile(zip_file_path, 'w') as z:
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname='u2/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname='u2/v.py')
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname=f'{module_name}/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname=f'{module_name}/v.py')

importlib.invalidate_caches()
sys.path.insert(0, zip_file_path)
orig_isfile = os.path.isfile

def isfile_except_zip_file_path(path):
return False if path == zip_file_path else orig_isfile(path)

os.path.isfile = isfile_except_zip_file_path

try:
# noinspection PyUnresolvedReferences
with self.assertRaises(dlb.ex.DefinitionAmbiguityError) as cm:
import u2.v
import single_use_module1.v
finally:
os.path.isfile = orig_isfile
del sys.path[0]

msg = (
"invalid tool definition: location of definition is unknown\n"
" | class: <class 'u2.v.A'>\n"
" | define the class in a regular file or in a zip archive ending in '.zip'\n"
" | note also the significance of upper and lower case of module search paths "
"on case-insensitive filesystems"
f"invalid tool definition: location of definition is unknown\n"
f" | class: <class '{module_name}.v.A'>\n"
f" | define the class in a regular file or in a zip archive ending in '.zip'\n"
f" | note also the significance of upper and lower case of module search paths "
f"on case-insensitive filesystems"
)
self.assertEqual(msg, str(cm.exception))

def test_location_in_zip_archive_package_is_normalized(self): # os.path.altsep is replaced by os.path.sep
module_name = 'single_use_module2'
self.assertNotIn(module_name, sys.modules) # needs a name different from all already loaded modules

with tempfile.TemporaryDirectory() as tmp_dir_path:
with tempfile.TemporaryDirectory() as content_tmp_dir_path:
open(os.path.join(content_tmp_dir_path, '__init__.py'), 'w').close()
Expand All @@ -71,8 +78,8 @@ def test_location_in_zip_archive_package_is_normalized(self): # os.path.altsep

zip_file_path = os.path.join(tmp_dir_path, 'abc.zip')
with zipfile.ZipFile(zip_file_path, 'w') as z:
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname='u1/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname='u1/v.py')
z.write(os.path.join(content_tmp_dir_path, '__init__.py'), arcname=f'{module_name}/__init__.py')
z.write(os.path.join(content_tmp_dir_path, 'v.py'), arcname=f'{module_name}/v.py')

orig_getframeinfo = inspect.getframeinfo
orig_altsep = os.path.altsep
Expand All @@ -81,8 +88,8 @@ def test_location_in_zip_archive_package_is_normalized(self): # os.path.altsep
if fake_altsep is None:
fake_altsep = '/' if os.path.altsep == '\\' else '\\'

module_path_in_zip = f'{zip_file_path}{os.path.sep}u1{os.path.sep}v.py'
fake_module_path_in_zip = f'{zip_file_path}{os.path.sep}u1{fake_altsep}v.py'
module_path_in_zip = f'{zip_file_path}{os.path.sep}{module_name}{os.path.sep}v.py'
fake_module_path_in_zip = f'{zip_file_path}{os.path.sep}{module_name}{fake_altsep}v.py'

def getframeinfo_except_zip_file_path(frame, context=1):
f = orig_getframeinfo(frame, context)
Expand All @@ -93,13 +100,15 @@ def getframeinfo_except_zip_file_path(frame, context=1):
os.path.altsep = fake_altsep
inspect.getframeinfo = getframeinfo_except_zip_file_path

importlib.invalidate_caches()
sys.path.insert(0, zip_file_path)
try:
# noinspection PyUnresolvedReferences
import u1.v
import single_use_module2.v
finally:
os.path.altsep = orig_altsep
inspect.getsourcefile = getframeinfo_except_zip_file_path
del sys.path[0]

self.assertEqual(u1.v.A.definition_location, (os.path.realpath(zip_file_path), os.path.join('u1', 'v.py'), 2))
self.assertEqual((os.path.realpath(zip_file_path), os.path.join(module_name, 'v.py'), 2),
single_use_module2.v.A.definition_location)

0 comments on commit 930fc72

Please sign in to comment.