mirror of https://github.com/python/cpython
bpo-25872: Add unit tests for linecache and threading (GH-25913)
This commit is contained in:
parent
834498e178
commit
115dea9e26
|
@ -239,5 +239,47 @@ class LineCacheTests(unittest.TestCase):
|
|||
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()
|
||||
|
|
|
@ -4,7 +4,7 @@ Tests for the threading module.
|
|||
|
||||
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 os
|
|||
import subprocess
|
||||
import signal
|
||||
import textwrap
|
||||
import traceback
|
||||
|
||||
from unittest import mock
|
||||
from test import lock_tests
|
||||
|
@ -1345,6 +1346,22 @@ class ThreadingExceptionTests(BaseTestCase):
|
|||
# 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):
|
||||
|
|
Loading…
Reference in New Issue