| """ Tests for the linecache module """ | |
| import importlib | |
| import linecache | |
| import unittest | |
| import os.path | |
| import sys | |
| import tempfile | |
| import threading | |
| import tokenize | |
| import zipfile | |
| import zipimport | |
| from importlib.machinery import ModuleSpec | |
| from test import support | |
| from test.support import import_helper | |
| from test.support import os_helper | |
| from test.support import threading_helper | |
| from test.support.script_helper import assert_python_ok | |
| FILENAME = linecache.__file__ | |
| NONEXISTENT_FILENAME = FILENAME + '.missing' | |
| INVALID_NAME = '!@$)(!@#_1' | |
| EMPTY = '' | |
| TEST_PATH = os.path.dirname(__file__) | |
| MODULES = "linecache abc".split() | |
| MODULE_PATH = os.path.dirname(FILENAME) | |
| SOURCE_1 = ''' | |
| " Docstring " | |
| def function(): | |
| return result | |
| ''' | |
| SOURCE_2 = ''' | |
| def f(): | |
| return 1 + 1 | |
| a = f() | |
| ''' | |
| SOURCE_3 = ''' | |
| def f(): | |
| return 3''' # No ending newline | |
| class TempFile: | |
| def setUp(self): | |
| super().setUp() | |
| with tempfile.NamedTemporaryFile(delete=False) as fp: | |
| self.file_name = fp.name | |
| fp.write(self.file_byte_string) | |
| self.addCleanup(os_helper.unlink, self.file_name) | |
| class GetLineTestsGoodData(TempFile): | |
| # file_list = ['list\n', 'of\n', 'good\n', 'strings\n'] | |
| def setUp(self): | |
| self.file_byte_string = ''.join(self.file_list).encode('utf-8') | |
| super().setUp() | |
| def test_getline(self): | |
| with tokenize.open(self.file_name) as fp: | |
| for index, line in enumerate(fp): | |
| if not line.endswith('\n'): | |
| line += '\n' | |
| cached_line = linecache.getline(self.file_name, index + 1) | |
| self.assertEqual(line, cached_line) | |
| def test_getlines(self): | |
| lines = linecache.getlines(self.file_name) | |
| self.assertEqual(lines, self.file_list) | |
| class GetLineTestsBadData(TempFile): | |
| # file_byte_string = b'Bad data goes here' | |
| def test_getline(self): | |
| self.assertEqual(linecache.getline(self.file_name, 1), '') | |
| def test_getlines(self): | |
| self.assertEqual(linecache.getlines(self.file_name), []) | |
| class EmptyFile(GetLineTestsGoodData, unittest.TestCase): | |
| file_list = [] | |
| def test_getlines(self): | |
| lines = linecache.getlines(self.file_name) | |
| self.assertEqual(lines, ['\n']) | |
| class SingleEmptyLine(GetLineTestsGoodData, unittest.TestCase): | |
| file_list = ['\n'] | |
| class GoodUnicode(GetLineTestsGoodData, unittest.TestCase): | |
| file_list = ['á\n', 'b\n', 'abcdef\n', 'ááááá\n'] | |
| class BadUnicode_NoDeclaration(GetLineTestsBadData, unittest.TestCase): | |
| file_byte_string = b'\n\x80abc' | |
| class BadUnicode_WithDeclaration(GetLineTestsBadData, unittest.TestCase): | |
| file_byte_string = b'# coding=utf-8\n\x80abc' | |
| class FakeLoader: | |
| def get_source(self, fullname): | |
| return f'source for {fullname}' | |
| class NoSourceLoader: | |
| def get_source(self, fullname): | |
| return None | |
| class LineCacheTests(unittest.TestCase): | |
| def test_getline(self): | |
| getline = linecache.getline | |
| # Bad values for line number should return an empty string | |
| self.assertEqual(getline(FILENAME, 2**15), EMPTY) | |
| self.assertEqual(getline(FILENAME, -1), EMPTY) | |
| # Float values currently raise TypeError, should it? | |
| self.assertRaises(TypeError, getline, FILENAME, 1.1) | |
| # Bad filenames should return an empty string | |
| self.assertEqual(getline(EMPTY, 1), EMPTY) | |
| self.assertEqual(getline(INVALID_NAME, 1), EMPTY) | |
| # Check module loading | |
| for entry in MODULES: | |
| filename = os.path.join(MODULE_PATH, entry) + '.py' | |
| with open(filename, encoding='utf-8') as file: | |
| for index, line in enumerate(file): | |
| self.assertEqual(line, getline(filename, index + 1)) | |
| # Check that bogus data isn't returned (issue #1309567) | |
| empty = linecache.getlines('a/b/c/__init__.py') | |
| self.assertEqual(empty, []) | |
| def test_no_ending_newline(self): | |
| self.addCleanup(os_helper.unlink, os_helper.TESTFN) | |
| with open(os_helper.TESTFN, "w", encoding='utf-8') as fp: | |
| fp.write(SOURCE_3) | |
| lines = linecache.getlines(os_helper.TESTFN) | |
| self.assertEqual(lines, ["\n", "def f():\n", " return 3\n"]) | |
| def test_clearcache(self): | |
| cached = [] | |
| for entry in MODULES: | |
| filename = os.path.join(MODULE_PATH, entry) + '.py' | |
| cached.append(filename) | |
| linecache.getline(filename, 1) | |
| # Are all files cached? | |
| self.assertNotEqual(cached, []) | |
| cached_empty = [fn for fn in cached if fn not in linecache.cache] | |
| self.assertEqual(cached_empty, []) | |
| # Can we clear the cache? | |
| linecache.clearcache() | |
| cached_empty = [fn for fn in cached if fn in linecache.cache] | |
| self.assertEqual(cached_empty, []) | |
| def test_checkcache(self): | |
| getline = linecache.getline | |
| # Create a source file and cache its contents | |
| source_name = os_helper.TESTFN + '.py' | |
| self.addCleanup(os_helper.unlink, source_name) | |
| with open(source_name, 'w', encoding='utf-8') as source: | |
| source.write(SOURCE_1) | |
| getline(source_name, 1) | |
| # Keep a copy of the old contents | |
| source_list = [] | |
| with open(source_name, encoding='utf-8') as source: | |
| for index, line in enumerate(source): | |
| self.assertEqual(line, getline(source_name, index + 1)) | |
| source_list.append(line) | |
| with open(source_name, 'w', encoding='utf-8') as source: | |
| source.write(SOURCE_2) | |
| # Try to update a bogus cache entry | |
| linecache.checkcache('dummy') | |
| # Check that the cache matches the old contents | |
| for index, line in enumerate(source_list): | |
| self.assertEqual(line, getline(source_name, index + 1)) | |
| # Update the cache and check whether it matches the new source file | |
| linecache.checkcache(source_name) | |
| with open(source_name, encoding='utf-8') as source: | |
| for index, line in enumerate(source): | |
| self.assertEqual(line, getline(source_name, index + 1)) | |
| source_list.append(line) | |
| def test_lazycache_no_globals(self): | |
| lines = linecache.getlines(FILENAME) | |
| linecache.clearcache() | |
| self.assertEqual(False, linecache.lazycache(FILENAME, None)) | |
| self.assertEqual(lines, linecache.getlines(FILENAME)) | |
| def test_lazycache_smoke(self): | |
| lines = linecache.getlines(NONEXISTENT_FILENAME, globals()) | |
| linecache.clearcache() | |
| self.assertEqual( | |
| True, linecache.lazycache(NONEXISTENT_FILENAME, globals())) | |
| self.assertEqual(1, len(linecache.cache[NONEXISTENT_FILENAME])) | |
| # Note here that we're looking up a nonexistent filename with no | |
| # globals: this would error if the lazy value wasn't resolved. | |
| self.assertEqual(lines, linecache.getlines(NONEXISTENT_FILENAME)) | |
| def test_lazycache_provide_after_failed_lookup(self): | |
| linecache.clearcache() | |
| lines = linecache.getlines(NONEXISTENT_FILENAME, globals()) | |
| linecache.clearcache() | |
| linecache.getlines(NONEXISTENT_FILENAME) | |
| linecache.lazycache(NONEXISTENT_FILENAME, globals()) | |
| self.assertEqual(lines, linecache.updatecache(NONEXISTENT_FILENAME)) | |
| def test_lazycache_check(self): | |
| linecache.clearcache() | |
| linecache.lazycache(NONEXISTENT_FILENAME, globals()) | |
| linecache.checkcache() | |
| def test_lazycache_bad_filename(self): | |
| linecache.clearcache() | |
| self.assertEqual(False, linecache.lazycache('', globals())) | |
| self.assertEqual(False, linecache.lazycache('<foo>', globals())) | |
| def test_lazycache_already_cached(self): | |
| linecache.clearcache() | |
| lines = linecache.getlines(NONEXISTENT_FILENAME, globals()) | |
| self.assertEqual( | |
| False, | |
| linecache.lazycache(NONEXISTENT_FILENAME, globals())) | |
| self.assertEqual(4, len(linecache.cache[NONEXISTENT_FILENAME])) | |
| def test_memoryerror(self): | |
| lines = linecache.getlines(FILENAME) | |
| self.assertTrue(lines) | |
| def raise_memoryerror(*args, **kwargs): | |
| raise MemoryError | |
| with support.swap_attr(linecache, 'updatecache', raise_memoryerror): | |
| lines2 = linecache.getlines(FILENAME) | |
| self.assertEqual(lines2, lines) | |
| linecache.clearcache() | |
| with support.swap_attr(linecache, 'updatecache', raise_memoryerror): | |
| lines3 = linecache.getlines(FILENAME) | |
| self.assertEqual(lines3, []) | |
| self.assertEqual(linecache.getlines(FILENAME), lines) | |
| def test_loader(self): | |
| filename = 'scheme://path' | |
| linecache.clearcache() | |
| module_globals = {'__name__': 'a.b.c', '__loader__': None} | |
| self.assertEqual(linecache.getlines(filename, module_globals), []) | |
| for loader in object(), NoSourceLoader(): | |
| linecache.clearcache() | |
| module_globals = {'__name__': 'a.b.c', '__loader__': loader} | |
| with self.assertWarns(DeprecationWarning) as w: | |
| self.assertEqual(linecache.getlines(filename, module_globals), []) | |
| self.assertEqual(str(w.warning), | |
| 'Module globals is missing a __spec__.loader') | |
| linecache.clearcache() | |
| module_globals = {'__name__': 'a.b.c', '__loader__': FakeLoader()} | |
| with self.assertWarns(DeprecationWarning) as w: | |
| self.assertEqual(linecache.getlines(filename, module_globals), | |
| ['source for a.b.c\n']) | |
| self.assertEqual(str(w.warning), | |
| 'Module globals is missing a __spec__.loader') | |
| for spec in None, object(): | |
| linecache.clearcache() | |
| module_globals = {'__name__': 'a.b.c', '__loader__': FakeLoader(), | |
| '__spec__': spec} | |
| with self.assertWarns(DeprecationWarning) as w: | |
| self.assertEqual(linecache.getlines(filename, module_globals), | |
| ['source for a.b.c\n']) | |
| self.assertEqual(str(w.warning), | |
| 'Module globals is missing a __spec__.loader') | |
| linecache.clearcache() | |
| module_globals = {'__name__': 'a.b.c', '__loader__': FakeLoader(), | |
| '__spec__': ModuleSpec('', FakeLoader())} | |
| with self.assertWarns(DeprecationWarning) as w: | |
| self.assertEqual(linecache.getlines(filename, module_globals), | |
| ['source for a.b.c\n']) | |
| self.assertEqual(str(w.warning), | |
| 'Module globals; __loader__ != __spec__.loader') | |
| linecache.clearcache() | |
| spec = ModuleSpec('x.y.z', FakeLoader()) | |
| module_globals = {'__name__': 'a.b.c', '__loader__': spec.loader, | |
| '__spec__': spec} | |
| self.assertEqual(linecache.getlines(filename, module_globals), | |
| ['source for x.y.z\n']) | |
| def test_frozen(self): | |
| filename = '<frozen fakemodule>' | |
| module_globals = {'__file__': FILENAME} | |
| empty = linecache.getlines(filename) | |
| self.assertEqual(empty, []) | |
| lines = linecache.getlines(filename, module_globals) | |
| self.assertGreater(len(lines), 0) | |
| lines_cached = linecache.getlines(filename) | |
| self.assertEqual(lines, lines_cached) | |
| linecache.clearcache() | |
| empty = linecache.getlines(filename) | |
| self.assertEqual(empty, []) | |
| def test_invalid_names(self): | |
| for name, desc in [ | |
| ('\x00', 'NUL bytes filename'), | |
| (__file__ + '\x00', 'filename with embedded NUL bytes'), | |
| # A filename with surrogate codes. A UnicodeEncodeError is raised | |
| # by os.stat() upon querying, which is a subclass of ValueError. | |
| ("\uD834\uDD1E.py", 'surrogate codes (MUSICAL SYMBOL G CLEF)'), | |
| # For POSIX platforms, an OSError will be raised but for Windows | |
| # platforms, a ValueError is raised due to the path_t converter. | |
| # See: https://github.com/python/cpython/issues/122170 | |
| ('a' * 1_000_000, 'very long filename'), | |
| ]: | |
| with self.subTest(f'updatecache: {desc}'): | |
| linecache.clearcache() | |
| lines = linecache.updatecache(name) | |
| self.assertListEqual(lines, []) | |
| self.assertNotIn(name, linecache.cache) | |
| # hack into the cache (it shouldn't be allowed | |
| # but we never know what people do...) | |
| for key, fullname in [(name, 'ok'), ('key', name), (name, name)]: | |
| with self.subTest(f'checkcache: {desc}', | |
| key=key, fullname=fullname): | |
| linecache.clearcache() | |
| linecache.cache[key] = (0, 1234, [], fullname) | |
| linecache.checkcache(key) | |
| self.assertNotIn(key, linecache.cache) | |
| # just to be sure that we did not mess with cache | |
| linecache.clearcache() | |
| def test_linecache_python_string(self): | |
| cmdline = "import linecache;assert len(linecache.cache) == 0" | |
| retcode, stdout, stderr = assert_python_ok('-c', cmdline) | |
| self.assertEqual(retcode, 0) | |
| self.assertEqual(stdout, b'') | |
| self.assertEqual(stderr, b'') | |
| def test_path_importer_cache_None(self): | |
| # sys.path_importer_cache is set to None while the interpreter is | |
| # shutting down, before objects with a __del__ that may end up here | |
| # are released. | |
| filename = os.path.abspath(os_helper.TESTFN + '.py') | |
| with support.swap_attr(sys, 'path_importer_cache', None): | |
| self.assertEqual(linecache.getlines(filename), []) | |
| self.assertEqual(linecache.getline(filename, 1), '') | |
| self.assertNotIn(filename, linecache.cache) | |
| class LineCacheInvalidationTests(unittest.TestCase): | |
| def setUp(self): | |
| super().setUp() | |
| linecache.clearcache() | |
| self.deleted_file = os_helper.TESTFN + '.1' | |
| self.modified_file = os_helper.TESTFN + '.2' | |
| self.unchanged_file = os_helper.TESTFN + '.3' | |
| for fname in (self.deleted_file, | |
| self.modified_file, | |
| self.unchanged_file): | |
| self.addCleanup(os_helper.unlink, fname) | |
| with open(fname, 'w', encoding='utf-8') as source: | |
| source.write(f'print("I am {fname}")') | |
| self.assertNotIn(fname, linecache.cache) | |
| linecache.getlines(fname) | |
| self.assertIn(fname, linecache.cache) | |
| os.remove(self.deleted_file) | |
| with open(self.modified_file, 'w', encoding='utf-8') as source: | |
| source.write('print("was modified")') | |
| def test_checkcache_for_deleted_file(self): | |
| linecache.checkcache(self.deleted_file) | |
| self.assertNotIn(self.deleted_file, linecache.cache) | |
| self.assertIn(self.modified_file, linecache.cache) | |
| self.assertIn(self.unchanged_file, linecache.cache) | |
| def test_checkcache_for_modified_file(self): | |
| linecache.checkcache(self.modified_file) | |
| self.assertIn(self.deleted_file, linecache.cache) | |
| self.assertNotIn(self.modified_file, linecache.cache) | |
| self.assertIn(self.unchanged_file, linecache.cache) | |
| def test_checkcache_with_no_parameter(self): | |
| linecache.checkcache() | |
| self.assertNotIn(self.deleted_file, linecache.cache) | |
| self.assertNotIn(self.modified_file, linecache.cache) | |
| self.assertIn(self.unchanged_file, linecache.cache) | |
| class ZipArchiveTests(unittest.TestCase): | |
| """Sources of modules imported from a zip archive on sys.path.""" | |
| MODULE_SOURCE = ( | |
| '"""A module inside a zip archive."""\n' | |
| '\n' | |
| 'def f():\n' | |
| ' return "from the zip"\n' | |
| ) | |
| PACKAGE_SOURCE = 'value = 42\n' | |
| LATIN1_SOURCE = ( | |
| '# -*- coding: latin-1 -*-\n' | |
| 'value = "caf\xe9"\n' | |
| ) | |
| def setUp(self): | |
| linecache.clearcache() | |
| self.addCleanup(linecache.clearcache) | |
| tmpdir = self.enterContext(os_helper.temp_dir()) | |
| self.zip_name = os.path.join(tmpdir, 'sources.zip') | |
| with zipfile.ZipFile(self.zip_name, 'w') as zf: | |
| zf.writestr('zipmod.py', self.MODULE_SOURCE) | |
| zf.writestr('zippkg/__init__.py', self.PACKAGE_SOURCE) | |
| zf.writestr('ziplatin1.py', self.LATIN1_SOURCE.encode('latin-1')) | |
| self.enterContext(import_helper.DirsOnSysPath(self.zip_name)) | |
| for name in 'zipmod', 'zippkg', 'ziplatin1': | |
| self.addCleanup(import_helper.unload, name) | |
| self.addCleanup(sys.path_importer_cache.pop, self.zip_name, None) | |
| self.addCleanup(zipimport._zip_directory_cache.pop, | |
| self.zip_name, None) | |
| self.zipmod = importlib.import_module('zipmod') | |
| def test_getlines_without_module_globals(self): | |
| filename = self.zipmod.__file__ | |
| self.assertEqual(filename, os.path.join(self.zip_name, 'zipmod.py')) | |
| self.assertFalse(os.path.exists(filename)) | |
| lines = self.MODULE_SOURCE.splitlines(keepends=True) | |
| self.assertEqual(linecache.getlines(filename), lines) | |
| self.assertEqual(linecache.getline(filename, 4), | |
| ' return "from the zip"\n') | |
| self.assertEqual(linecache.getline(filename, 5), '') | |
| code = self.zipmod.f.__code__ | |
| self.assertEqual(code.co_filename, filename) | |
| self.assertEqual(linecache.getline(filename, code.co_firstlineno), | |
| 'def f():\n') | |
| def test_relative_archive_path(self): | |
| # A relative sys.path entry gives its modules a relative __file__. | |
| tmpdir, zip_base = os.path.split(self.zip_name) | |
| self.addCleanup(sys.path_importer_cache.pop, zip_base, None) | |
| self.addCleanup(zipimport._zip_directory_cache.pop, zip_base, None) | |
| sys.path.insert(0, zip_base) | |
| self.addCleanup(sys.path.remove, zip_base) | |
| with os_helper.change_cwd(tmpdir): | |
| zippkg = importlib.import_module('zippkg') | |
| self.assertEqual(zippkg.__file__, | |
| os.path.join(zip_base, 'zippkg', '__init__.py')) | |
| self.assertEqual(linecache.getlines(zippkg.__file__), | |
| ['value = 42\n']) | |
| def test_package(self): | |
| zippkg = importlib.import_module('zippkg') | |
| self.assertEqual(linecache.getlines(zippkg.__file__), | |
| ['value = 42\n']) | |
| def test_encoding_declaration(self): | |
| ziplatin1 = importlib.import_module('ziplatin1') | |
| self.assertEqual(linecache.getlines(ziplatin1.__file__), | |
| self.LATIN1_SOURCE.splitlines(keepends=True)) | |
| def test_missing_file(self): | |
| filename = os.path.join(self.zip_name, 'missing.py') | |
| self.assertEqual(linecache.getlines(filename), []) | |
| self.assertEqual(linecache.getline(filename, 1), '') | |
| self.assertNotIn(filename, linecache.cache) | |
| def test_checkcache_and_clearcache(self): | |
| filename = self.zipmod.__file__ | |
| lines = linecache.getlines(filename) | |
| self.assertIn(filename, linecache.cache) | |
| # A file inside an archive has no mtime of its own, so checkcache() | |
| # keeps the entry, as it does for entries loaded through a loader. | |
| self.assertIsNone(linecache.cache[filename][1]) | |
| linecache.checkcache(filename) | |
| linecache.checkcache() | |
| self.assertIn(filename, linecache.cache) | |
| self.assertEqual(linecache.getlines(filename), lines) | |
| linecache.clearcache() | |
| self.assertNotIn(filename, linecache.cache) | |
| self.assertEqual(linecache.getlines(filename), lines) | |
| class MultiThreadingTest(unittest.TestCase): | |
| @threading_helper.reap_threads | |
| @threading_helper.requires_working_threading() | |
| def test_read_write_safety(self): | |
| with tempfile.TemporaryDirectory() as tmpdirname: | |
| filenames = [] | |
| for i in range(10): | |
| name = os.path.join(tmpdirname, f"test_{i}.py") | |
| with open(name, "w") as h: | |
| h.write("import time\n") | |
| h.write("import system\n") | |
| filenames.append(name) | |
| def linecache_get_line(b): | |
| b.wait() | |
| for _ in range(100): | |
| for name in filenames: | |
| linecache.getline(name, 1) | |
| def check(funcs): | |
| barrier = threading.Barrier(len(funcs)) | |
| threads = [] | |
| for func in funcs: | |
| thread = threading.Thread(target=func, args=(barrier,)) | |
| threads.append(thread) | |
| with threading_helper.start_threads(threads): | |
| pass | |
| check([linecache_get_line] * 20) | |
| if __name__ == "__main__": | |
| unittest.main() |