summaryrefslogtreecommitdiffstats
path: root/Lib
diff options
context:
space:
mode:
Diffstat (limited to 'Lib')
-rw-r--r--Lib/sqlite3/__main__.py9
-rw-r--r--Lib/test/test_sqlite3/test_cli.py148
-rw-r--r--Lib/test/test_sqlite3/test_dbapi.py2
-rw-r--r--Lib/test/test_sqlite3/test_transactions.py16
4 files changed, 74 insertions, 101 deletions
diff --git a/Lib/sqlite3/__main__.py b/Lib/sqlite3/__main__.py
index 3228dbc..10a2e9e 100644
--- a/Lib/sqlite3/__main__.py
+++ b/Lib/sqlite3/__main__.py
@@ -62,7 +62,7 @@ class SqliteInteractiveConsole(InteractiveConsole):
return False
-def main():
+def main(*args):
parser = ArgumentParser(
description="Python sqlite3 CLI",
prog="python -m sqlite3",
@@ -86,7 +86,7 @@ def main():
version=f"SQLite version {sqlite3.sqlite_version}",
help="Print underlying SQLite library version",
)
- args = parser.parse_args()
+ args = parser.parse_args(*args)
if args.filename == ":memory:":
db_name = "a transient in-memory database"
@@ -120,5 +120,8 @@ def main():
finally:
con.close()
+ sys.exit(0)
-main()
+
+if __name__ == "__main__":
+ main(sys.argv)
diff --git a/Lib/test/test_sqlite3/test_cli.py b/Lib/test/test_sqlite3/test_cli.py
index d374f8e..e681f5c 100644
--- a/Lib/test/test_sqlite3/test_cli.py
+++ b/Lib/test/test_sqlite3/test_cli.py
@@ -1,42 +1,35 @@
"""sqlite3 CLI tests."""
-
-import sqlite3 as sqlite
-import subprocess
-import sys
+import sqlite3
import unittest
-from test.support import SHORT_TIMEOUT, requires_subprocess
+from sqlite3.__main__ import main as cli
from test.support.os_helper import TESTFN, unlink
+from test.support import captured_stdout, captured_stderr, captured_stdin
-@requires_subprocess()
class CommandLineInterface(unittest.TestCase):
def _do_test(self, *args, expect_success=True):
- with subprocess.Popen(
- [sys.executable, "-Xutf8", "-m", "sqlite3", *args],
- encoding="utf-8",
- bufsize=0,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- ) as proc:
- proc.wait()
- if expect_success == bool(proc.returncode):
- self.fail("".join(proc.stderr))
- stdout = proc.stdout.read()
- stderr = proc.stderr.read()
- if expect_success:
- self.assertEqual(stderr, "")
- else:
- self.assertEqual(stdout, "")
- return stdout, stderr
+ with (
+ captured_stdout() as out,
+ captured_stderr() as err,
+ self.assertRaises(SystemExit) as cm
+ ):
+ cli(args)
+ return out.getvalue(), err.getvalue(), cm.exception.code
def expect_success(self, *args):
- out, _ = self._do_test(*args)
+ out, err, code = self._do_test(*args)
+ self.assertEqual(code, 0,
+ "\n".join([f"Unexpected failure: {args=}", out, err]))
+ self.assertEqual(err, "")
return out
def expect_failure(self, *args):
- _, err = self._do_test(*args, expect_success=False)
+ out, err, code = self._do_test(*args, expect_success=False)
+ self.assertNotEqual(code, 0,
+ "\n".join([f"Unexpected failure: {args=}", out, err]))
+ self.assertEqual(out, "")
return err
def test_cli_help(self):
@@ -45,7 +38,7 @@ class CommandLineInterface(unittest.TestCase):
def test_cli_version(self):
out = self.expect_success("-v")
- self.assertIn(sqlite.sqlite_version, out)
+ self.assertIn(sqlite3.sqlite_version, out)
def test_cli_execute_sql(self):
out = self.expect_success(":memory:", "select 1")
@@ -68,87 +61,68 @@ class CommandLineInterface(unittest.TestCase):
self.assertIn("(0,)", out)
-@requires_subprocess()
class InteractiveSession(unittest.TestCase):
- TIMEOUT = SHORT_TIMEOUT / 10.
MEMORY_DB_MSG = "Connected to a transient in-memory database"
PS1 = "sqlite> "
PS2 = "... "
- def start_cli(self, *args):
- return subprocess.Popen(
- [sys.executable, "-Xutf8", "-m", "sqlite3", *args],
- encoding="utf-8",
- bufsize=0,
- stdin=subprocess.PIPE,
- # Note: the banner is printed to stderr, the prompt to stdout.
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- )
-
- def expect_success(self, proc):
- proc.wait()
- if proc.returncode:
- self.fail("".join(proc.stderr))
+ def run_cli(self, *args, commands=()):
+ with (
+ captured_stdin() as stdin,
+ captured_stdout() as stdout,
+ captured_stderr() as stderr,
+ self.assertRaises(SystemExit) as cm
+ ):
+ for cmd in commands:
+ stdin.write(cmd + "\n")
+ stdin.seek(0)
+ cli(args)
+
+ out = stdout.getvalue()
+ err = stderr.getvalue()
+ self.assertEqual(cm.exception.code, 0,
+ f"Unexpected failure: {args=}\n{out}\n{err}")
+ return out, err
def test_interact(self):
- with self.start_cli() as proc:
- out, err = proc.communicate(timeout=self.TIMEOUT)
- self.assertIn(self.MEMORY_DB_MSG, err)
- self.assertIn(self.PS1, out)
- self.expect_success(proc)
+ out, err = self.run_cli()
+ self.assertIn(self.MEMORY_DB_MSG, err)
+ self.assertIn(self.PS1, out)
def test_interact_quit(self):
- with self.start_cli() as proc:
- out, err = proc.communicate(input=".quit", timeout=self.TIMEOUT)
- self.assertIn(self.MEMORY_DB_MSG, err)
- self.assertIn(self.PS1, out)
- self.expect_success(proc)
+ out, err = self.run_cli(commands=(".quit",))
+ self.assertIn(self.PS1, out)
def test_interact_version(self):
- with self.start_cli() as proc:
- out, err = proc.communicate(input=".version", timeout=self.TIMEOUT)
- self.assertIn(self.MEMORY_DB_MSG, err)
- self.assertIn(sqlite.sqlite_version, out)
- self.expect_success(proc)
+ out, err = self.run_cli(commands=(".version",))
+ self.assertIn(self.MEMORY_DB_MSG, err)
+ self.assertIn(sqlite3.sqlite_version, out)
def test_interact_valid_sql(self):
- with self.start_cli() as proc:
- out, err = proc.communicate(input="select 1;",
- timeout=self.TIMEOUT)
- self.assertIn(self.MEMORY_DB_MSG, err)
- self.assertIn("(1,)", out)
- self.expect_success(proc)
+ out, err = self.run_cli(commands=("SELECT 1;",))
+ self.assertIn(self.MEMORY_DB_MSG, err)
+ self.assertIn("(1,)", out)
def test_interact_valid_multiline_sql(self):
- with self.start_cli() as proc:
- out, err = proc.communicate(input="select 1\n;",
- timeout=self.TIMEOUT)
- self.assertIn(self.MEMORY_DB_MSG, err)
- self.assertIn(self.PS2, out)
- self.assertIn("(1,)", out)
- self.expect_success(proc)
+ out, err = self.run_cli(commands=("SELECT 1\n;",))
+ self.assertIn(self.MEMORY_DB_MSG, err)
+ self.assertIn(self.PS2, out)
+ self.assertIn("(1,)", out)
def test_interact_invalid_sql(self):
- with self.start_cli() as proc:
- out, err = proc.communicate(input="sel;", timeout=self.TIMEOUT)
- self.assertIn(self.MEMORY_DB_MSG, err)
- self.assertIn("OperationalError (SQLITE_ERROR)", err)
- self.expect_success(proc)
+ out, err = self.run_cli(commands=("sel;",))
+ self.assertIn(self.MEMORY_DB_MSG, err)
+ self.assertIn("OperationalError (SQLITE_ERROR)", err)
def test_interact_on_disk_file(self):
self.addCleanup(unlink, TESTFN)
- with self.start_cli(TESTFN) as proc:
- out, err = proc.communicate(input="create table t(t);",
- timeout=self.TIMEOUT)
- self.assertIn(TESTFN, err)
- self.assertIn(self.PS1, out)
- self.expect_success(proc)
- with self.start_cli(TESTFN, "select count(t) from t") as proc:
- out = proc.stdout.read()
- err = proc.stderr.read()
- self.assertIn("(0,)", out)
- self.expect_success(proc)
+
+ out, err = self.run_cli(TESTFN, commands=("CREATE TABLE t(t);",))
+ self.assertIn(TESTFN, err)
+ self.assertIn(self.PS1, out)
+
+ out, _ = self.run_cli(TESTFN, commands=("SELECT count(t) FROM t;",))
+ self.assertIn("(0,)", out)
if __name__ == "__main__":
diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index 9dedbdb..f3efe0f 100644
--- a/Lib/test/test_sqlite3/test_dbapi.py
+++ b/Lib/test/test_sqlite3/test_dbapi.py
@@ -1854,7 +1854,7 @@ class SqliteOnConflictTests(unittest.TestCase):
@requires_subprocess()
class MultiprocessTests(unittest.TestCase):
- CONNECTION_TIMEOUT = SHORT_TIMEOUT / 1000. # Defaults to 30 ms
+ CONNECTION_TIMEOUT = 0 # Disable the busy timeout.
def tearDown(self):
unlink(TESTFN)
diff --git a/Lib/test/test_sqlite3/test_transactions.py b/Lib/test/test_sqlite3/test_transactions.py
index b7b231d..a3de7a7 100644
--- a/Lib/test/test_sqlite3/test_transactions.py
+++ b/Lib/test/test_sqlite3/test_transactions.py
@@ -24,7 +24,6 @@ import unittest
import sqlite3 as sqlite
from contextlib import contextmanager
-from test.support import LOOPBACK_TIMEOUT
from test.support.os_helper import TESTFN, unlink
from test.support.script_helper import assert_python_ok
@@ -32,15 +31,14 @@ from .util import memory_database
from .util import MemoryDatabaseMixin
-TIMEOUT = LOOPBACK_TIMEOUT / 10
-
-
class TransactionTests(unittest.TestCase):
def setUp(self):
- self.con1 = sqlite.connect(TESTFN, timeout=TIMEOUT)
+ # We can disable the busy handlers, since we control
+ # the order of SQLite C API operations.
+ self.con1 = sqlite.connect(TESTFN, timeout=0)
self.cur1 = self.con1.cursor()
- self.con2 = sqlite.connect(TESTFN, timeout=TIMEOUT)
+ self.con2 = sqlite.connect(TESTFN, timeout=0)
self.cur2 = self.con2.cursor()
def tearDown(self):
@@ -120,10 +118,8 @@ class TransactionTests(unittest.TestCase):
self.cur2.execute("insert into test(i) values (5)")
def test_locking(self):
- """
- This tests the improved concurrency with pysqlite 2.3.4. You needed
- to roll back con2 before you could commit con1.
- """
+ # This tests the improved concurrency with pysqlite 2.3.4. You needed
+ # to roll back con2 before you could commit con1.
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
with self.assertRaises(sqlite.OperationalError):
generated by cgit v0.12 at 2025-02-01 16:02:08 (GMT)