Skip to content
Prev Previous commit
Add managed resource helper
...and sort imports
  • Loading branch information
Erlend E. Aasland
Erlend E. Aasland committed Jun 2, 2021
commit aba8dd724618df1a3ab2cad90b03436cb9bc9fd8
44 changes: 24 additions & 20 deletions Lib/sqlite3/test/dbapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,26 @@
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.

import threading
import unittest
import contextlib
import sqlite3 as sqlite
import sys
import threading
import unittest

from test.support.os_helper import TESTFN, unlink


# Helper for tests using TESTFN
@contextlib.contextmanager
def managed_connect(*args, **kwargs):
cx = sqlite.connect(*args, **kwargs)
try:
yield cx
finally:
cx.close()
unlink(TESTFN)


class ModuleTests(unittest.TestCase):
def test_api_level(self):
self.assertEqual(sqlite.apilevel, "2.0",
Expand Down Expand Up @@ -189,10 +201,8 @@ def test_in_transaction_ro(self):
with self.assertRaises(AttributeError):
self.cx.in_transaction = True


class OpenTests(unittest.TestCase):
def tearDown(self):
unlink(TESTFN)
_sql = "create table test(id integer)"

def test_open_with_path_like_object(self):
""" Checks that we can successfully connect to a database using an object that
Expand All @@ -201,23 +211,17 @@ class Path:
def __fspath__(self):
return TESTFN
path = Path()
with sqlite.connect(path) as cx:
cx.execute('create table test(id integer)')
cx.close()
with managed_connect(path) as cx:
cx.execute(self._sql)

def test_open_uri(self):
with sqlite.connect(TESTFN) as cx:
cx.execute('create table test(id integer)')
cx.close()

with sqlite.connect('file:' + TESTFN, uri=True) as cx:
cx.execute('insert into test(id) values(0)')
cx.close()

with sqlite.connect('file:' + TESTFN + '?mode=ro', uri=True) as cx:
with self.assertRaises(sqlite.OperationalError):
cx.execute('insert into test(id) values(1)')
cx.close()
with managed_connect(TESTFN) as cx:
cx.execute(self._sql)
with managed_connect(f"file:{TESTFN}", uri=True) as cx:
cx.execute(self._sql)
with self.assertRaises(sqlite.OperationalError):
with managed_connect(f"file:{TESTFN}?mode=ro", uri=True) as cx:
cx.execute(self._sql)


class CursorTests(unittest.TestCase):
Expand Down