Commits

Brian Kearns  committed 3b60aa1

simplify sqlite tests by providing a funcarg that automatically connects/closes

  • Participants
  • Parent commits 53926a2

Comments (0)

Files changed (1)

File pypy/module/test_lib_pypy/test_sqlite3.py

 import pytest
 from lib_pypy import _sqlite3
 
-def test_list_ddl():
+def pytest_funcarg__con(request):
+    con = _sqlite3.connect(':memory:')
+    request.addfinalizer(lambda: con.close())
+    return con
+
+def test_list_ddl(con):
     """From issue996.  Mostly just looking for lack of exceptions."""
-    connection = _sqlite3.connect(':memory:')
-    cursor = connection.cursor()
+    cursor = con.cursor()
     cursor.execute('CREATE TABLE foo (bar INTEGER)')
     result = list(cursor)
     assert result == []
     result = list(cursor)
     assert result == [(42,)]
 
-def test_total_changes_after_close():
-    con = _sqlite3.connect(':memory:')
+def test_total_changes_after_close(con):
     con.close()
     pytest.raises(_sqlite3.ProgrammingError, "con.total_changes")
 
     e = pytest.raises(_sqlite3.ProgrammingError, "con.cursor()")
     assert '__init__' in e.value.message
 
-def test_cursor_check_init():
+def test_cursor_check_init(con):
     class Cursor(_sqlite3.Cursor):
         def __init__(self, name):
             pass
 
-    con = _sqlite3.connect(":memory:")
     cur = Cursor(con)
     e = pytest.raises(_sqlite3.ProgrammingError, "cur.execute('select 1')")
     assert '__init__' in e.value.message
 
-def test_connection_after_close():
-    con = _sqlite3.connect(':memory:')
+def test_connection_after_close(con):
     pytest.raises(TypeError, "con()")
     con.close()
     # raises ProgrammingError because should check closed before check args
     pytest.raises(_sqlite3.ProgrammingError, "con()")
 
-def test_cursor_iter():
-    con = _sqlite3.connect(':memory:')
+def test_cursor_iter(con):
     cur = con.cursor()
     with pytest.raises(StopIteration):
         next(cur)
     with pytest.raises(StopIteration):
         next(cur)
 
-def test_cursor_after_close():
-     con = _sqlite3.connect(':memory:')
+def test_cursor_after_close(con):
      cur = con.execute('select 1')
      cur.close()
      con.close()
     finally:
         resource.setrlimit(resource.RLIMIT_NOFILE, limit)
 
-def test_on_conflict_rollback_executemany():
+def test_on_conflict_rollback_executemany(con):
     major, minor, micro = _sqlite3.sqlite_version.split('.')
     if (int(major), int(minor), int(micro)) < (3, 2, 2):
         pytest.skip("requires sqlite3 version >= 3.2.2")
-    con = _sqlite3.connect(":memory:")
     con.execute("create table foo(x, unique(x) on conflict rollback)")
     con.execute("insert into foo(x) values (1)")
     try:
         con.commit()
     except _sqlite3.OperationalError:
         pytest.fail("_sqlite3 knew nothing about the implicit ROLLBACK")
-    con.close()
 
-def test_statement_arg_checking():
-    con = _sqlite3.connect(':memory:')
+def test_statement_arg_checking(con):
     with pytest.raises(_sqlite3.Warning) as e:
         con(123)
     assert str(e.value) == 'SQL is of wrong type. Must be string or unicode.'
         con.executescript(123)
     assert str(e.value) == 'script argument must be unicode or string.'
 
-def test_statement_param_checking():
-    con = _sqlite3.connect(':memory:')
+def test_statement_param_checking(con):
     con.execute('create table foo(x)')
     con.execute('insert into foo(x) values (?)', [2])
     con.execute('insert into foo(x) values (?)', (2,))
     with pytest.raises(ValueError) as e:
         con.execute('insert into foo(x) values (?)', 2)
     assert str(e.value) == 'parameters are of unsupported type'
-    con.close()
 
-def test_explicit_begin():
-    con = _sqlite3.connect(':memory:')
+def test_explicit_begin(con):
     con.execute('BEGIN')
     con.execute('BEGIN ')
     con.execute('BEGIN')
     con.execute('BEGIN')
     con.commit()
 
-def test_row_factory_use():
-    con = _sqlite3.connect(':memory:')
+def test_row_factory_use(con):
     con.row_factory = 42
     con.execute('select 1')
 
-def test_returning_blob_must_own_memory():
+def test_returning_blob_must_own_memory(con):
     import gc
-    con = _sqlite3.connect(":memory:")
     con.create_function("returnblob", 0, lambda: buffer("blob"))
-    cur = con.cursor()
-    cur.execute("select returnblob()")
+    cur = con.execute("select returnblob()")
     val = cur.fetchone()[0]
     for i in range(5):
         gc.collect()
         got = (val[0], val[1], val[2], val[3])
         assert got == ('b', 'l', 'o', 'b')
 
-def test_description_after_fetchall():
-    con = _sqlite3.connect(":memory:")
+def test_description_after_fetchall(con):
     cur = con.cursor()
     cur.execute("select 42").fetchall()
     assert cur.description is not None
 
-def test_executemany_lastrowid():
-    con = _sqlite3.connect(':memory:')
+def test_executemany_lastrowid(con):
     cur = con.cursor()
     cur.execute("create table test(a)")
     cur.executemany("insert into test values (?)", [[1], [2], [3]])