diff --git a/Lib/test/test_linecache.py b/Lib/test/test_linecache.py index 59e00da242ad63..c6e2dadbb25e1f 100644 --- a/Lib/test/test_linecache.py +++ b/Lib/test/test_linecache.py @@ -239,5 +239,47 @@ def raise_memoryerror(*args, **kwargs): self.assertEqual(linecache.getlines(FILENAME), lines) +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) + + if __name__ == "__main__": unittest.main() diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py index 08c0ccd9a79b2a..b563797cbd0d39 100644 --- a/Lib/test/test_threading.py +++ b/Lib/test/test_threading.py @@ -4,7 +4,7 @@ import test.support from test.support import threading_helper -from test.support import verbose, cpython_only +from test.support import verbose, cpython_only, os_helper from test.support.import_helper import import_module from test.support.script_helper import assert_python_ok, assert_python_failure @@ -19,6 +19,7 @@ import subprocess import signal import textwrap +import traceback from unittest import mock from test import lock_tests @@ -1345,6 +1346,22 @@ def run(self): # explicitly break the reference cycle to not leak a dangling thread thread.exc = None + def test_multithread_modify_file_noerror(self): + # See issue25872 + def modify_file(): + with open(os_helper.TESTFN, 'w', encoding='utf-8') as fp: + fp.write(' ') + traceback.format_stack() + + self.addCleanup(os_helper.unlink, os_helper.TESTFN) + threads = [ + threading.Thread(target=modify_file) + for i in range(100) + ] + for t in threads: + t.start() + t.join() + class ThreadRunFail(threading.Thread): def run(self):