From bb9cc1216873604459724860d606283c398ea06b Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Thu, 2 Feb 2017 20:25:20 +0100 Subject: [feat] add support for the blob interface Pysqlcipher support for the sqlite blob interface: https://sqlite.org/c3ref/blob_open.html Copying the code from the PR in pysqlite: https://github.com/ghaering/pysqlite/pull/93 --- lib/test/dbapi.py | 264 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 263 insertions(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/test/dbapi.py b/lib/test/dbapi.py index 05dfd4b..71e5e72 100644 --- a/lib/test/dbapi.py +++ b/lib/test/dbapi.py @@ -465,6 +465,155 @@ class CursorTests(unittest.TestCase): except TypeError: pass + +class BlobTests(unittest.TestCase): + def setUp(self): + self.cx = sqlite.connect(":memory:") + self.cx.execute("create table test(id integer primary key, blob_col blob)") + self.blob_data = "a" * 100 + self.cx.execute("insert into test(blob_col) values (?)", (self.blob_data, )) + self.blob = self.cx.blob("test", "blob_col", 1, 1) + self.second_data = "b" * 100 + + def tearDown(self): + self.blob.close() + self.cx.close() + + def CheckLength(self): + self.assertEqual(self.blob.length(), 100) + + def CheckTell(self): + self.assertEqual(self.blob.tell(), 0) + + def CheckSeekFromBlobStart(self): + self.blob.seek(10) + self.assertEqual(self.blob.tell(), 10) + self.blob.seek(10, 0) + self.assertEqual(self.blob.tell(), 10) + + def CheckSeekFromCurrentPosition(self): + self.blob.seek(10,1) + self.blob.seek(10,1) + self.assertEqual(self.blob.tell(), 20) + + def CheckSeekFromBlobEnd(self): + self.blob.seek(-10,2) + self.assertEqual(self.blob.tell(), 90) + + def CheckBlobSeekOverBlobSize(self): + try: + self.blob.seek(1000) + self.fail("should have raised a ValueError") + except ValueError: + pass + except Exception: + self.fail("should have raised a ValueError") + + def CheckBlobSeekUnderBlobSize(self): + try: + self.blob.seek(-10) + self.fail("should have raised a ValueError") + except ValueError: + pass + except Exception: + self.fail("should have raised a ValueError") + + def CheckBlobRead(self): + self.assertEqual(self.blob.read(), self.blob_data) + + def CheckBlobReadSize(self): + self.assertEqual(len(self.blob.read(10)), 10) + + def CheckBlobReadAdvanceOffset(self): + self.blob.read(10) + self.assertEqual(self.blob.tell(), 10) + + def CheckBlobReadStartAtOffset(self): + self.blob.seek(10) + self.blob.write(self.second_data[:10]) + self.blob.seek(10) + self.assertEqual(self.blob.read(10), self.second_data[:10]) + + def CheckBlobWrite(self): + self.blob.write(self.second_data) + self.assertEqual(str(self.cx.execute("select blob_col from test").fetchone()[0]), self.second_data) + + def CheckBlobWriteAtOffset(self): + self.blob.seek(50) + self.blob.write(self.second_data[:50]) + self.assertEqual(str(self.cx.execute("select blob_col from test").fetchone()[0]), + self.blob_data[:50] + self.second_data[:50]) + + def CheckBlobWriteAdvanceOffset(self): + self.blob.write(self.second_data[:50]) + self.assertEqual(self.blob.tell(), 50) + + def CheckBlobWriteMoreThenBlobSize(self): + try: + self.blob.write("a" * 1000) + self.fail("should have raised a sqlite.OperationalError") + except sqlite.OperationalError: + pass + except Exception: + self.fail("should have raised a sqlite.OperationalError") + + def CheckBlobReadAfterRowChange(self): + self.cx.execute("UPDATE test SET blob_col='aaaa' where id=1") + try: + self.blob.read() + self.fail("should have raised a sqlite.OperationalError") + except sqlite.OperationalError: + pass + except Exception: + self.fail("should have raised a sqlite.OperationalError") + + def CheckBlobWriteAfterRowChange(self): + self.cx.execute("UPDATE test SET blob_col='aaaa' where id=1") + try: + self.blob.write("aaa") + self.fail("should have raised a sqlite.OperationalError") + except sqlite.OperationalError: + pass + except Exception: + self.fail("should have raised a sqlite.OperationalError") + + def CheckBlobOpenWithBadDb(self): + try: + self.cx.blob("test", "blob_col", 1, 1, dbname="notexisintg") + self.fail("should have raised a sqlite.OperationalError") + except sqlite.OperationalError: + pass + except Exception: + self.fail("should have raised a sqlite.OperationalError") + + def CheckBlobOpenWithBadTable(self): + try: + self.cx.blob("notexisintg", "blob_col", 1, 1) + self.fail("should have raised a sqlite.OperationalError") + except sqlite.OperationalError: + pass + except Exception: + self.fail("should have raised a sqlite.OperationalError") + + def CheckBlobOpenWithBadColumn(self): + try: + self.cx.blob("test", "notexisting", 1, 1) + self.fail("should have raised a sqlite.OperationalError") + except sqlite.OperationalError: + pass + except Exception: + self.fail("should have raised a sqlite.OperationalError") + + def CheckBlobOpenWithBadRow(self): + try: + self.cx.blob("test", "blob_col", 2, 1) + self.fail("should have raised a sqlite.OperationalError") + except sqlite.OperationalError: + pass + except Exception: + self.fail("should have raised a sqlite.OperationalError") + + class ThreadTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") @@ -763,6 +912,20 @@ class ClosedConTests(unittest.TestCase): except: self.fail("Should have raised a ProgrammingError") + def CheckClosedBlobRead(self): + con = sqlite.connect(":memory:") + con.execute("create table test(id integer primary key, blob_col blob)") + con.execute("insert into test(blob_col) values (zeroblob(100))") + blob = con.blob("test", "blob_col", 1) + con.close() + try: + blob.read() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") + def CheckClosedCreateFunction(self): con = sqlite.connect(":memory:") con.close() @@ -859,16 +1022,115 @@ class ClosedCurTests(unittest.TestCase): except: self.fail("Should have raised a ProgrammingError: " + method_name) + +class ClosedBlobTests(unittest.TestCase): + def setUp(self): + self.cx = sqlite.connect(":memory:") + self.cx.execute("create table test(id integer primary key, blob_col blob)") + self.cx.execute("insert into test(blob_col) values (zeroblob(100))") + + def tearDown(self): + self.cx.close() + + def CheckClosedRead(self): + self.blob = self.cx.blob("test", "blob_col", 1) + self.blob.close() + try: + self.blob.read() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except Exception: + self.fail("Should have raised a ProgrammingError") + + def CheckClosedWrite(self): + self.blob = self.cx.blob("test", "blob_col", 1) + self.blob.close() + try: + self.blob.write("aaaaaaaaa") + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except Exception: + self.fail("Should have raised a ProgrammingError") + + def CheckClosedSeek(self): + self.blob = self.cx.blob("test", "blob_col", 1) + self.blob.close() + try: + self.blob.seek(10) + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except Exception: + self.fail("Should have raised a ProgrammingError") + + def CheckClosedTell(self): + self.blob = self.cx.blob("test", "blob_col", 1) + self.blob.close() + try: + self.blob.tell() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except Exception: + self.fail("Should have raised a ProgrammingError") + + def CheckClosedClose(self): + self.blob = self.cx.blob("test", "blob_col", 1) + self.blob.close() + try: + self.blob.close() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except Exception: + self.fail("Should have raised a ProgrammingError") + + +class BlobContextManagerTests(unittest.TestCase): + def setUp(self): + self.cx = sqlite.connect(":memory:") + self.cx.execute("create table test(id integer primary key, blob_col blob)") + self.cx.execute("insert into test(blob_col) values (zeroblob(100))") + + def tearDown(self): + self.cx.close() + + def CheckContextExecute(self): + data = "a" * 100 + with self.cx.blob("test", "blob_col", 1, 1) as blob: + blob.write("a" * 100) + self.assertEqual(str(self.cx.execute("select blob_col from test").fetchone()[0]), data) + + def CheckContextCloseBlob(self): + with self.cx.blob("test", "blob_col", 1) as blob: + blob.seek(10) + try: + blob.close() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except Exception: + self.fail("Should have raised a ProgrammingError") + + + def suite(): module_suite = unittest.makeSuite(ModuleTests, "Check") connection_suite = unittest.makeSuite(ConnectionTests, "Check") cursor_suite = unittest.makeSuite(CursorTests, "Check") + blob_suite = unittest.makeSuite(BlobTests, "Check") thread_suite = unittest.makeSuite(ThreadTests, "Check") constructor_suite = unittest.makeSuite(ConstructorTests, "Check") ext_suite = unittest.makeSuite(ExtensionTests, "Check") closed_con_suite = unittest.makeSuite(ClosedConTests, "Check") closed_cur_suite = unittest.makeSuite(ClosedCurTests, "Check") - return unittest.TestSuite((module_suite, connection_suite, cursor_suite, thread_suite, constructor_suite, ext_suite, closed_con_suite, closed_cur_suite)) + closed_blob_suite = unittest.makeSuite(ClosedBlobTests, "Check") + blob_context_manager_suite = unittest.makeSuite(BlobContextManagerTests, "Check") + return unittest.TestSuite((module_suite, connection_suite, cursor_suite, blob_suite, thread_suite, + constructor_suite, ext_suite, closed_con_suite, closed_cur_suite, closed_blob_suite, + blob_context_manager_suite, context_suite)) def test(): runner = unittest.TextTestRunner() -- cgit v1.2.3