mirror of
https://github.com/python/cpython.git
synced 2026-10-28 10:25:05 +08:00
gh-157146: Let linecache read sources from zip archives on sys.path (GH-157147)
linecache.getline() now works for a .zip archive on sys.path without the caller having to pass module_globals. Callers that only have a file name, such as pdb, warnings, and doctest, now get source. Reads of such files go through the get_data() method of the path entry finder registered for the archive in sys.path_importer_cache. Prior to this: os.stat() would fail on the archive-internal path, the loader lookup needed the globals, and the sys.path search would only handle relative names and assumed a filesystem rather than using an importer.
This commit is contained in:
1 parent
e2311cfb3d
commit
1d6e7d5987
3 files changed
+165
-8
No files matched your search
+54
-8
@@ -144,6 +144,7 @@ def updatecache(filename, module_globals=None):
|
||||
lazy_entry = entry if entry is not None and len(entry) == 1 else None
|
||||
if lazy_entry is None:
|
||||
lazy_entry = _make_lazycache_entry(filename, module_globals)
|
||||
data = None
|
||||
if lazy_entry is not None:
|
||||
try:
|
||||
data = lazy_entry[0]()
|
||||
@@ -154,14 +155,23 @@ def updatecache(filename, module_globals=None):
|
||||
# No luck, the PEP302 loader cannot find the source
|
||||
# for this module.
|
||||
return []
|
||||
entry = (
|
||||
len(data),
|
||||
None,
|
||||
[line + '\n' for line in data.splitlines()],
|
||||
fullname
|
||||
)
|
||||
cache[filename] = entry
|
||||
return entry[2]
|
||||
if data is None:
|
||||
# The file may be inside an archive on the module search path,
|
||||
# such as a zip file.
|
||||
try:
|
||||
data = _read_from_archive(fullname)
|
||||
except ImportError:
|
||||
# Can happen if the interpreter is shutting down.
|
||||
return []
|
||||
if data is not None:
|
||||
entry = (
|
||||
len(data),
|
||||
None,
|
||||
[line + '\n' for line in data.splitlines()],
|
||||
fullname
|
||||
)
|
||||
cache[filename] = entry
|
||||
return entry[2]
|
||||
|
||||
# Try looking through the module search path, which is only useful
|
||||
# when handling a relative filename.
|
||||
@@ -197,6 +207,42 @@ def updatecache(filename, module_globals=None):
|
||||
return lines
|
||||
|
||||
|
||||
def _read_from_archive(filename):
|
||||
"""Return the decoded contents of a file inside an archive on sys.path.
|
||||
|
||||
Path entry finders for archives, such as zipimport.zipimporter, have a
|
||||
get_data() method that reads files by their path below the archive,
|
||||
which is what __file__ and co_filename contain for modules imported
|
||||
from it. The archive is one of the parent directories of the file, so
|
||||
look for a finder registered for one of them. Return None if the file
|
||||
is not in such an archive.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
importers = sys.path_importer_cache
|
||||
if importers is None:
|
||||
# Cleared while the interpreter is shutting down.
|
||||
return None
|
||||
path = filename
|
||||
while True:
|
||||
parent = os.path.dirname(path)
|
||||
if parent == path:
|
||||
return None
|
||||
path = parent
|
||||
get_data = getattr(importers.get(path), 'get_data', None)
|
||||
if get_data is None:
|
||||
continue
|
||||
try:
|
||||
data = get_data(filename)
|
||||
except (ImportError, OSError):
|
||||
continue
|
||||
import importlib.util
|
||||
try:
|
||||
return importlib.util.decode_source(data)
|
||||
except (UnicodeDecodeError, SyntaxError):
|
||||
return None
|
||||
|
||||
|
||||
def lazycache(filename, module_globals):
|
||||
"""Seed the cache for filename with module_globals.
|
||||
|
||||
|
||||
@@ -1,13 +1,18 @@
|
||||
""" 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
|
||||
@@ -356,6 +361,17 @@ class LineCacheTests(unittest.TestCase):
|
||||
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()
|
||||
@@ -398,6 +414,98 @@ class LineCacheInvalidationTests(unittest.TestCase):
|
||||
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()
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
:mod:`linecache` can now read the source of a module that was imported from
|
||||
a zip archive on :data:`sys.path` when given only the file name, as
|
||||
:mod:`pdb`, :mod:`warnings` and :mod:`doctest` do.
|
||||
Reference in new issue
Block a user