Skip to content
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Address review: Apply Sergey's suggestions
  • Loading branch information
Erlend E. Aasland
Erlend E. Aasland committed Oct 19, 2021
commit 51ab5518d6f4b55a782fd910d2382a9fa32b626a
38 changes: 23 additions & 15 deletions Lib/sqlite3/test/test_regression.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,10 @@
import sqlite3 as sqlite
import weakref
import functools
from test import support

from .test_dbapi import managed_connect
from test import support
from unittest.mock import patch

class RegressionTests(unittest.TestCase):
def setUp(self):
Expand Down Expand Up @@ -489,27 +490,34 @@ def test_executescript_step_through_select(self):

class ConverterProgrammingErrorTestCase(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES)
self.con = sqlite.connect(":memory:",
detect_types=sqlite.PARSE_COLNAMES)
self.cur = self.con.cursor()
self.cur.execute("create table test(x foo)")
self.cur.executemany("insert into test(x) values (?)", [("foo",), ("bar",)])

sqlite.converters["CURSOR_INIT"] = lambda x: self.cur.__init__(self.con)
sqlite.converters["CURSOR_CLOSE"] = lambda x: self.cur.close()
sqlite.converters["CURSOR_ITER"] = lambda x, l=[]: self.cur.fetchone() if l else l.append(None)
self.cur.executemany("insert into test(x) values (?)",
[("foo",), ("bar",)])

def tearDown(self):
del sqlite.converters["CURSOR_INIT"]
del sqlite.converters["CURSOR_CLOSE"]
del sqlite.converters["CURSOR_ITER"]
self.cur.close()
self.con.close()

def test_recursive_cursor_usage(self):
for converter in "CURSOR_INIT", "CURSOR_CLOSE", "CURSOR_ITER":
with self.subTest(converter=converter):
self.cur.execute(f'select x as "x [{converter}]", x from test')
self.assertRaises(sqlite.ProgrammingError, self.cur.fetchall)
def test_recursive_cursor_init(self):
conv = lambda x: self.cur.__init__(self.con)
with patch.dict(sqlite.converters, {"INIT": conv}):
self.cur.execute(f'select x as "x [INIT]", x from test')
self.assertRaises(sqlite.ProgrammingError, self.cur.fetchall)

def test_recursive_cursor_close(self):
conv = lambda x: self.cur.close()
with patch.dict(sqlite.converters, {"CLOSE": conv}):
self.cur.execute(f'select x as "x [CLOSE]", x from test')
self.assertRaises(sqlite.ProgrammingError, self.cur.fetchall)

def test_recursive_cursor_iter(self):
conv = lambda x, l=[]: self.cur.fetchone() if l else l.append(None)
with patch.dict(sqlite.converters, {"ITER": conv}):
self.cur.execute(f'select x as "x [ITER]", x from test')
self.assertRaises(sqlite.ProgrammingError, self.cur.fetchall)


if __name__ == "__main__":
Expand Down