diff --git a/Lib/code.py b/Lib/code.py
index b134886dc..df1d7199e 100644
--- a/Lib/code.py
+++ b/Lib/code.py
@@ -224,7 +224,7 @@ class InteractiveConsole(InteractiveInterpreter):
sys.ps1 = ">>> "
delete_ps1_after = True
try:
- _ps2 = sys.ps2
+ sys.ps2
delete_ps2_after = False
except AttributeError:
sys.ps2 = "... "
@@ -385,7 +385,7 @@ def interact(banner=None, readfunc=None, local=None, exitmsg=None, local_exit=Fa
if __name__ == "__main__":
import argparse
- parser = argparse.ArgumentParser(color=True)
+ parser = argparse.ArgumentParser()
parser.add_argument('-q', action='store_true',
help="don't print version and copyright messages")
args = parser.parse_args()
diff --git a/Lib/dbm/dumb.py b/Lib/dbm/dumb.py
index def120ffc..a080f4e86 100644
--- a/Lib/dbm/dumb.py
+++ b/Lib/dbm/dumb.py
@@ -9,7 +9,7 @@ XXX TO DO:
- seems to contain a bug when updating...
- reclaim free space (currently, space once occupied by deleted or expanded
-items is never reused)
+items is not reused except if .reorganize() is called)
- support concurrent access (currently, if two processes take turns making
updates, they can mess up the index)
@@ -17,8 +17,6 @@ updates, they can mess up the index)
- support efficient access to large databases (currently, the whole index
is read when the database is opened, and some updates rewrite the whole index)
-- support opening for read-only (flag = 'm')
-
"""
import ast as _ast
@@ -289,6 +287,35 @@ class _Database(collections.abc.MutableMapping):
def __exit__(self, *args):
self.close()
+ def reorganize(self):
+ if self._readonly:
+ raise error('The database is opened for reading only')
+ self._verify_open()
+ # Ensure all changes are committed before reorganizing.
+ self._commit()
+ # Open file in r+ to allow changing in-place.
+ with _io.open(self._datfile, 'rb+') as f:
+ reorganize_pos = 0
+
+ # Iterate over existing keys, sorted by starting byte.
+ for key in sorted(self._index, key = lambda k: self._index[k][0]):
+ pos, siz = self._index[key]
+ f.seek(pos)
+ val = f.read(siz)
+
+ f.seek(reorganize_pos)
+ f.write(val)
+ self._index[key] = (reorganize_pos, siz)
+
+ blocks_occupied = (siz + _BLOCKSIZE - 1) // _BLOCKSIZE
+ reorganize_pos += blocks_occupied * _BLOCKSIZE
+
+ f.truncate(reorganize_pos)
+ self._modified = True
+ # Commit changes to index, which were not in-place.
+ self._commit()
+
+
def open(file, flag='c', mode=0o666):
"""Open the database file, filename, and return corresponding object.
diff --git a/Lib/dbm/sqlite3.py b/Lib/dbm/sqlite3.py
index d0eed54e0..c8ee6f184 100644
--- a/Lib/dbm/sqlite3.py
+++ b/Lib/dbm/sqlite3.py
@@ -15,6 +15,7 @@ LOOKUP_KEY = "SELECT value FROM Dict WHERE key = CAST(? AS BLOB)"
STORE_KV = "REPLACE INTO Dict (key, value) VALUES (CAST(? AS BLOB), CAST(? AS BLOB))"
DELETE_KEY = "DELETE FROM Dict WHERE key = CAST(? AS BLOB)"
ITER_KEYS = "SELECT key FROM Dict"
+REORGANIZE = "VACUUM"
class error(OSError):
@@ -126,6 +127,9 @@ class _Database(MutableMapping):
def __exit__(self, *args):
self.close()
+ def reorganize(self):
+ self._execute(REORGANIZE)
+
def open(filename, /, flag="r", mode=0o666):
"""Open a dbm.sqlite3 database and return the dbm object.
diff --git a/Lib/ensurepip/__init__.py b/Lib/ensurepip/__init__.py
index 7d13414ff..bc2fb742c 100644
--- a/Lib/ensurepip/__init__.py
+++ b/Lib/ensurepip/__init__.py
@@ -10,7 +10,7 @@ from shutil import copy2
__all__ = ["version", "bootstrap"]
-_PIP_VERSION = "26.1.2"
+_PIP_VERSION = "26.2.1"
# Directory of system wheel packages. Some Linux distribution packaging
# policies recommend against bundling dependencies. For example, Fedora
@@ -131,6 +131,15 @@ def _bootstrap(*, root=None, upgrade=False, user=False,
Note that calling this function will alter both sys.path and os.environ.
"""
+
+ try:
+ import zlib
+ except ImportError:
+ raise ModuleNotFoundError(
+ "ensurepip requires the standard library module 'zlib' "
+ "to install pip."
+ ) from None
+
if altinstall and default_pip:
raise ValueError("Cannot use altinstall and default_pip together")
@@ -169,6 +178,8 @@ def _bootstrap(*, root=None, upgrade=False, user=False,
args += ["--user"]
if verbosity:
args += ["-" + "v" * verbosity]
+ if sys.implementation.cache_tag is None:
+ args += ["--no-compile"]
return _run_pip([*args, "pip"], [os.fsdecode(tmp_wheel_path)])
@@ -206,7 +217,7 @@ def _uninstall_helper(*, verbosity=0):
def _main(argv=None):
import argparse
- parser = argparse.ArgumentParser(color=True)
+ parser = argparse.ArgumentParser()
parser.add_argument(
"--version",
action="version",
diff --git a/Lib/ensurepip/_bundled/pip-26.1.2-py3-none-any.whl b/Lib/ensurepip/_bundled/pip-26.2.1-py3-none-any.whl
similarity index 57%
rename from Lib/ensurepip/_bundled/pip-26.1.2-py3-none-any.whl
rename to Lib/ensurepip/_bundled/pip-26.2.1-py3-none-any.whl
index 24b5cc90a..bc442f6b9 100644
Binary files a/Lib/ensurepip/_bundled/pip-26.1.2-py3-none-any.whl and b/Lib/ensurepip/_bundled/pip-26.2.1-py3-none-any.whl differ
diff --git a/Lib/pyclbr.py b/Lib/pyclbr.py
index 37f86995d..35772cc01 100644
--- a/Lib/pyclbr.py
+++ b/Lib/pyclbr.py
@@ -158,7 +158,6 @@ def _readmodule(module, path, inpackage=None):
return _readmodule(submodule, parent['__path__'], package)
# Search the path for the module.
- f = None
if inpackage is not None:
search_path = path
else:
diff --git a/Lib/shelve.py b/Lib/shelve.py
index 50584716e..9f6296667 100644
--- a/Lib/shelve.py
+++ b/Lib/shelve.py
@@ -56,12 +56,16 @@ entries in the cache, and empty the cache (d.sync() also synchronizes
the persistent dictionary on disk, if feasible).
"""
-from pickle import DEFAULT_PROTOCOL, Pickler, Unpickler
-from io import BytesIO
+from pickle import DEFAULT_PROTOCOL, dumps, loads
import collections.abc
-__all__ = ["Shelf", "BsdDbShelf", "DbfilenameShelf", "open"]
+__all__ = ["ShelveError", "Shelf", "BsdDbShelf", "DbfilenameShelf", "open"]
+
+
+class ShelveError(Exception):
+ pass
+
class _ClosedDict(collections.abc.MutableMapping):
'Marker for a closed dict. Access attempts raise a ValueError.'
@@ -82,7 +86,7 @@ class Shelf(collections.abc.MutableMapping):
"""
def __init__(self, dict, protocol=None, writeback=False,
- keyencoding="utf-8"):
+ keyencoding="utf-8", *, serializer=None, deserializer=None):
self.dict = dict
if protocol is None:
protocol = DEFAULT_PROTOCOL
@@ -91,6 +95,16 @@ class Shelf(collections.abc.MutableMapping):
self.cache = {}
self.keyencoding = keyencoding
+ if serializer is None and deserializer is None:
+ self.serializer = dumps
+ self.deserializer = loads
+ elif (serializer is None) ^ (deserializer is None):
+ raise ShelveError("serializer and deserializer must be "
+ "defined together")
+ else:
+ self.serializer = serializer
+ self.deserializer = deserializer
+
def __iter__(self):
for k in self.dict.keys():
yield k.decode(self.keyencoding)
@@ -110,8 +124,8 @@ class Shelf(collections.abc.MutableMapping):
try:
value = self.cache[key]
except KeyError:
- f = BytesIO(self.dict[key.encode(self.keyencoding)])
- value = Unpickler(f).load()
+ f = self.dict[key.encode(self.keyencoding)]
+ value = self.deserializer(f)
if self.writeback:
self.cache[key] = value
return value
@@ -119,10 +133,8 @@ class Shelf(collections.abc.MutableMapping):
def __setitem__(self, key, value):
if self.writeback:
self.cache[key] = value
- f = BytesIO()
- p = Pickler(f, self._protocol)
- p.dump(value)
- self.dict[key.encode(self.keyencoding)] = f.getvalue()
+ serialized_value = self.serializer(value, self._protocol)
+ self.dict[key.encode(self.keyencoding)] = serialized_value
def __delitem__(self, key):
del self.dict[key.encode(self.keyencoding)]
@@ -171,6 +183,11 @@ class Shelf(collections.abc.MutableMapping):
if hasattr(self.dict, 'sync'):
self.dict.sync()
+ def reorganize(self):
+ self.sync()
+ if hasattr(self.dict, 'reorganize'):
+ self.dict.reorganize()
+
class BsdDbShelf(Shelf):
"""Shelf implementation using the "BSD" db interface.
@@ -186,33 +203,29 @@ class BsdDbShelf(Shelf):
"""
def __init__(self, dict, protocol=None, writeback=False,
- keyencoding="utf-8"):
- Shelf.__init__(self, dict, protocol, writeback, keyencoding)
+ keyencoding="utf-8", *, serializer=None, deserializer=None):
+ Shelf.__init__(self, dict, protocol, writeback, keyencoding,
+ serializer=serializer, deserializer=deserializer)
def set_location(self, key):
(key, value) = self.dict.set_location(key)
- f = BytesIO(value)
- return (key.decode(self.keyencoding), Unpickler(f).load())
+ return (key.decode(self.keyencoding), self.deserializer(value))
def next(self):
(key, value) = next(self.dict)
- f = BytesIO(value)
- return (key.decode(self.keyencoding), Unpickler(f).load())
+ return (key.decode(self.keyencoding), self.deserializer(value))
def previous(self):
(key, value) = self.dict.previous()
- f = BytesIO(value)
- return (key.decode(self.keyencoding), Unpickler(f).load())
+ return (key.decode(self.keyencoding), self.deserializer(value))
def first(self):
(key, value) = self.dict.first()
- f = BytesIO(value)
- return (key.decode(self.keyencoding), Unpickler(f).load())
+ return (key.decode(self.keyencoding), self.deserializer(value))
def last(self):
(key, value) = self.dict.last()
- f = BytesIO(value)
- return (key.decode(self.keyencoding), Unpickler(f).load())
+ return (key.decode(self.keyencoding), self.deserializer(value))
class DbfilenameShelf(Shelf):
@@ -222,9 +235,11 @@ class DbfilenameShelf(Shelf):
See the module's __doc__ string for an overview of the interface.
"""
- def __init__(self, filename, flag='c', protocol=None, writeback=False):
+ def __init__(self, filename, flag='c', protocol=None, writeback=False, *,
+ serializer=None, deserializer=None):
import dbm
- Shelf.__init__(self, dbm.open(filename, flag), protocol, writeback)
+ Shelf.__init__(self, dbm.open(filename, flag), protocol, writeback,
+ serializer=serializer, deserializer=deserializer)
def clear(self):
"""Remove all items from the shelf."""
@@ -233,8 +248,8 @@ class DbfilenameShelf(Shelf):
self.cache.clear()
self.dict.clear()
-
-def open(filename, flag='c', protocol=None, writeback=False):
+def open(filename, flag='c', protocol=None, writeback=False, *,
+ serializer=None, deserializer=None):
"""Open a persistent dictionary for reading and writing.
The filename parameter is the base filename for the underlying
@@ -247,4 +262,5 @@ def open(filename, flag='c', protocol=None, writeback=False):
See the module's __doc__ string for an overview of the interface.
"""
- return DbfilenameShelf(filename, flag, protocol, writeback)
+ return DbfilenameShelf(filename, flag, protocol, writeback,
+ serializer=serializer, deserializer=deserializer)
diff --git a/Lib/sqlite3/__main__.py b/Lib/sqlite3/__main__.py
index 4ccf292dd..ec72c6943 100644
--- a/Lib/sqlite3/__main__.py
+++ b/Lib/sqlite3/__main__.py
@@ -10,9 +10,12 @@ import sys
from argparse import ArgumentParser
from code import InteractiveConsole
from textwrap import dedent
+from _colorize import get_theme, theme_no_color
+
+from ._completer import completer
-def execute(c, sql, suppress_errors=True):
+def execute(c, sql, suppress_errors=True, theme=theme_no_color):
"""Helper that wraps execution of SQL code.
This is used both by the REPL and by direct execution from the CLI.
@@ -25,11 +28,15 @@ def execute(c, sql, suppress_errors=True):
for row in c.execute(sql):
print(row)
except sqlite3.Error as e:
+ t = theme.traceback
tp = type(e).__name__
try:
- print(f"{tp} ({e.sqlite_errorname}): {e}", file=sys.stderr)
+ tp += f" ({e.sqlite_errorname})"
except AttributeError:
- print(f"{tp}: {e}", file=sys.stderr)
+ pass
+ print(
+ f"{t.type}{tp}{t.reset}: {t.message}{e}{t.reset}", file=sys.stderr
+ )
if not suppress_errors:
sys.exit(1)
@@ -37,10 +44,11 @@ def execute(c, sql, suppress_errors=True):
class SqliteInteractiveConsole(InteractiveConsole):
"""A simple SQLite REPL."""
- def __init__(self, connection):
+ def __init__(self, connection, use_color=False):
super().__init__()
self._con = connection
self._cur = connection.cursor()
+ self._use_color = use_color
def runsource(self, source, filename="", symbol="single"):
"""Override runsource, the core of the InteractiveConsole REPL.
@@ -48,37 +56,42 @@ class SqliteInteractiveConsole(InteractiveConsole):
Return True if more input is needed; buffering is done automatically.
Return False if input is a complete statement ready for execution.
"""
+ theme = get_theme(force_no_color=not self._use_color)
+
if not source or source.isspace():
return False
+ # Remember to update CLI_COMMANDS in _completer.py
if source[0] == ".":
match source[1:].strip():
case "version":
- print(f"{sqlite3.sqlite_version}")
+ print(sqlite3.sqlite_version)
case "help":
- print("Enter SQL code and press enter.")
+ t = theme.syntax
+ print(f"Enter SQL code or one of the below commands, and press enter.\n\n"
+ f"{t.builtin}.version{t.reset} Print underlying SQLite library version\n"
+ f"{t.builtin}.help{t.reset} Print this help message\n"
+ f"{t.builtin}.quit{t.reset} Exit the CLI, equivalent to CTRL-D\n")
case "quit":
sys.exit(0)
case "":
pass
case _ as unknown:
- self.write("Error: unknown command or invalid arguments:"
- f' "{unknown}".\n')
+ t = theme.traceback
+ self.write(f'{t.type}Error{t.reset}: {t.message}unknown '
+ f'command: "{unknown}"{t.reset}\n')
else:
if not sqlite3.complete_statement(source):
return True
- execute(self._cur, source)
+ execute(self._cur, source, theme=theme)
return False
def main(*args):
- parser = ArgumentParser(
- description="Python sqlite3 CLI",
- color=True,
- )
+ parser = ArgumentParser(description="Python sqlite3 CLI")
parser.add_argument(
"filename", type=str, default=":memory:", nargs="?",
help=(
- "SQLite database to open (defaults to ':memory:'). "
+ "SQLite database to open (defaults to `:memory:`). "
"A new database is created if the file does not previously exist."
),
)
@@ -86,7 +99,7 @@ def main(*args):
"sql", type=str, nargs="?",
help=(
"An SQL query to execute. "
- "Any returned rows are printed to stdout."
+ "Any returned rows are printed to `stdout`."
),
)
parser.add_argument(
@@ -113,22 +126,26 @@ def main(*args):
Each command will be run using execute() on the cursor.
Type ".help" for more information; type ".quit" or {eofkey} to quit.
""").strip()
- sys.ps1 = "sqlite> "
- sys.ps2 = " ... "
+
+ theme = get_theme()
+ s = theme.syntax
+
+ # Use RL_PROMPT_START_IGNORE (\001) and RL_PROMPT_END_IGNORE (\002) to
+ # bracket non-printing characters. This tells readline to ignore them
+ # when calculating screen space for redisplay during history scrolling.
+ sys.ps1 = f"\001{s.prompt}\002sqlite> \001{s.reset}\002"
+ sys.ps2 = f"\001{s.prompt}\002 ... \001{s.reset}\002"
con = sqlite3.connect(args.filename, isolation_level=None)
try:
if args.sql:
# SQL statement provided on the command-line; execute it directly.
- execute(con, args.sql, suppress_errors=False)
+ execute(con, args.sql, suppress_errors=False, theme=theme)
else:
# No SQL provided; start the REPL.
- console = SqliteInteractiveConsole(con)
- try:
- import readline # noqa: F401
- except ImportError:
- pass
- console.interact(banner, exitmsg="")
+ with completer(con):
+ console = SqliteInteractiveConsole(con, use_color=True)
+ console.interact(banner, exitmsg="")
finally:
con.close()
diff --git a/Lib/sqlite3/_completer.py b/Lib/sqlite3/_completer.py
new file mode 100644
index 000000000..ba580f968
--- /dev/null
+++ b/Lib/sqlite3/_completer.py
@@ -0,0 +1,111 @@
+from _sqlite3 import OperationalError
+from contextlib import contextmanager
+
+try:
+ from _sqlite3 import SQLITE_KEYWORDS
+except ImportError:
+ SQLITE_KEYWORDS = ()
+
+CLI_COMMANDS = ('.quit', '.help', '.version')
+
+_completion_matches = []
+
+
+def _complete(con, text, state):
+ global _completion_matches
+
+ if state == 0:
+ if text.startswith("."):
+ _completion_matches = [
+ c + " " for c in CLI_COMMANDS if c.startswith(text)
+ ]
+ else:
+ text_upper = text.upper()
+ _completion_matches = [
+ c + " " for c in SQLITE_KEYWORDS if c.startswith(text_upper)
+ ]
+
+ cursor = con.cursor()
+ schemata = tuple(row[1] for row
+ in cursor.execute("PRAGMA database_list"))
+ # tables, indexes, triggers, and views
+ # escape '_' which can appear in attached database names
+ select_clauses = (
+ f"""\
+ SELECT name || ' ' FROM \"{schema}\".sqlite_master
+ WHERE name LIKE REPLACE(:text, '_', '^_') || '%' ESCAPE '^'"""
+ for schema in schemata
+ )
+ _completion_matches.extend(
+ row[0]
+ for row in cursor.execute(
+ " UNION ".join(select_clauses), {"text": text}
+ )
+ )
+ # columns
+ try:
+ select_clauses = (
+ f"""\
+ SELECT pti.name || ' ' FROM "{schema}".sqlite_master AS sm
+ JOIN pragma_table_xinfo(sm.name,'{schema}') AS pti
+ WHERE sm.type='table' AND
+ pti.name LIKE REPLACE(:text, '_', '^_') || '%' ESCAPE '^'"""
+ for schema in schemata
+ )
+ _completion_matches.extend(
+ row[0]
+ for row in cursor.execute(
+ " UNION ".join(select_clauses), {"text": text}
+ )
+ )
+ except OperationalError:
+ # skip on SQLite<3.16.0 where pragma table-valued function is
+ # not supported yet
+ pass
+ # functions
+ try:
+ _completion_matches.extend(
+ row[0] for row in cursor.execute("""\
+ SELECT DISTINCT UPPER(name) || '('
+ FROM pragma_function_list()
+ WHERE name NOT IN ('->', '->>') AND
+ name LIKE REPLACE(:text, '_', '^_') || '%' ESCAPE '^'""",
+ {"text": text},
+ )
+ )
+ except OperationalError:
+ # skip on SQLite<3.30.0 where function_list is not supported yet
+ pass
+ # schemata
+ text_lower = text.lower()
+ _completion_matches.extend(c for c in schemata
+ if c.lower().startswith(text_lower))
+ _completion_matches = sorted(set(_completion_matches))
+ try:
+ return _completion_matches[state]
+ except IndexError:
+ return None
+
+
+@contextmanager
+def completer(con):
+ try:
+ import readline
+ except ImportError:
+ yield
+ return
+
+ old_completer = readline.get_completer()
+ def complete(text, state):
+ return _complete(con, text, state)
+ try:
+ readline.set_completer(complete)
+ if readline.backend == "editline":
+ # libedit uses "^I" instead of "tab"
+ command_string = "bind ^I rl_complete"
+ else:
+ command_string = "tab: complete"
+ readline.parse_and_bind(command_string)
+ yield
+ finally:
+ readline.set_completer(old_completer)
diff --git a/Lib/sqlite3/dump.py b/Lib/sqlite3/dump.py
index 57e6a3b4f..95e14a665 100644
--- a/Lib/sqlite3/dump.py
+++ b/Lib/sqlite3/dump.py
@@ -11,8 +11,12 @@ def _quote_name(name):
return '"{0}"'.format(name.replace('"', '""'))
+def _escape_single_quotes(value):
+ return value.replace("'", "''")
+
+
def _quote_value(value):
- return "'{0}'".format(value.replace("'", "''"))
+ return "'{0}'".format(_escape_single_quotes(value))
def _iterdump(connection, *, filter=None):
@@ -80,11 +84,12 @@ def _iterdump(connection, *, filter=None):
table_name_ident = _quote_name(table_name)
res = cu.execute(f'PRAGMA table_info({table_name_ident})')
column_names = [str(table_info[1]) for table_info in res.fetchall()]
- q = "SELECT 'INSERT INTO {0} VALUES('{1}')' FROM {0};".format(
- table_name_ident,
+ q = "SELECT 'INSERT INTO {0} VALUES('{1}')' FROM {2};".format(
+ _escape_single_quotes(table_name_ident),
"','".join(
"||quote({0})||".format(_quote_name(col)) for col in column_names
- )
+ ),
+ table_name_ident,
)
query_res = cu.execute(q)
for row in query_res:
diff --git a/Lib/test/test_dbm.py b/Lib/test/test_dbm.py
index e68eca295..24147a8ac 100644
--- a/Lib/test/test_dbm.py
+++ b/Lib/test/test_dbm.py
@@ -67,7 +67,7 @@ class AnyDBMTestCase:
return keys
def test_error(self):
- self.assertTrue(issubclass(self.module.error, OSError))
+ self.assertIsSubclass(self.module.error, OSError)
def test_anydbm_not_existing(self):
self.assertRaises(dbm.error, dbm.open, _fname)
@@ -136,6 +136,67 @@ class AnyDBMTestCase:
assert(f[key] == b"Python:")
f.close()
+ def test_anydbm_readonly_reorganize(self):
+ self.init_db()
+ with dbm.open(_fname, 'r') as d:
+ # Early stopping.
+ if not hasattr(d, 'reorganize'):
+ self.skipTest("method reorganize not available this dbm submodule")
+
+ self.assertRaises(dbm.error, lambda: d.reorganize())
+
+ def test_anydbm_reorganize_not_changed_content(self):
+ self.init_db()
+ with dbm.open(_fname, 'c') as d:
+ # Early stopping.
+ if not hasattr(d, 'reorganize'):
+ self.skipTest("method reorganize not available this dbm submodule")
+
+ keys_before = sorted(d.keys())
+ values_before = [d[k] for k in keys_before]
+ d.reorganize()
+ keys_after = sorted(d.keys())
+ values_after = [d[k] for k in keys_before]
+ self.assertEqual(keys_before, keys_after)
+ self.assertEqual(values_before, values_after)
+
+ def test_anydbm_reorganize_decreased_size(self):
+
+ def _calculate_db_size(db_path):
+ if os.path.isfile(db_path):
+ return os.path.getsize(db_path)
+ total_size = 0
+ for root, _, filenames in os.walk(db_path):
+ for filename in filenames:
+ file_path = os.path.join(root, filename)
+ total_size += os.path.getsize(file_path)
+ return total_size
+
+ # This test requires relatively large databases to reliably show difference in size before and after reorganizing.
+ with dbm.open(_fname, 'n') as f:
+ # Early stopping.
+ if not hasattr(f, 'reorganize'):
+ self.skipTest("method reorganize not available this dbm submodule")
+
+ for k in self._dict:
+ f[k.encode('ascii')] = self._dict[k] * 100000
+ db_keys = list(f.keys())
+
+ # Make sure to calculate size of database only after file is closed to ensure file content are flushed to disk.
+ size_before = _calculate_db_size(os.path.dirname(_fname))
+
+ # Delete some elements from the start of the database.
+ keys_to_delete = db_keys[:len(db_keys) // 2]
+ with dbm.open(_fname, 'c') as f:
+ for k in keys_to_delete:
+ del f[k]
+ f.reorganize()
+
+ # Make sure to calculate size of database only after file is closed to ensure file content are flushed to disk.
+ size_after = _calculate_db_size(os.path.dirname(_fname))
+
+ self.assertLess(size_after, size_before)
+
def test_open_with_bytes(self):
dbm.open(os.fsencode(_fname), "c").close()
@@ -261,5 +322,6 @@ for mod in dbm_iterator():
cls = unittest.skip("TODO: RUSTPYTHON; sqlite3 file locking on Windows")(cls)
globals()[testname] = cls
+
if __name__ == "__main__":
unittest.main()
diff --git a/Lib/test/test_dbm_dumb.py b/Lib/test/test_dbm_dumb.py
index 672f90922..d977a8187 100644
--- a/Lib/test/test_dbm_dumb.py
+++ b/Lib/test/test_dbm_dumb.py
@@ -114,6 +114,16 @@ class DumbDBMTestCase(unittest.TestCase):
with contextlib.closing(dumbdbm.open(_fname)) as f:
self.assertEqual(f[b'1'], b'hello2')
+ def test_reorganize_persists_changed_offsets(self):
+ with dumbdbm.open(_fname, 'n') as f:
+ f[b'deleted'] = b'x'
+ f[b'retained'] = b'value'
+ del f[b'deleted']
+ f.reorganize()
+
+ with dumbdbm.open(_fname, 'r') as f:
+ self.assertEqual(f[b'retained'], b'value')
+
def test_str_read(self):
self.init_db()
with contextlib.closing(dumbdbm.open(_fname, 'r')) as f:
diff --git a/Lib/test/test_ensurepip.py b/Lib/test/test_ensurepip.py
index a8bfbcd4a..20a56ed71 100644
--- a/Lib/test/test_ensurepip.py
+++ b/Lib/test/test_ensurepip.py
@@ -13,6 +13,12 @@ import ensurepip
import ensurepip._uninstall
+if sys.implementation.cache_tag is None:
+ COMPILE_OPT = ["--no-compile"]
+else:
+ COMPILE_OPT = []
+
+
class TestPackages(unittest.TestCase):
def touch(self, directory, filename):
fullname = os.path.join(directory, filename)
@@ -70,6 +76,11 @@ class EnsurepipMixin:
self.run_pip.return_value = 0
self.addCleanup(run_pip_patch.stop)
+ # Allow testing on zlib-less platforms by avoiding the check for zlib in _bootstrap()
+ zlib_patch = unittest.mock.patch.dict('sys.modules', {'zlib': unittest.mock.MagicMock()})
+ zlib_patch.start()
+ self.addCleanup(zlib_patch.stop)
+
# Avoid side effects on the actual os module
real_devnull = os.devnull
os_patch = unittest.mock.patch("ensurepip.os")
@@ -90,7 +101,7 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "pip",
+ unittest.mock.ANY, *COMPILE_OPT, "pip",
],
unittest.mock.ANY,
)
@@ -104,7 +115,7 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "--root", "/foo/bar/",
+ unittest.mock.ANY, "--root", "/foo/bar/", *COMPILE_OPT,
"pip",
],
unittest.mock.ANY,
@@ -116,7 +127,7 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "--user", "pip",
+ unittest.mock.ANY, "--user", *COMPILE_OPT, "pip",
],
unittest.mock.ANY,
)
@@ -127,7 +138,7 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "--upgrade", "pip",
+ unittest.mock.ANY, "--upgrade", *COMPILE_OPT, "pip",
],
unittest.mock.ANY,
)
@@ -138,7 +149,7 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "-v", "pip",
+ unittest.mock.ANY, "-v", *COMPILE_OPT, "pip",
],
unittest.mock.ANY,
)
@@ -149,7 +160,7 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "-vv", "pip",
+ unittest.mock.ANY, "-vv", *COMPILE_OPT, "pip",
],
unittest.mock.ANY,
)
@@ -160,7 +171,7 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "-vvv", "pip",
+ unittest.mock.ANY, "-vvv", *COMPILE_OPT, "pip",
],
unittest.mock.ANY,
)
@@ -195,6 +206,16 @@ class TestBootstrap(EnsurepipMixin, unittest.TestCase):
ensurepip.bootstrap()
self.assertEqual(self.os_environ["PIP_CONFIG_FILE"], os.devnull)
+ def test_missing_zlib(self):
+ with unittest.mock.patch.dict('sys.modules', {'zlib': None}):
+ with self.assertRaises(ModuleNotFoundError) as cm:
+ ensurepip.bootstrap()
+
+ error_msg = str(cm.exception)
+ self.assertIn("ensurepip requires the standard library module 'zlib'", error_msg)
+
+ self.assertFalse(self.run_pip.called)
+
@contextlib.contextmanager
def fake_pip(version=ensurepip.version()):
if version is None:
@@ -307,7 +328,7 @@ class TestBootstrappingMainFunction(EnsurepipMixin, unittest.TestCase):
self.run_pip.assert_called_once_with(
[
"install", "--no-cache-dir", "--no-index", "--find-links",
- unittest.mock.ANY, "pip",
+ unittest.mock.ANY, *COMPILE_OPT, "pip",
],
unittest.mock.ANY,
)
diff --git a/Lib/test/test_pyclbr.py b/Lib/test/test_pyclbr.py
index 17a19b213..b5ec41b17 100644
--- a/Lib/test/test_pyclbr.py
+++ b/Lib/test/test_pyclbr.py
@@ -11,9 +11,6 @@ from types import FunctionType, MethodType, BuiltinFunctionType
import pyclbr
from unittest import TestCase, main as unittest_main
from test.test_importlib import util as test_importlib_util
-import warnings
-
-import unittest # XXX: RUSTPYTHON; importing to be able to skip tests
StaticMethodType = type(staticmethod(lambda: None))
@@ -248,9 +245,6 @@ class PyclbrTest(TestCase):
# These were once some of the longest modules.
cm('random', ignore=('Random',)) # from _random import Random as CoreGenerator
cm('pickle', ignore=('partial', 'PickleBuffer'))
- with warnings.catch_warnings():
- warnings.simplefilter('ignore', DeprecationWarning)
- cm('sre_parse', ignore=('dump', 'groups', 'pos')) # from sre_constants import *; property
with temporary_main_spec():
cm(
'pdb',
@@ -258,7 +252,8 @@ class PyclbrTest(TestCase):
ignore=('Union', '_ModuleTarget', '_ScriptTarget', '_ZipTarget', 'curframe_locals',
'_InteractState', 'rlcompleter'),
)
- cm('pydoc', ignore=('input', 'output',)) # properties
+ cm('pydoc', ignore=('input', 'output', # properties
+ 'getpager', 'plainpager', )) # aliases
# Tests for modules inside packages
cm('email.parser')
diff --git a/Lib/test/test_shelve.py b/Lib/test/test_shelve.py
index 08c6562f2..5f6a030e0 100644
--- a/Lib/test/test_shelve.py
+++ b/Lib/test/test_shelve.py
@@ -1,10 +1,11 @@
+import array
import unittest
import dbm
import shelve
import pickle
import os
-from test.support import os_helper
+from test.support import import_helper, os_helper, subTests
from collections.abc import MutableMapping
from test.test_dbm import dbm_iterator
@@ -165,6 +166,255 @@ class TestCase(unittest.TestCase):
with shelve.Shelf({}) as s:
self.assertEqual(s._protocol, pickle.DEFAULT_PROTOCOL)
+ def test_custom_serializer_and_deserializer(self):
+ os.mkdir(self.dirname)
+ self.addCleanup(os_helper.rmtree, self.dirname)
+
+ def serializer(obj, protocol):
+ if isinstance(obj, (bytes, bytearray, str)):
+ if protocol == 5:
+ if isinstance(obj, bytearray):
+ return bytes(obj) # DBM backends expect bytes
+ return obj
+ return type(obj).__name__
+ elif isinstance(obj, array.array):
+ return obj.tobytes()
+ raise TypeError(f"Unsupported type for serialization: {type(obj)}")
+
+ def deserializer(data):
+ if isinstance(data, (bytes, bytearray, str)):
+ return data.decode("utf-8")
+ elif isinstance(data, array.array):
+ return array.array("b", data)
+ raise TypeError(
+ f"Unsupported type for deserialization: {type(data)}"
+ )
+
+ for proto in range(pickle.HIGHEST_PROTOCOL + 1):
+ with self.subTest(proto=proto), shelve.open(
+ self.fn,
+ protocol=proto,
+ serializer=serializer,
+ deserializer=deserializer
+ ) as s:
+ bar = "bar"
+ bytes_data = b"Hello, world!"
+ bytearray_data = bytearray(b"\x00\x01\x02\x03\x04")
+ array_data = array.array("i", [1, 2, 3, 4, 5])
+
+ s["foo"] = bar
+ s["bytes_data"] = bytes_data
+ s["bytearray_data"] = bytearray_data
+ s["array_data"] = array_data
+
+ if proto == 5:
+ self.assertEqual(s["foo"], str(bar))
+ self.assertEqual(s["bytes_data"], "Hello, world!")
+ self.assertEqual(
+ s["bytearray_data"], bytearray_data.decode()
+ )
+ self.assertEqual(
+ s["array_data"], array_data.tobytes().decode()
+ )
+ else:
+ self.assertEqual(s["foo"], "str")
+ self.assertEqual(s["bytes_data"], "bytes")
+ self.assertEqual(s["bytearray_data"], "bytearray")
+ self.assertEqual(
+ s["array_data"], array_data.tobytes().decode()
+ )
+
+ @subTests("serialized", [None, ["invalid type"]])
+ def test_custom_invalid_serializer(self, serialized):
+ test_dir = f"{self.dirname}_{id(serialized)}"
+ os.mkdir(test_dir)
+ self.addCleanup(os_helper.rmtree, test_dir)
+ test_fn = os.path.join(test_dir, "shelftemp.db")
+
+ def serializer(obj, protocol=None):
+ return serialized
+
+ def deserializer(data):
+ return data.decode("utf-8")
+
+ # Since the serializer returns an invalid type or None,
+ # dbm.error is raised by dbm.sqlite3 and TypeError is raised
+ # by other backends.
+ with self.assertRaises((TypeError, dbm.error)):
+ with shelve.open(test_fn, serializer=serializer,
+ deserializer=deserializer) as s:
+ s["foo"] = "bar"
+
+ def test_custom_incomplete_deserializer(self):
+ os.mkdir(self.dirname)
+ self.addCleanup(os_helper.rmtree, self.dirname)
+
+ def serializer(obj, protocol=None):
+ return type(obj).__name__.encode("utf-8")
+
+ def deserializer(data):
+ pass
+
+ with shelve.open(self.fn, serializer=serializer,
+ deserializer=deserializer) as s:
+ s["foo"] = "bar"
+ self.assertNotEqual(s["foo"], "bar")
+ self.assertIsNone(s["foo"])
+
+ def test_custom_serializer_and_deserializer_bsd_db_shelf(self):
+ berkeleydb = import_helper.import_module("berkeleydb")
+ os.mkdir(self.dirname)
+ self.addCleanup(os_helper.rmtree, self.dirname)
+
+ def serializer(obj, protocol=None):
+ data = obj.__class__.__name__
+ if protocol == 5:
+ data = str(len(data))
+ return data.encode("utf-8")
+
+ def deserializer(data):
+ return data.decode("utf-8")
+
+ def type_name_len(obj):
+ return f"{(len(type(obj).__name__))}"
+
+ for proto in range(pickle.HIGHEST_PROTOCOL + 1):
+ with self.subTest(proto=proto), shelve.BsdDbShelf(
+ berkeleydb.btopen(self.fn),
+ protocol=proto,
+ serializer=serializer,
+ deserializer=deserializer,
+ ) as s:
+ bar = "bar"
+ bytes_obj = b"Hello, world!"
+ bytearray_obj = bytearray(b"\x00\x01\x02\x03\x04")
+ arr_obj = array.array("i", [1, 2, 3, 4, 5])
+
+ s["foo"] = bar
+ s["bytes_data"] = bytes_obj
+ s["bytearray_data"] = bytearray_obj
+ s["array_data"] = arr_obj
+
+ if proto == 5:
+ self.assertEqual(s["foo"], type_name_len(bar))
+ self.assertEqual(s["bytes_data"], type_name_len(bytes_obj))
+ self.assertEqual(s["bytearray_data"],
+ type_name_len(bytearray_obj))
+ self.assertEqual(s["array_data"], type_name_len(arr_obj))
+
+ k, v = s.set_location(b"foo")
+ self.assertEqual(k, "foo")
+ self.assertEqual(v, type_name_len(bar))
+
+ k, v = s.previous()
+ self.assertEqual(k, "bytes_data")
+ self.assertEqual(v, type_name_len(bytes_obj))
+
+ k, v = s.previous()
+ self.assertEqual(k, "bytearray_data")
+ self.assertEqual(v, type_name_len(bytearray_obj))
+
+ k, v = s.previous()
+ self.assertEqual(k, "array_data")
+ self.assertEqual(v, type_name_len(arr_obj))
+
+ k, v = s.next()
+ self.assertEqual(k, "bytearray_data")
+ self.assertEqual(v, type_name_len(bytearray_obj))
+
+ k, v = s.next()
+ self.assertEqual(k, "bytes_data")
+ self.assertEqual(v, type_name_len(bytes_obj))
+
+ k, v = s.first()
+ self.assertEqual(k, "array_data")
+ self.assertEqual(v, type_name_len(arr_obj))
+ else:
+ k, v = s.set_location(b"foo")
+ self.assertEqual(k, "foo")
+ self.assertEqual(v, "str")
+
+ k, v = s.previous()
+ self.assertEqual(k, "bytes_data")
+ self.assertEqual(v, "bytes")
+
+ k, v = s.previous()
+ self.assertEqual(k, "bytearray_data")
+ self.assertEqual(v, "bytearray")
+
+ k, v = s.previous()
+ self.assertEqual(k, "array_data")
+ self.assertEqual(v, "array")
+
+ k, v = s.next()
+ self.assertEqual(k, "bytearray_data")
+ self.assertEqual(v, "bytearray")
+
+ k, v = s.next()
+ self.assertEqual(k, "bytes_data")
+ self.assertEqual(v, "bytes")
+
+ k, v = s.first()
+ self.assertEqual(k, "array_data")
+ self.assertEqual(v, "array")
+
+ self.assertEqual(s["foo"], "str")
+ self.assertEqual(s["bytes_data"], "bytes")
+ self.assertEqual(s["bytearray_data"], "bytearray")
+ self.assertEqual(s["array_data"], "array")
+
+ def test_custom_incomplete_deserializer_bsd_db_shelf(self):
+ berkeleydb = import_helper.import_module("berkeleydb")
+ os.mkdir(self.dirname)
+ self.addCleanup(os_helper.rmtree, self.dirname)
+
+ def serializer(obj, protocol=None):
+ return type(obj).__name__.encode("utf-8")
+
+ def deserializer(data):
+ pass
+
+ with shelve.BsdDbShelf(berkeleydb.btopen(self.fn),
+ serializer=serializer,
+ deserializer=deserializer) as s:
+ s["foo"] = "bar"
+ self.assertIsNone(s["foo"])
+ self.assertNotEqual(s["foo"], "bar")
+
+ def test_custom_incomplete_serializer_bsd_db_shelf(self):
+ berkeleydb = import_helper.import_module("berkeleydb")
+ os.mkdir(self.dirname)
+ self.addCleanup(os_helper.rmtree, self.dirname)
+
+ def serializer(obj, protocol=None):
+ pass
+
+ def deserializer(data):
+ return data.decode("utf-8")
+
+ with shelve.BsdDbShelf(berkeleydb.btopen(self.fn),
+ serializer=serializer,
+ deserializer=deserializer) as s:
+ s["foo"] = "bar"
+ self.assertEqual(s["foo"], "")
+ self.assertNotEqual(s["foo"], "bar")
+
+ def test_missing_custom_deserializer(self):
+ def serializer(obj, protocol=None):
+ pass
+
+ kwargs = dict(protocol=2, writeback=False, serializer=serializer)
+ self.assertRaises(shelve.ShelveError, shelve.Shelf, {}, **kwargs)
+ self.assertRaises(shelve.ShelveError, shelve.BsdDbShelf, {}, **kwargs)
+
+ def test_missing_custom_serializer(self):
+ def deserializer(data):
+ pass
+
+ kwargs = dict(protocol=2, writeback=False, deserializer=deserializer)
+ self.assertRaises(shelve.ShelveError, shelve.Shelf, {}, **kwargs)
+ self.assertRaises(shelve.ShelveError, shelve.BsdDbShelf, {}, **kwargs)
+
class TestShelveBase:
type2test = shelve.Shelf
diff --git a/Lib/test/test_sqlite3/test_cli.py b/Lib/test/test_sqlite3/test_cli.py
index a03d7cbe1..0d235d90f 100644
--- a/Lib/test/test_sqlite3/test_cli.py
+++ b/Lib/test/test_sqlite3/test_cli.py
@@ -1,17 +1,26 @@
"""sqlite3 CLI tests."""
import sqlite3
+import sys
+import textwrap
import unittest
+import unittest.mock
+import os
from sqlite3.__main__ import main as cli
+from test.support.import_helper import import_module
from test.support.os_helper import TESTFN, unlink
+from test.support.pty_helper import run_pty
from test.support import (
captured_stdout,
captured_stderr,
captured_stdin,
- force_not_colorized,
+ force_not_colorized_test_class,
+ requires_subprocess,
+ verbose,
)
+@force_not_colorized_test_class
class CommandLineInterface(unittest.TestCase):
def _do_test(self, *args, expect_success=True):
@@ -37,7 +46,6 @@ class CommandLineInterface(unittest.TestCase):
self.assertEqual(out, "")
return err
- @force_not_colorized
def test_cli_help(self):
out = self.expect_success("-h")
self.assertIn("usage: ", out)
@@ -69,10 +77,11 @@ class CommandLineInterface(unittest.TestCase):
self.assertIn("(0,)", out)
+@force_not_colorized_test_class
class InteractiveSession(unittest.TestCase):
MEMORY_DB_MSG = "Connected to a transient in-memory database"
- PS1 = "sqlite> "
- PS2 = "... "
+ PS1 = "\001\002sqlite> \001\002"
+ PS2 = "\001\002 ... \001\002"
def run_cli(self, *args, commands=()):
with (
@@ -129,7 +138,7 @@ class InteractiveSession(unittest.TestCase):
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 2)
self.assertEqual(out.count(self.PS2), 0)
- self.assertIn("Error", err)
+ self.assertIn('Error: unknown command: "', err)
# test "unknown_command" is pointed out in the error message
self.assertIn("unknown_command", err)
@@ -190,6 +199,247 @@ class InteractiveSession(unittest.TestCase):
out, _ = self.run_cli(TESTFN, commands=("SELECT count(t) FROM t;",))
self.assertIn("(0,)\n", out)
+ def test_color(self):
+ with unittest.mock.patch("_colorize.can_colorize", return_value=True):
+ out, err = self.run_cli(commands="TEXT\n")
+ self.assertIn("\x01\x1b[1;35m\x02sqlite> \x01\x1b[0m\x02", out)
+ self.assertIn("\x01\x1b[1;35m\x02 ... \x01\x1b[0m\x02\x01\x1b", out)
+ out, err = self.run_cli(commands=("sel;",))
+ self.assertIn('\x1b[1;35mOperationalError (SQLITE_ERROR)\x1b[0m: '
+ '\x1b[35mnear "sel": syntax error\x1b[0m', err)
+
+
+@requires_subprocess()
+@force_not_colorized_test_class
+class Completion(unittest.TestCase):
+ # run_pty() creates a real terminal environment, where sqlite3 CLI
+ # SqliteInteractiveConsole invokes GNU Readline for input. Readline's
+ # _rl_strip_prompt() strips \001 and \002 from the output, so test
+ # assertions use the plain prompt.
+ PS1 = "sqlite> "
+
+ @classmethod
+ def setUpClass(cls):
+ readline = import_module("readline")
+ if readline.backend == "editline":
+ raise unittest.SkipTest("libedit readline is not supported")
+ if sys.platform.startswith("openbsd"):
+ # OpenBSD's readline does not honor "completion-query-items 0".
+ raise unittest.SkipTest("OpenBSD readline hangs on completion")
+
+ def write_input(self, input_, env=None):
+ script = textwrap.dedent("""
+ import readline
+ from sqlite3.__main__ import main
+
+ # Configure readline to ...:
+ # - hide control sequences surrounding each candidate
+ # - hide "Display all xxx possibilities? (y or n)"
+ # - show candidates one per line
+ readline.parse_and_bind("set colored-completion-prefix off")
+ readline.parse_and_bind("set completion-query-items 0")
+ readline.parse_and_bind("set page-completions off")
+ readline.parse_and_bind("set completion-display-width 0")
+
+ main()
+ """)
+ return run_pty(script, input_, env)
+
+ def test_complete_sql_keywords(self):
+ _sqlite3 = import_module("_sqlite3")
+ if not hasattr(_sqlite3, "SQLITE_KEYWORDS"):
+ raise unittest.SkipTest("unable to determine SQLite keywords")
+
+ # List candidates starting with 'S', there should be multiple matches.
+ input_ = b"S\t\tEL\t 1;\n.quit\n"
+ output = self.write_input(input_)
+ self.assertIn(b"SELECT", output)
+ self.assertIn(b"SET", output)
+ self.assertIn(b"SAVEPOINT", output)
+ self.assertIn(b"(1,)", output)
+
+ # Keywords are completed in upper case for even lower case user input.
+ input_ = b"sel\t\t 1;\n.quit\n"
+ output = self.write_input(input_)
+ self.assertIn(b"SELECT", output)
+ self.assertIn(b"(1,)", output)
+
+ # .commands are completed without changing case
+ input_ = b".ver\t\n.quit\n"
+ output = self.write_input(input_)
+ self.assertIn(b".version", output)
+
+ def test_complete_table_indexes_triggers_views(self):
+ input_ = textwrap.dedent("""\
+ CREATE TABLE _Table (id);
+ CREATE INDEX _Index ON _table (id);
+ CREATE TRIGGER _Trigger BEFORE INSERT
+ ON _Table BEGIN SELECT 1; END;
+ CREATE VIEW _View AS SELECT 1;
+
+ CREATE TEMP TABLE _Temp_table (id);
+ CREATE INDEX temp._Temp_index ON _Temp_table (id);
+ CREATE TEMP TRIGGER _Temp_trigger BEFORE INSERT
+ ON _Table BEGIN SELECT 1; END;
+ CREATE TEMP VIEW _Temp_view AS SELECT 1;
+
+ ATTACH ':memory:' AS attached;
+ CREATE TABLE attached._Attached_table (id);
+ CREATE INDEX attached._Attached_index ON _Attached_table (id);
+ CREATE TRIGGER attached._Attached_trigger BEFORE INSERT
+ ON _Attached_table BEGIN SELECT 1; END;
+ CREATE VIEW attached._Attached_view AS SELECT 1;
+
+ SELECT id FROM _\t\tta\t;
+ .quit\n""").encode()
+ output = self.write_input(input_)
+ lines = output.decode().splitlines()
+ indices = [i for i, line in enumerate(lines)
+ if line.startswith(self.PS1)]
+ start, end = indices[-3], indices[-2]
+ candidates = [l.strip() for l in lines[start+1:end]]
+ self.assertEqual(candidates,
+ [
+ "_Attached_index",
+ "_Attached_table",
+ "_Attached_trigger",
+ "_Attached_view",
+ "_Index",
+ "_Table",
+ "_Temp_index",
+ "_Temp_table",
+ "_Temp_trigger",
+ "_Temp_view",
+ "_Trigger",
+ "_View",
+ ],
+ )
+ start, end = indices[-2], indices[-1]
+ # direct match with '_Table' completed, no candidates displayed
+ candidates = [l.strip() for l in lines[start+1:end]]
+ self.assertEqual(len(candidates), 0)
+
+ @unittest.skipIf(sqlite3.sqlite_version_info < (3, 16, 0),
+ "PRAGMA table-valued function is not available until "
+ "SQLite 3.16.0")
+ def test_complete_columns(self):
+ input_ = textwrap.dedent("""\
+ CREATE TABLE _table (_col_table);
+ CREATE TEMP TABLE _temp_table (_col_temp);
+ ATTACH ':memory:' AS attached;
+ CREATE TABLE attached._attached_table (_col_attached);
+
+ SELECT _col_\t\tta\tFROM _table;
+ .quit\n""").encode()
+ output = self.write_input(input_)
+ lines = output.decode().splitlines()
+ indices = [
+ i for i, line in enumerate(lines) if line.startswith(self.PS1)
+ ]
+ start, end = indices[-3], indices[-2]
+ candidates = [l.strip() for l in lines[start+1:end]]
+
+ self.assertEqual(
+ candidates, ["_col_attached", "_col_table", "_col_temp"]
+ )
+
+ @unittest.skipIf(sqlite3.sqlite_version_info < (3, 30, 0),
+ "PRAGMA function_list is not available until "
+ "SQLite 3.30.0")
+ def test_complete_functions(self):
+ input_ = b"SELECT AV\t1);\n.quit\n"
+ output = self.write_input(input_)
+ self.assertIn(b"AVG(1);", output)
+ self.assertIn(b"(1.0,)", output)
+
+ # Functions are completed in upper case for even lower case user input.
+ input_ = b"SELECT av\t1);\n.quit\n"
+ output = self.write_input(input_)
+ self.assertIn(b"AVG(1);", output)
+ self.assertIn(b"(1.0,)", output)
+
+ def test_complete_schemata(self):
+ input_ = textwrap.dedent("""\
+ ATTACH ':memory:' AS MixedCase;
+ -- Test '_' is escaped in Like pattern filtering
+ ATTACH ':memory:' AS _underscore;
+ -- Let database_list pragma have a 'temp' schema entry
+ CREATE TEMP TABLE _table (id);
+
+ SELECT * FROM \t\tmIX\t.sqlite_master;
+ SELECT * FROM _und\t.sqlite_master;
+ .quit\n""").encode()
+ output = self.write_input(input_)
+ lines = output.decode().splitlines()
+ indices = [
+ i for i, line in enumerate(lines) if line.startswith(self.PS1)
+ ]
+ start, end = indices[-4], indices[-3]
+ candidates = [l.strip() for l in lines[start+1:end]]
+ self.assertIn("MixedCase", candidates)
+ self.assertIn("_underscore", candidates)
+ self.assertIn("main", candidates)
+ self.assertIn("temp", candidates)
+
+ @unittest.skipIf(sys.platform.startswith(("freebsd", "dragonfly", "sunos")),
+ "Two actual tabs are inserted when there are no matching"
+ " completions in the pseudo-terminal opened by run_pty()"
+ " on this platform")
+ def test_complete_no_match(self):
+ input_ = b"xyzzy\t\t\b\b\b\b\b\b\b.quit\n"
+ # Set NO_COLOR to disable coloring for self.PS1.
+ output = self.write_input(input_, env={**os.environ, "NO_COLOR": "1"})
+ lines = output.decode().splitlines()
+ indices = (
+ i for i, line in enumerate(lines, 1)
+ if line.startswith(f"{self.PS1}xyzzy")
+ )
+ line_num = next(indices, -1)
+ self.assertNotEqual(line_num, -1)
+ # Completions occupy lines, assert no extra lines when there is nothing
+ # to complete.
+ self.assertEqual(line_num, len(lines))
+
+ def test_complete_no_input(self):
+ script = textwrap.dedent("""
+ import readline
+ from sqlite3.__main__ import main
+
+ # Configure readline to ...:
+ # - hide control sequences surrounding each candidate
+ # - hide "Display all xxx possibilities? (y or n)"
+ # - hide "--More--"
+ # - show candidates one per line
+ readline.parse_and_bind("set colored-completion-prefix off")
+ readline.parse_and_bind("set colored-stats off")
+ readline.parse_and_bind("set completion-query-items 0")
+ readline.parse_and_bind("set page-completions off")
+ readline.parse_and_bind("set completion-display-width 0")
+ readline.parse_and_bind("set show-all-if-ambiguous off")
+ readline.parse_and_bind("set show-all-if-unmodified off")
+
+ main()
+ """)
+ input_ = b"\t\t.quit\n"
+ output = run_pty(script, input_, env={**os.environ, "NO_COLOR": "1"})
+ try:
+ lines = output.decode().splitlines()
+ indices = [
+ i for i, line in enumerate(lines)
+ if line.startswith(self.PS1)
+ ]
+ self.assertEqual(len(indices), 2)
+ start, end = indices
+ candidates = [l.strip() for l in lines[start+1:end]]
+ self.assertEqual(candidates, sorted(candidates))
+ except:
+ if verbose:
+ print(' PTY output: '.center(30, '-'))
+ print(output.decode(errors='replace'))
+ print(' end PTY output '.center(30, '-'))
+ raise
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index ef8acd0f3..d4602ac43 100644
--- a/Lib/test/test_sqlite3/test_dbapi.py
+++ b/Lib/test/test_sqlite3/test_dbapi.py
@@ -551,17 +551,9 @@ class ConnectionTests(unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; deprecation warning not emitted for positional args
def test_connect_positional_arguments(self):
- regex = (
- r"Passing more than 1 positional argument to sqlite3.connect\(\)"
- " is deprecated. Parameters 'timeout', 'detect_types', "
- "'isolation_level', 'check_same_thread', 'factory', "
- "'cached_statements' and 'uri' will become keyword-only "
- "parameters in Python 3.15."
- )
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
- cx = sqlite.connect(":memory:", 1.0)
- cx.close()
- self.assertEqual(cm.filename, __file__)
+ with self.assertRaisesRegex(TypeError,
+ r'connect\(\) takes at most 1 positional arguments'):
+ sqlite.connect(":memory:", 1.0)
@unittest.expectedFailure # TODO: RUSTPYTHON; ResourceWarning not emitted
def test_connection_resource_warning(self):
@@ -1409,6 +1401,18 @@ class BlobTests(unittest.TestCase):
actual = self.cx.execute("select b from test").fetchone()[0]
self.assertEqual(actual, expected)
+ def test_blob_set_slice_with_step_keeps_bytes_intact(self):
+ # The buffer used for the read-patch-write cycle must not be the
+ # bytes object read from the blob: for a single byte it is an
+ # immortal singleton.
+ old_byte = self.data[5]
+ self.blob[5:6:2] = b"\xab"
+ self.assertEqual(bytes([old_byte])[0], old_byte)
+ self.assertEqual(self.blob[5:6], b"\xab")
+ expected = self.data[:5] + b"\xab" + self.data[6:]
+ actual = self.cx.execute("select b from test").fetchone()[0]
+ self.assertEqual(actual, expected)
+
def test_blob_set_empty_slice(self):
self.blob[0:0] = b""
self.assertEqual(self.blob[:], self.data)
diff --git a/Lib/test/test_sqlite3/test_dump.py b/Lib/test/test_sqlite3/test_dump.py
index e18a207e9..7841b610c 100644
--- a/Lib/test/test_sqlite3/test_dump.py
+++ b/Lib/test/test_sqlite3/test_dump.py
@@ -56,6 +56,24 @@ class DumpTests(MemoryDatabaseMixin, unittest.TestCase):
[self.assertEqual(expected_sqls[i], actual_sqls[i])
for i in range(len(expected_sqls))]
+ def test_dump_single_quote_in_identifier(self):
+ # A single quote in a table or column name must not break the dump.
+ self.cu.execute("""CREATE TABLE "a'b" ("c'd" text);""")
+ self.cu.execute("""INSERT INTO "a'b" VALUES('x''y');""")
+ expected = [
+ "BEGIN TRANSACTION;",
+ """CREATE TABLE "a'b" ("c'd" text);""",
+ """INSERT INTO "a'b" VALUES('x''y');""",
+ "COMMIT;",
+ ]
+ actual = list(self.cx.iterdump())
+ self.assertEqual(expected, actual)
+ # The dump restores into a fresh database.
+ with memory_database() as cx2:
+ cx2.executescript("".join(actual))
+ row = cx2.execute("""SELECT "c'd" FROM "a'b";""").fetchone()
+ self.assertEqual(row[0], "x'y")
+
def test_table_dump_filter(self):
all_table_sqls = [
"""CREATE TABLE "some_table_2" ("id_1" INTEGER);""",
diff --git a/Lib/test/test_sqlite3/test_factory.py b/Lib/test/test_sqlite3/test_factory.py
index 81170c653..79bb0b0f2 100644
--- a/Lib/test/test_sqlite3/test_factory.py
+++ b/Lib/test/test_sqlite3/test_factory.py
@@ -74,18 +74,9 @@ class ConnectionFactoryTests(unittest.TestCase):
def __init__(self, *args, **kwargs):
super(Factory, self).__init__(*args, **kwargs)
- regex = (
- r"Passing more than 1 positional argument to _sqlite3.Connection\(\) "
- r"is deprecated. Parameters 'timeout', 'detect_types', "
- r"'isolation_level', 'check_same_thread', 'factory', "
- r"'cached_statements' and 'uri' will become keyword-only "
- r"parameters in Python 3.15."
- )
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
- with memory_database(5.0, 0, None, True, Factory) as con:
- self.assertIsNone(con.isolation_level)
- self.assertIsInstance(con, Factory)
- self.assertEqual(cm.filename, __file__)
+ with self.assertRaisesRegex(TypeError,
+ r'connect\(\) takes at most 1 positional arguments'):
+ memory_database(5.0, 0, None, True, Factory)
class CursorFactoryTests(MemoryDatabaseMixin, unittest.TestCase):
@@ -251,6 +242,17 @@ class RowFactoryTests(MemoryDatabaseMixin, unittest.TestCase):
self.assertEqual(hash(row_1), hash(row_2))
+ def test_sqlite_row_hash_unhashable(self):
+ # An unhashable value must raise TypeError, not SystemError.
+ sqlite.register_converter("LST", lambda b: [1, 2, 3])
+ self.addCleanup(sqlite.converters.pop, "LST", None)
+ with memory_database(detect_types=sqlite.PARSE_DECLTYPES) as con:
+ con.row_factory = sqlite.Row
+ con.execute("create table t(x LST)")
+ con.execute("insert into t values(?)", (b"x",))
+ row = con.execute("select x from t").fetchone()
+ self.assertRaisesRegex(TypeError, "unhashable", hash, row)
+
def test_sqlite_row_as_sequence(self):
# Checks if the row object can act like a sequence.
row = self.con.execute("select 1 as a, 2 as b").fetchone()
diff --git a/Lib/test/test_sqlite3/test_hooks.py b/Lib/test/test_sqlite3/test_hooks.py
index e5b946bba..77b7498ab 100644
--- a/Lib/test/test_sqlite3/test_hooks.py
+++ b/Lib/test/test_sqlite3/test_hooks.py
@@ -290,16 +290,9 @@ class ProgressTests(MemoryDatabaseMixin, unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; keyword-only arguments not supported for set_progress_handler
def test_set_progress_handler_keyword_args(self):
- regex = (
- r"Passing keyword argument 'progress_handler' to "
- r"_sqlite3.Connection.set_progress_handler\(\) is deprecated. "
- r"Parameter 'progress_handler' will become positional-only in "
- r"Python 3.15."
- )
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
+ with self.assertRaisesRegex(TypeError,
+ 'takes at least 1 positional argument'):
self.con.set_progress_handler(progress_handler=lambda: None, n=1)
- self.assertEqual(cm.filename, __file__)
# When a handler has an invalid signature, the exception raised is
# the same that would be raised if the handler "negatively" replied.
@@ -464,16 +457,9 @@ class TraceCallbackTests(MemoryDatabaseMixin, unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; keyword-only arguments not supported for set_trace_callback
def test_set_trace_callback_keyword_args(self):
- regex = (
- r"Passing keyword argument 'trace_callback' to "
- r"_sqlite3.Connection.set_trace_callback\(\) is deprecated. "
- r"Parameter 'trace_callback' will become positional-only in "
- r"Python 3.15."
- )
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
+ with self.assertRaisesRegex(TypeError,
+ 'takes exactly 1 positional argument'):
self.con.set_trace_callback(trace_callback=lambda: None)
- self.assertEqual(cm.filename, __file__)
# When a handler has an invalid signature, the exception raised is
# the same that would be raised if the handler "negatively" replied,
diff --git a/Lib/test/test_sqlite3/test_userfunctions.py b/Lib/test/test_sqlite3/test_userfunctions.py
index d63bccf96..0ee1a7d2c 100644
--- a/Lib/test/test_sqlite3/test_userfunctions.py
+++ b/Lib/test/test_sqlite3/test_userfunctions.py
@@ -429,27 +429,9 @@ class FunctionTests(unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; deprecation warning not emitted for keyword args
def test_func_keyword_args(self):
- regex = (
- r"Passing keyword arguments 'name', 'narg' and 'func' to "
- r"_sqlite3.Connection.create_function\(\) is deprecated. "
- r"Parameters 'name', 'narg' and 'func' will become "
- r"positional-only in Python 3.15."
- )
-
- def noop():
- return None
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
- self.con.create_function("noop", 0, func=noop)
- self.assertEqual(cm.filename, __file__)
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
- self.con.create_function("noop", narg=0, func=noop)
- self.assertEqual(cm.filename, __file__)
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
- self.con.create_function(name="noop", narg=0, func=noop)
- self.assertEqual(cm.filename, __file__)
+ with self.assertRaisesRegex(TypeError,
+ 'takes exactly 3 positional arguments'):
+ self.con.create_function("noop", 0, func=lambda: None)
class WindowSumInt:
@@ -753,25 +735,9 @@ class AggregateTests(unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; keyword-only arguments not supported for create_aggregate
def test_agg_keyword_args(self):
- regex = (
- r"Passing keyword arguments 'name', 'n_arg' and 'aggregate_class' to "
- r"_sqlite3.Connection.create_aggregate\(\) is deprecated. "
- r"Parameters 'name', 'n_arg' and 'aggregate_class' will become "
- r"positional-only in Python 3.15."
- )
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
+ with self.assertRaisesRegex(TypeError,
+ 'takes exactly 3 positional arguments'):
self.con.create_aggregate("test", 1, aggregate_class=AggrText)
- self.assertEqual(cm.filename, __file__)
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
- self.con.create_aggregate("test", n_arg=1, aggregate_class=AggrText)
- self.assertEqual(cm.filename, __file__)
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
- self.con.create_aggregate(name="test", n_arg=0,
- aggregate_class=AggrText)
- self.assertEqual(cm.filename, __file__)
class AuthorizerTests(unittest.TestCase):
@@ -817,16 +783,9 @@ class AuthorizerTests(unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; keyword-only arguments not supported for set_authorizer
def test_authorizer_keyword_args(self):
- regex = (
- r"Passing keyword argument 'authorizer_callback' to "
- r"_sqlite3.Connection.set_authorizer\(\) is deprecated. "
- r"Parameter 'authorizer_callback' will become positional-only in "
- r"Python 3.15."
- )
-
- with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
+ with self.assertRaisesRegex(TypeError,
+ 'takes exactly 1 positional argument'):
self.con.set_authorizer(authorizer_callback=lambda: None)
- self.assertEqual(cm.filename, __file__)
class AuthorizerRaiseExceptionTests(AuthorizerTests):
diff --git a/Lib/trace.py b/Lib/trace.py
index cf8817f43..43ec201c4 100644
--- a/Lib/trace.py
+++ b/Lib/trace.py
@@ -604,7 +604,7 @@ class Trace:
def main():
import argparse
- parser = argparse.ArgumentParser(color=True)
+ parser = argparse.ArgumentParser()
parser.add_argument('--version', action='version', version='trace 2.0')
grp = parser.add_argument_group('Main options',
@@ -612,27 +612,27 @@ def main():
grp.add_argument('-c', '--count', action='store_true',
help='Count the number of times each line is executed and write '
- 'the counts to .cover for each module executed, in '
- 'the module\'s directory. See also --coverdir, --file, '
- '--no-report below.')
+ 'the counts to `.cover` for each module executed, in '
+ 'the module\'s directory. See also `--coverdir`, `--file`, '
+ '`--no-report` below.')
grp.add_argument('-t', '--trace', action='store_true',
- help='Print each line to sys.stdout before it is executed')
+ help='Print each line to `sys.stdout` before it is executed')
grp.add_argument('-l', '--listfuncs', action='store_true',
help='Keep track of which functions are executed at least once '
- 'and write the results to sys.stdout after the program exits. '
- 'Cannot be specified alongside --trace or --count.')
+ 'and write the results to `sys.stdout` after the program exits. '
+ 'Cannot be specified alongside `--trace` or `--count`.')
grp.add_argument('-T', '--trackcalls', action='store_true',
help='Keep track of caller/called pairs and write the results to '
- 'sys.stdout after the program exits.')
+ '`sys.stdout` after the program exits.')
grp = parser.add_argument_group('Modifiers')
_grp = grp.add_mutually_exclusive_group()
_grp.add_argument('-r', '--report', action='store_true',
help='Generate a report from a counts file; does not execute any '
- 'code. --file must specify the results file to read, which '
- 'must have been created in a previous run with --count '
- '--file=FILE')
+ 'code. `--file` must specify the results file to read, which '
+ 'must have been created in a previous run with `--count` '
+ '`--file=FILE`')
_grp.add_argument('-R', '--no-report', action='store_true',
help='Do not generate the coverage report files. '
'Useful if you want to accumulate over several runs.')
@@ -641,14 +641,14 @@ def main():
help='File to accumulate counts over several runs')
grp.add_argument('-C', '--coverdir',
help='Directory where the report files go. The coverage report '
- 'for . will be written to file '
- '//.cover')
+ 'for `.` will be written to file '
+ '`//.cover`')
grp.add_argument('-m', '--missing', action='store_true',
help='Annotate executable lines that were not executed with '
'">>>>>> "')
grp.add_argument('-s', '--summary', action='store_true',
- help='Write a brief summary for each file to sys.stdout. '
- 'Can only be used with --count or --report')
+ help='Write a brief summary for each file to `sys.stdout`. '
+ 'Can only be used with `--count` or `--report`')
grp.add_argument('-g', '--timing', action='store_true',
help='Prefix each line with the time since the program started. '
'Only used while tracing')
@@ -661,7 +661,7 @@ def main():
'module names.')
grp.add_argument('--ignore-dir', action='append', default=[],
help='Ignore files in the given directory '
- '(multiple directories can be joined by os.pathsep).')
+ '(multiple directories can be joined by `os.pathsep`).')
parser.add_argument('--module', action='store_true', default=False,
help='Trace a module. ')
@@ -721,7 +721,6 @@ def main():
'__package__': mod_spec.parent,
'__loader__': mod_spec.loader,
'__spec__': mod_spec,
- '__cached__': None,
}
else:
sys.argv = [opts.progname, *opts.arguments]
@@ -734,7 +733,6 @@ def main():
'__file__': opts.progname,
'__name__': '__main__',
'__package__': None,
- '__cached__': None,
}
t.runctx(code, globs, globs)
except OSError as err: