| """ Tests for the linecache module """ |
| |
| import linecache |
| import unittest |
| import os.path |
| import tempfile |
| import threading |
| import tokenize |
| from importlib.machinery import ModuleSpec |
| from test import support |
| 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' |
| |
| for loader in (None, object(), NoSourceLoader()): |
| linecache.clearcache() |
| module_globals = {'__name__': 'a.b.c', '__loader__': loader} |
| self.assertEqual(linecache.getlines(filename, module_globals), []) |
| |
| linecache.clearcache() |
| module_globals = {'__name__': 'a.b.c', '__loader__': FakeLoader()} |
| self.assertEqual(linecache.getlines(filename, module_globals), |
| ['source for a.b.c\n']) |
| |
| for spec in (None, object(), ModuleSpec('', FakeLoader())): |
| linecache.clearcache() |
| module_globals = {'__name__': 'a.b.c', '__loader__': FakeLoader(), |
| '__spec__': spec} |
| self.assertEqual(linecache.getlines(filename, module_globals), |
| ['source for a.b.c\n']) |
| |
| 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'') |
| |
| 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 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() |
