1 """sqlite3 CLI tests."""
2 import sqlite3
3 import unittest
4
5 from sqlite3.__main__ import main as cli
6 from test.support.os_helper import TESTFN, unlink
7 from test.support import captured_stdout, captured_stderr, captured_stdin
8
9
10 class ESC[4;38;5;81mCommandLineInterface(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
11
12 def _do_test(self, *args, expect_success=True):
13 with (
14 captured_stdout() as out,
15 captured_stderr() as err,
16 self.assertRaises(SystemExit) as cm
17 ):
18 cli(args)
19 return out.getvalue(), err.getvalue(), cm.exception.code
20
21 def expect_success(self, *args):
22 out, err, code = self._do_test(*args)
23 self.assertEqual(code, 0,
24 "\n".join([f"Unexpected failure: {args=}", out, err]))
25 self.assertEqual(err, "")
26 return out
27
28 def expect_failure(self, *args):
29 out, err, code = self._do_test(*args, expect_success=False)
30 self.assertNotEqual(code, 0,
31 "\n".join([f"Unexpected failure: {args=}", out, err]))
32 self.assertEqual(out, "")
33 return err
34
35 def test_cli_help(self):
36 out = self.expect_success("-h")
37 self.assertIn("usage: python -m sqlite3", out)
38
39 def test_cli_version(self):
40 out = self.expect_success("-v")
41 self.assertIn(sqlite3.sqlite_version, out)
42
43 def test_cli_execute_sql(self):
44 out = self.expect_success(":memory:", "select 1")
45 self.assertIn("(1,)", out)
46
47 def test_cli_execute_too_much_sql(self):
48 stderr = self.expect_failure(":memory:", "select 1; select 2")
49 err = "ProgrammingError: You can only execute one statement at a time"
50 self.assertIn(err, stderr)
51
52 def test_cli_execute_incomplete_sql(self):
53 stderr = self.expect_failure(":memory:", "sel")
54 self.assertIn("OperationalError (SQLITE_ERROR)", stderr)
55
56 def test_cli_on_disk_db(self):
57 self.addCleanup(unlink, TESTFN)
58 out = self.expect_success(TESTFN, "create table t(t)")
59 self.assertEqual(out, "")
60 out = self.expect_success(TESTFN, "select count(t) from t")
61 self.assertIn("(0,)", out)
62
63
64 class ESC[4;38;5;81mInteractiveSession(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
65 MEMORY_DB_MSG = "Connected to a transient in-memory database"
66 PS1 = "sqlite> "
67 PS2 = "... "
68
69 def run_cli(self, *args, commands=()):
70 with (
71 captured_stdin() as stdin,
72 captured_stdout() as stdout,
73 captured_stderr() as stderr,
74 self.assertRaises(SystemExit) as cm
75 ):
76 for cmd in commands:
77 stdin.write(cmd + "\n")
78 stdin.seek(0)
79 cli(args)
80
81 out = stdout.getvalue()
82 err = stderr.getvalue()
83 self.assertEqual(cm.exception.code, 0,
84 f"Unexpected failure: {args=}\n{out}\n{err}")
85 return out, err
86
87 def test_interact(self):
88 out, err = self.run_cli()
89 self.assertIn(self.MEMORY_DB_MSG, err)
90 self.assertIn(self.MEMORY_DB_MSG, err)
91 self.assertTrue(out.endswith(self.PS1))
92 self.assertEqual(out.count(self.PS1), 1)
93 self.assertEqual(out.count(self.PS2), 0)
94
95 def test_interact_quit(self):
96 out, err = self.run_cli(commands=(".quit",))
97 self.assertIn(self.MEMORY_DB_MSG, err)
98 self.assertTrue(out.endswith(self.PS1))
99 self.assertEqual(out.count(self.PS1), 1)
100 self.assertEqual(out.count(self.PS2), 0)
101
102 def test_interact_version(self):
103 out, err = self.run_cli(commands=(".version",))
104 self.assertIn(self.MEMORY_DB_MSG, err)
105 self.assertIn(sqlite3.sqlite_version + "\n", out)
106 self.assertTrue(out.endswith(self.PS1))
107 self.assertEqual(out.count(self.PS1), 2)
108 self.assertEqual(out.count(self.PS2), 0)
109 self.assertIn(sqlite3.sqlite_version, out)
110
111 def test_interact_valid_sql(self):
112 out, err = self.run_cli(commands=("SELECT 1;",))
113 self.assertIn(self.MEMORY_DB_MSG, err)
114 self.assertIn("(1,)\n", out)
115 self.assertTrue(out.endswith(self.PS1))
116 self.assertEqual(out.count(self.PS1), 2)
117 self.assertEqual(out.count(self.PS2), 0)
118
119 def test_interact_incomplete_multiline_sql(self):
120 out, err = self.run_cli(commands=("SELECT 1",))
121 self.assertIn(self.MEMORY_DB_MSG, err)
122 self.assertTrue(out.endswith(self.PS2))
123 self.assertEqual(out.count(self.PS1), 1)
124 self.assertEqual(out.count(self.PS2), 1)
125
126 def test_interact_valid_multiline_sql(self):
127 out, err = self.run_cli(commands=("SELECT 1\n;",))
128 self.assertIn(self.MEMORY_DB_MSG, err)
129 self.assertIn(self.PS2, out)
130 self.assertIn("(1,)\n", out)
131 self.assertTrue(out.endswith(self.PS1))
132 self.assertEqual(out.count(self.PS1), 2)
133 self.assertEqual(out.count(self.PS2), 1)
134
135 def test_interact_invalid_sql(self):
136 out, err = self.run_cli(commands=("sel;",))
137 self.assertIn(self.MEMORY_DB_MSG, err)
138 self.assertIn("OperationalError (SQLITE_ERROR)", err)
139 self.assertTrue(out.endswith(self.PS1))
140 self.assertEqual(out.count(self.PS1), 2)
141 self.assertEqual(out.count(self.PS2), 0)
142
143 def test_interact_on_disk_file(self):
144 self.addCleanup(unlink, TESTFN)
145
146 out, err = self.run_cli(TESTFN, commands=("CREATE TABLE t(t);",))
147 self.assertIn(TESTFN, err)
148 self.assertTrue(out.endswith(self.PS1))
149
150 out, _ = self.run_cli(TESTFN, commands=("SELECT count(t) FROM t;",))
151 self.assertIn("(0,)\n", out)
152
153
154 if __name__ == "__main__":
155 unittest.main()