Lines Matching refs:self
59 def test_api_level(self):
60 self.assertEqual(sqlite.apilevel, "2.0",
63 def test_thread_safety(self):
64 self.assertIn(sqlite.threadsafety, {0, 1, 3},
68 def test_param_style(self):
69 self.assertEqual(sqlite.paramstyle, "qmark",
73 def test_warning(self):
74 self.assertTrue(issubclass(sqlite.Warning, Exception),
77 def test_error(self):
78 self.assertTrue(issubclass(sqlite.Error, Exception),
81 def test_interface_error(self):
82 self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error),
85 def test_database_error(self):
86 self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error),
89 def test_data_error(self):
90 self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError),
93 def test_operational_error(self):
94 self.assertTrue(issubclass(sqlite.OperationalError, sqlite.DatabaseError),
97 def test_integrity_error(self):
98 self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError),
101 def test_internal_error(self):
102 self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError),
105 def test_programming_error(self):
106 self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError),
109 def test_not_supported_error(self):
110 self.assertTrue(issubclass(sqlite.NotSupportedError,
114 def test_module_constants(self):
305 with self.subTest(const=const):
306 self.assertTrue(hasattr(sqlite, const))
308 def test_error_code_on_exception(self):
316 with self.assertRaisesRegex(sqlite.Error, err_msg) as cm:
319 self.assertEqual(e.sqlite_errorcode, err_code)
320 self.assertTrue(e.sqlite_errorname.startswith("SQLITE_CANTOPEN"))
324 def test_extended_error_code_on_exception(self):
329 with self.assertRaisesRegex(sqlite.IntegrityError, errmsg) as cm:
332 self.assertEqual(exc.sqlite_errorcode,
334 self.assertEqual(exc.sqlite_errorname, "SQLITE_CONSTRAINT_CHECK")
339 def test_shared_cache_deprecated(self):
341 with self.assertWarns(DeprecationWarning) as cm:
343 self.assertIn("dbapi.py", cm.filename)
345 def test_disallow_instantiation(self):
347 check_disallow_instantiation(self, type(cx("select 1")))
348 check_disallow_instantiation(self, sqlite.Blob)
350 def test_complete_statement(self):
351 self.assertFalse(sqlite.complete_statement("select t"))
352 self.assertTrue(sqlite.complete_statement("create table t(t);"))
357 def setUp(self):
358 self.cx = sqlite.connect(":memory:")
359 cu = self.cx.cursor()
363 def tearDown(self):
364 self.cx.close()
366 def test_commit(self):
367 self.cx.commit()
369 def test_commit_after_no_changes(self):
373 self.cx.commit()
374 self.cx.commit()
376 def test_rollback(self):
377 self.cx.rollback()
379 def test_rollback_after_no_changes(self):
383 self.cx.rollback()
384 self.cx.rollback()
386 def test_cursor(self):
387 cu = self.cx.cursor()
389 def test_failed_open(self):
391 with self.assertRaises(sqlite.OperationalError):
394 def test_close(self):
395 self.cx.close()
397 def test_use_after_close(self):
399 cu = self.cx.cursor()
401 self.cx.close()
402 self.assertRaises(sqlite.ProgrammingError, res.fetchall)
403 self.assertRaises(sqlite.ProgrammingError, cu.execute, sql)
404 self.assertRaises(sqlite.ProgrammingError, cu.executemany, sql, [])
405 self.assertRaises(sqlite.ProgrammingError, cu.executescript, sql)
406 self.assertRaises(sqlite.ProgrammingError, self.cx.execute, sql)
407 self.assertRaises(sqlite.ProgrammingError,
408 self.cx.executemany, sql, [])
409 self.assertRaises(sqlite.ProgrammingError, self.cx.executescript, sql)
410 self.assertRaises(sqlite.ProgrammingError,
411 self.cx.create_function, "t", 1, lambda x: x)
412 self.assertRaises(sqlite.ProgrammingError, self.cx.cursor)
413 with self.assertRaises(sqlite.ProgrammingError):
414 with self.cx:
417 def test_exceptions(self):
419 self.assertEqual(self.cx.Warning, sqlite.Warning)
420 self.assertEqual(self.cx.Error, sqlite.Error)
421 self.assertEqual(self.cx.InterfaceError, sqlite.InterfaceError)
422 self.assertEqual(self.cx.DatabaseError, sqlite.DatabaseError)
423 self.assertEqual(self.cx.DataError, sqlite.DataError)
424 self.assertEqual(self.cx.OperationalError, sqlite.OperationalError)
425 self.assertEqual(self.cx.IntegrityError, sqlite.IntegrityError)
426 self.assertEqual(self.cx.InternalError, sqlite.InternalError)
427 self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError)
428 self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError)
430 def test_in_transaction(self):
434 self.assertEqual(cx.in_transaction, False)
436 self.assertEqual(cx.in_transaction, False)
438 self.assertEqual(cx.in_transaction, True)
441 self.assertEqual(cx.in_transaction, True)
443 self.assertEqual(cx.in_transaction, False)
446 self.assertEqual(cx.in_transaction, False)
448 def test_in_transaction_ro(self):
449 with self.assertRaises(AttributeError):
450 self.cx.in_transaction = True
452 def test_connection_exceptions(self):
465 with self.subTest(exc=exc):
466 self.assertTrue(hasattr(self.cx, exc))
467 self.assertIs(getattr(sqlite, exc), getattr(self.cx, exc))
469 def test_interrupt_on_closed_db(self):
472 with self.assertRaises(sqlite.ProgrammingError):
475 def test_interrupt(self):
476 self.assertIsNone(self.cx.interrupt())
478 def test_drop_unused_refs(self):
480 cu = self.cx.execute(f"select {n}")
481 self.assertEqual(cu.fetchone()[0], n)
483 def test_connection_limits(self):
485 saved_limit = self.cx.getlimit(category)
488 prev_limit = self.cx.setlimit(category, new_limit)
489 self.assertEqual(saved_limit, prev_limit)
490 self.assertEqual(self.cx.getlimit(category), new_limit)
492 self.assertRaisesRegex(sqlite.DataError, msg,
493 self.cx.execute, "select 1 as '16'")
495 self.cx.setlimit(category, saved_limit)
497 def test_connection_bad_limit_category(self):
500 self.assertRaisesRegex(sqlite.ProgrammingError, msg,
501 self.cx.getlimit, cat)
502 self.assertRaisesRegex(sqlite.ProgrammingError, msg,
503 self.cx.setlimit, cat, 0)
505 def test_connection_init_bad_isolation_level(self):
521 with self.subTest(level=level):
522 with self.assertRaisesRegex(ValueError, msg):
525 with self.assertRaisesRegex(ValueError, msg):
528 self.assertEqual(cx.isolation_level, "")
530 def test_connection_init_good_isolation_levels(self):
532 with self.subTest(level=level):
534 self.assertEqual(cx.isolation_level, level)
536 self.assertEqual(cx.isolation_level, "")
538 self.assertEqual(cx.isolation_level, level)
540 def test_connection_reinit(self):
552 self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
553 self.assertEqual([r[0] for r in rows], [b"0", b"1"])
562 self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
563 self.assertEqual([r[0] for r in rows], ["2", "3"])
565 def test_connection_bad_reinit(self):
570 self.assertRaisesRegex(sqlite.OperationalError,
573 self.assertRaisesRegex(sqlite.ProgrammingError,
580 def setUp(self):
581 self.cx = sqlite.Connection.__new__(sqlite.Connection)
583 def test_uninit_operations(self):
585 lambda: self.cx.isolation_level,
586 lambda: self.cx.total_changes,
587 lambda: self.cx.in_transaction,
588 lambda: self.cx.iterdump(),
589 lambda: self.cx.cursor(),
590 lambda: self.cx.close(),
593 with self.subTest(func=func):
594 self.assertRaisesRegex(sqlite.ProgrammingError,
602 def test_serialize_deserialize(self):
612 with self.assertRaisesRegex(sqlite.OperationalError, regex):
619 def test_deserialize_wrong_args(self):
627 with self.subTest(exc=exc, arg=arg):
629 self.assertRaises(exc, cx.deserialize, arg)
631 def test_deserialize_corrupt_database(self):
634 with self.assertRaisesRegex(sqlite.DatabaseError, regex):
642 def test_deserialize_too_much_data_64bit(self):
644 with self.assertRaisesRegex(OverflowError, "'data' is too large"):
651 def test_open_with_path_like_object(self):
655 self.addCleanup(unlink, path)
656 self.assertFalse(os.path.exists(path))
658 self.assertTrue(os.path.exists(path))
659 cx.execute(self._sql)
665 def test_open_with_undecodable_path(self):
667 self.addCleanup(unlink, path)
668 self.assertFalse(os.path.exists(path))
670 self.assertTrue(os.path.exists(path))
671 cx.execute(self._sql)
673 def test_open_uri(self):
675 self.addCleanup(unlink, path)
677 self.assertFalse(os.path.exists(path))
679 self.assertTrue(os.path.exists(path))
680 cx.execute(self._sql)
682 def test_open_unquoted_uri(self):
684 self.addCleanup(unlink, path)
686 self.assertFalse(os.path.exists(path))
688 self.assertTrue(os.path.exists(path))
689 cx.execute(self._sql)
691 def test_open_uri_readonly(self):
693 self.addCleanup(unlink, path)
695 self.assertFalse(os.path.exists(path))
697 with self.assertRaises(sqlite.OperationalError):
699 self.assertFalse(os.path.exists(path))
701 self.assertTrue(os.path.exists(path))
704 with self.assertRaises(sqlite.OperationalError):
705 cx.execute(self._sql)
711 def test_open_undecodable_uri(self):
713 self.addCleanup(unlink, path)
715 self.assertFalse(os.path.exists(path))
717 self.assertTrue(os.path.exists(path))
718 cx.execute(self._sql)
720 def test_factory_database_arg(self):
730 self.assertEqual(database_arg, database)
732 def test_database_keyword(self):
734 self.assertEqual(type(cx), sqlite.Connection)
738 def setUp(self):
739 self.cx = sqlite.connect(":memory:")
740 self.cu = self.cx.cursor()
741 self.cu.execute(
745 self.cu.execute("insert into test(name) values (?)", ("foo",))
747 def tearDown(self):
748 self.cu.close()
749 self.cx.close()
751 def test_execute_no_args(self):
752 self.cu.execute("delete from test")
754 def test_execute_illegal_sql(self):
755 with self.assertRaises(sqlite.OperationalError):
756 self.cu.execute("select asdf")
758 def test_execute_multiple_statements(self):
776 with self.subTest(query=query):
777 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
778 self.cu.execute(query)
780 def test_execute_with_appended_comments(self):
794 with self.subTest(query=query):
795 self.cu.execute(query)
797 def test_execute_wrong_sql_arg(self):
798 with self.assertRaises(TypeError):
799 self.cu.execute(42)
801 def test_execute_arg_int(self):
802 self.cu.execute("insert into test(id) values (?)", (42,))
804 def test_execute_arg_float(self):
805 self.cu.execute("insert into test(income) values (?)", (2500.32,))
807 def test_execute_arg_string(self):
808 self.cu.execute("insert into test(name) values (?)", ("Hugo",))
810 def test_execute_arg_string_with_zero_byte(self):
811 self.cu.execute("insert into test(name) values (?)", ("Hu\x00go",))
813 self.cu.execute("select name from test where id=?", (self.cu.lastrowid,))
814 row = self.cu.fetchone()
815 self.assertEqual(row[0], "Hu\x00go")
817 def test_execute_non_iterable(self):
818 with self.assertRaises(sqlite.ProgrammingError) as cm:
819 self.cu.execute("insert into test(id) values (?)", 42)
820 self.assertEqual(str(cm.exception), 'parameters are of unsupported type')
822 def test_execute_wrong_no_of_args1(self):
824 with self.assertRaises(sqlite.ProgrammingError):
825 self.cu.execute("insert into test(id) values (?)", (17, "Egon"))
827 def test_execute_wrong_no_of_args2(self):
829 with self.assertRaises(sqlite.ProgrammingError):
830 self.cu.execute("insert into test(id) values (?)")
832 def test_execute_wrong_no_of_args3(self):
834 with self.assertRaises(sqlite.ProgrammingError):
835 self.cu.execute("insert into test(id) values (?)")
837 def test_execute_param_list(self):
838 self.cu.execute("insert into test(name) values ('foo')")
839 self.cu.execute("select name from test where name=?", ["foo"])
840 row = self.cu.fetchone()
841 self.assertEqual(row[0], "foo")
843 def test_execute_param_sequence(self):
845 def __len__(self):
847 def __getitem__(self, x):
851 self.cu.execute("insert into test(name) values ('foo')")
852 self.cu.execute("select name from test where name=?", L())
853 row = self.cu.fetchone()
854 self.assertEqual(row[0], "foo")
856 def test_execute_param_sequence_bad_len(self):
859 def __len__(self):
864 self.cu.execute("insert into test(name) values ('foo')")
865 with self.assertRaises(ZeroDivisionError):
866 self.cu.execute("select name from test where name=?", L())
868 def test_execute_too_many_params(self):
871 with cx_limit(self.cx, category=category, limit=1):
872 self.cu.execute("select * from test where id=?", (1,))
873 with self.assertRaisesRegex(sqlite.OperationalError, msg):
874 self.cu.execute("select * from test where id!=? and id!=?",
877 def test_execute_dict_mapping(self):
878 self.cu.execute("insert into test(name) values ('foo')")
879 self.cu.execute("select name from test where name=:name", {"name": "foo"})
880 row = self.cu.fetchone()
881 self.assertEqual(row[0], "foo")
883 def test_execute_dict_mapping_mapping(self):
885 def __missing__(self, key):
888 self.cu.execute("insert into test(name) values ('foo')")
889 self.cu.execute("select name from test where name=:name", D())
890 row = self.cu.fetchone()
891 self.assertEqual(row[0], "foo")
893 def test_execute_dict_mapping_too_little_args(self):
894 self.cu.execute("insert into test(name) values ('foo')")
895 with self.assertRaises(sqlite.ProgrammingError):
896 self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"})
898 def test_execute_dict_mapping_no_args(self):
899 self.cu.execute("insert into test(name) values ('foo')")
900 with self.assertRaises(sqlite.ProgrammingError):
901 self.cu.execute("select name from test where name=:name")
903 def test_execute_dict_mapping_unnamed(self):
904 self.cu.execute("insert into test(name) values ('foo')")
905 with self.assertRaises(sqlite.ProgrammingError):
906 self.cu.execute("select name from test where name=?", {"name": "foo"})
908 def test_close(self):
909 self.cu.close()
911 def test_rowcount_execute(self):
912 self.cu.execute("delete from test")
913 self.cu.execute("insert into test(name) values ('foo')")
914 self.cu.execute("insert into test(name) values ('foo')")
915 self.cu.execute("update test set name='bar'")
916 self.assertEqual(self.cu.rowcount, 2)
918 def test_rowcount_select(self):
924 self.cu.execute("select 5 union select 6")
925 self.assertEqual(self.cu.rowcount, -1)
927 def test_rowcount_executemany(self):
928 self.cu.execute("delete from test")
929 self.cu.executemany("insert into test(name) values (?)", [(1,), (2,), (3,)])
930 self.assertEqual(self.cu.rowcount, 3)
934 def test_rowcount_update_returning(self):
936 self.cu.execute("update test set name='bar' where name='foo' returning 1")
937 self.assertEqual(self.cu.fetchone()[0], 1)
938 self.assertEqual(self.cu.rowcount, 1)
940 def test_rowcount_prefixed_with_comment(self):
942 self.cu.execute("""
946 self.assertEqual(self.cu.rowcount, 2)
947 self.cu.execute("""
952 self.assertEqual(self.cu.rowcount, 1)
953 self.cu.execute("/* bar */ update test set name='bar' where name='foo'")
954 self.assertEqual(self.cu.rowcount, 3)
956 def test_rowcount_vaccuum(self):
958 self.cu.executemany("insert into test(income) values(?)", data)
959 self.assertEqual(self.cu.rowcount, 3)
960 self.cx.commit()
961 self.cu.execute("vacuum")
962 self.assertEqual(self.cu.rowcount, -1)
964 def test_total_changes(self):
965 self.cu.execute("insert into test(name) values ('foo')")
966 self.cu.execute("insert into test(name) values ('foo')")
967 self.assertLess(2, self.cx.total_changes, msg='total changes reported wrong value')
973 def test_execute_many_sequence(self):
974 self.cu.executemany("insert into test(income) values (?)", [(x,) for x in range(100, 110)])
976 def test_execute_many_iterator(self):
978 def __init__(self):
979 self.value = 5
981 def __iter__(self):
982 return self
984 def __next__(self):
985 if self.value == 10:
988 self.value += 1
989 return (self.value,)
991 self.cu.executemany("insert into test(income) values (?)", MyIter())
993 def test_execute_many_generator(self):
998 self.cu.executemany("insert into test(income) values (?)", mygen())
1000 def test_execute_many_wrong_sql_arg(self):
1001 with self.assertRaises(TypeError):
1002 self.cu.executemany(42, [(3,)])
1004 def test_execute_many_select(self):
1005 with self.assertRaises(sqlite.ProgrammingError):
1006 self.cu.executemany("select ?", [(3,)])
1008 def test_execute_many_not_iterable(self):
1009 with self.assertRaises(TypeError):
1010 self.cu.executemany("insert into test(income) values (?)", 42)
1012 def test_fetch_iter(self):
1014 self.cu.execute("delete from test")
1015 self.cu.execute("insert into test(id) values (?)", (5,))
1016 self.cu.execute("insert into test(id) values (?)", (6,))
1017 self.cu.execute("select id from test order by id")
1019 for row in self.cu:
1021 self.assertEqual(lst[0], 5)
1022 self.assertEqual(lst[1], 6)
1024 def test_fetchone(self):
1025 self.cu.execute("select name from test")
1026 row = self.cu.fetchone()
1027 self.assertEqual(row[0], "foo")
1028 row = self.cu.fetchone()
1029 self.assertEqual(row, None)
1031 def test_fetchone_no_statement(self):
1032 cur = self.cx.cursor()
1034 self.assertEqual(row, None)
1036 def test_array_size(self):
1038 self.assertEqual(self.cu.arraysize, 1)
1041 self.cu.arraysize = 2
1044 self.cu.execute("delete from test")
1045 self.cu.execute("insert into test(name) values ('A')")
1046 self.cu.execute("insert into test(name) values ('B')")
1047 self.cu.execute("insert into test(name) values ('C')")
1048 self.cu.execute("select name from test")
1049 res = self.cu.fetchmany()
1051 self.assertEqual(len(res), 2)
1053 def test_fetchmany(self):
1054 self.cu.execute("select name from test")
1055 res = self.cu.fetchmany(100)
1056 self.assertEqual(len(res), 1)
1057 res = self.cu.fetchmany(100)
1058 self.assertEqual(res, [])
1060 def test_fetchmany_kw_arg(self):
1062 self.cu.execute("select name from test")
1063 res = self.cu.fetchmany(size=100)
1064 self.assertEqual(len(res), 1)
1066 def test_fetchall(self):
1067 self.cu.execute("select name from test")
1068 res = self.cu.fetchall()
1069 self.assertEqual(len(res), 1)
1070 res = self.cu.fetchall()
1071 self.assertEqual(res, [])
1073 def test_setinputsizes(self):
1074 self.cu.setinputsizes([3, 4, 5])
1076 def test_setoutputsize(self):
1077 self.cu.setoutputsize(5, 0)
1079 def test_setoutputsize_no_column(self):
1080 self.cu.setoutputsize(42)
1082 def test_cursor_connection(self):
1084 self.assertEqual(self.cu.connection, self.cx)
1086 def test_wrong_cursor_callable(self):
1087 with self.assertRaises(TypeError):
1089 cur = self.cx.cursor(f)
1091 def test_cursor_wrong_class(self):
1094 with self.assertRaises(TypeError):
1097 def test_last_row_id_on_replace(self):
1103 with self.subTest(statement=statement):
1104 self.cu.execute(sql.format(statement), (1, 'foo'))
1105 self.assertEqual(self.cu.lastrowid, 1)
1107 def test_last_row_id_on_ignore(self):
1108 self.cu.execute(
1111 self.assertEqual(self.cu.lastrowid, 2)
1112 self.cu.execute(
1115 self.assertEqual(self.cu.lastrowid, 2)
1117 def test_last_row_id_insert_o_r(self):
1121 with self.subTest(statement='INSERT OR {}'.format(statement)):
1122 self.cu.execute(sql.format(statement), (statement,))
1123 results.append((statement, self.cu.lastrowid))
1124 with self.assertRaises(sqlite.IntegrityError):
1125 self.cu.execute(sql.format(statement), (statement,))
1126 results.append((statement, self.cu.lastrowid))
1132 self.assertEqual(results, expected)
1134 def test_column_count(self):
1137 res = self.cu.execute(select)
1140 self.cu.execute("alter table test add newcol")
1141 res = self.cu.execute(select)
1143 self.assertEqual(new_count - old_count, 1)
1145 def test_same_query_in_multiple_cursors(self):
1146 cursors = [self.cx.execute("select 1") for _ in range(3)]
1148 self.assertEqual(cu.fetchall(), [(1,)])
1152 def setUp(self):
1153 self.cx = sqlite.connect(":memory:")
1154 self.cx.execute("create table test(b blob)")
1155 self.data = b"this blob data string is exactly fifty bytes long!"
1156 self.cx.execute("insert into test(b) values (?)", (self.data,))
1157 self.blob = self.cx.blobopen("test", "b", 1)
1159 def tearDown(self):
1160 self.blob.close()
1161 self.cx.close()
1163 def test_blob_is_a_blob(self):
1164 self.assertIsInstance(self.blob, sqlite.Blob)
1166 def test_blob_seek_and_tell(self):
1167 self.blob.seek(10)
1168 self.assertEqual(self.blob.tell(), 10)
1170 self.blob.seek(10, SEEK_SET)
1171 self.assertEqual(self.blob.tell(), 10)
1173 self.blob.seek(10, SEEK_CUR)
1174 self.assertEqual(self.blob.tell(), 20)
1176 self.blob.seek(-10, SEEK_END)
1177 self.assertEqual(self.blob.tell(), 40)
1179 def test_blob_seek_error(self):
1185 (ValueError, msg_oor, lambda: self.blob.seek(1000)),
1186 (ValueError, msg_oor, lambda: self.blob.seek(-10)),
1187 (ValueError, msg_orig, lambda: self.blob.seek(10, -1)),
1188 (ValueError, msg_orig, lambda: self.blob.seek(10, 3)),
1191 with self.subTest(exc=exc, msg=msg, fn=fn):
1192 self.assertRaisesRegex(exc, msg, fn)
1195 self.blob.seek(1, SEEK_SET)
1196 with self.assertRaisesRegex(OverflowError, msg_of):
1197 self.blob.seek(INT_MAX, SEEK_CUR)
1198 with self.assertRaisesRegex(OverflowError, msg_of):
1199 self.blob.seek(INT_MAX, SEEK_END)
1201 def test_blob_read(self):
1202 buf = self.blob.read()
1203 self.assertEqual(buf, self.data)
1205 def test_blob_read_oversized(self):
1206 buf = self.blob.read(len(self.data) * 2)
1207 self.assertEqual(buf, self.data)
1209 def test_blob_read_advance_offset(self):
1211 buf = self.blob.read(n)
1212 self.assertEqual(buf, self.data[:n])
1213 self.assertEqual(self.blob.tell(), n)
1215 def test_blob_read_at_offset(self):
1216 self.blob.seek(10)
1217 self.assertEqual(self.blob.read(10), self.data[10:20])
1219 def test_blob_read_error_row_changed(self):
1220 self.cx.execute("update test set b='aaaa' where rowid=1")
1221 with self.assertRaises(sqlite.OperationalError):
1222 self.blob.read()
1224 def test_blob_write(self):
1226 self.blob.write(new_data)
1227 row = self.cx.execute("select b from test").fetchone()
1228 self.assertEqual(row[0], new_data)
1230 def test_blob_write_at_offset(self):
1232 self.blob.seek(25)
1233 self.blob.write(new_data)
1234 row = self.cx.execute("select b from test").fetchone()
1235 self.assertEqual(row[0], self.data[:25] + new_data)
1237 def test_blob_write_advance_offset(self):
1238 self.blob.write(b"d"*10)
1239 self.assertEqual(self.blob.tell(), 10)
1241 def test_blob_write_error_length(self):
1242 with self.assertRaisesRegex(ValueError, "data longer than blob"):
1243 self.blob.write(b"a" * 1000)
1245 self.blob.seek(0, SEEK_SET)
1246 n = len(self.blob)
1247 self.blob.write(b"a" * (n-1))
1248 self.blob.write(b"a")
1249 with self.assertRaisesRegex(ValueError, "data longer than blob"):
1250 self.blob.write(b"a")
1252 def test_blob_write_error_row_changed(self):
1253 self.cx.execute("update test set b='aaaa' where rowid=1")
1254 with self.assertRaises(sqlite.OperationalError):
1255 self.blob.write(b"aaa")
1257 def test_blob_write_error_readonly(self):
1258 ro_blob = self.cx.blobopen("test", "b", 1, readonly=True)
1259 with self.assertRaisesRegex(sqlite.OperationalError, "readonly"):
1263 def test_blob_open_error(self):
1272 with self.subTest(args=args, kwds=kwds):
1273 with self.assertRaisesRegex(sqlite.OperationalError, regex):
1274 self.cx.blobopen(*args, **kwds)
1276 def test_blob_length(self):
1277 self.assertEqual(len(self.blob), 50)
1279 def test_blob_get_item(self):
1280 self.assertEqual(self.blob[5], ord("b"))
1281 self.assertEqual(self.blob[6], ord("l"))
1282 self.assertEqual(self.blob[7], ord("o"))
1283 self.assertEqual(self.blob[8], ord("b"))
1284 self.assertEqual(self.blob[-1], ord("!"))
1286 def test_blob_set_item(self):
1287 self.blob[0] = ord("b")
1288 expected = b"b" + self.data[1:]
1289 actual = self.cx.execute("select b from test").fetchone()[0]
1290 self.assertEqual(actual, expected)
1292 def test_blob_set_item_with_offset(self):
1293 self.blob.seek(0, SEEK_END)
1294 self.assertEqual(self.blob.read(), b"") # verify that we're at EOB
1295 self.blob[0] = ord("T")
1296 self.blob[-1] = ord(".")
1297 self.blob.seek(0, SEEK_SET)
1299 self.assertEqual(self.blob.read(), expected)
1301 def test_blob_set_slice_buffer_object(self):
1303 self.blob[0:5] = memoryview(b"12345")
1304 self.assertEqual(self.blob[0:5], b"12345")
1306 self.blob[0:5] = bytearray(b"23456")
1307 self.assertEqual(self.blob[0:5], b"23456")
1309 self.blob[0:5] = array("b", [1, 2, 3, 4, 5])
1310 self.assertEqual(self.blob[0:5], b"\x01\x02\x03\x04\x05")
1312 def test_blob_set_item_negative_index(self):
1313 self.blob[-1] = 255
1314 self.assertEqual(self.blob[-1], 255)
1316 def test_blob_get_slice(self):
1317 self.assertEqual(self.blob[5:14], b"blob data")
1319 def test_blob_get_empty_slice(self):
1320 self.assertEqual(self.blob[5:5], b"")
1322 def test_blob_get_slice_negative_index(self):
1323 self.assertEqual(self.blob[5:-5], self.data[5:-5])
1325 def test_blob_get_slice_with_skip(self):
1326 self.assertEqual(self.blob[0:10:2], b"ti lb")
1328 def test_blob_set_slice(self):
1329 self.blob[0:5] = b"12345"
1330 expected = b"12345" + self.data[5:]
1331 actual = self.cx.execute("select b from test").fetchone()[0]
1332 self.assertEqual(actual, expected)
1334 def test_blob_set_empty_slice(self):
1335 self.blob[0:0] = b""
1336 self.assertEqual(self.blob[:], self.data)
1338 def test_blob_set_slice_with_skip(self):
1339 self.blob[0:10:2] = b"12345"
1340 actual = self.cx.execute("select b from test").fetchone()[0]
1341 expected = b"1h2s3b4o5 " + self.data[10:]
1342 self.assertEqual(actual, expected)
1344 def test_blob_mapping_invalid_index_type(self):
1346 with self.assertRaisesRegex(TypeError, msg):
1347 self.blob[5:5.5]
1348 with self.assertRaisesRegex(TypeError, msg):
1349 self.blob[1.5]
1350 with self.assertRaisesRegex(TypeError, msg):
1351 self.blob["a"] = b"b"
1353 def test_blob_get_item_error(self):
1354 dataset = [len(self.blob), 105, -105]
1356 with self.subTest(idx=idx):
1357 with self.assertRaisesRegex(IndexError, "index out of range"):
1358 self.blob[idx]
1359 with self.assertRaisesRegex(IndexError, "cannot fit 'int'"):
1360 self.blob[ULLONG_MAX]
1363 self.cx.execute("update test set b='aaaa' where rowid=1")
1364 with self.assertRaises(sqlite.OperationalError):
1365 self.blob[0]
1367 def test_blob_set_item_error(self):
1368 with self.assertRaisesRegex(TypeError, "cannot be interpreted"):
1369 self.blob[0] = b"multiple"
1370 with self.assertRaisesRegex(TypeError, "cannot be interpreted"):
1371 self.blob[0] = b"1"
1372 with self.assertRaisesRegex(TypeError, "cannot be interpreted"):
1373 self.blob[0] = bytearray(b"1")
1374 with self.assertRaisesRegex(TypeError, "doesn't support.*deletion"):
1375 del self.blob[0]
1376 with self.assertRaisesRegex(IndexError, "Blob index out of range"):
1377 self.blob[1000] = 0
1378 with self.assertRaisesRegex(ValueError, "must be in range"):
1379 self.blob[0] = -1
1380 with self.assertRaisesRegex(ValueError, "must be in range"):
1381 self.blob[0] = 256
1383 with self.assertRaisesRegex(ValueError, "must be in range"):
1384 self.blob[0] = 2**65
1386 def test_blob_set_slice_error(self):
1387 with self.assertRaisesRegex(IndexError, "wrong size"):
1388 self.blob[5:10] = b"a"
1389 with self.assertRaisesRegex(IndexError, "wrong size"):
1390 self.blob[5:10] = b"a" * 1000
1391 with self.assertRaisesRegex(TypeError, "doesn't support.*deletion"):
1392 del self.blob[5:10]
1393 with self.assertRaisesRegex(ValueError, "step cannot be zero"):
1394 self.blob[5:10:0] = b"12345"
1395 with self.assertRaises(BufferError):
1396 self.blob[5:10] = memoryview(b"abcde")[::2]
1398 def test_blob_sequence_not_supported(self):
1399 with self.assertRaisesRegex(TypeError, "unsupported operand"):
1400 self.blob + self.blob
1401 with self.assertRaisesRegex(TypeError, "unsupported operand"):
1402 self.blob * 5
1403 with self.assertRaisesRegex(TypeError, "is not iterable"):
1404 b"a" in self.blob
1406 def test_blob_context_manager(self):
1408 with self.cx.blobopen("test", "b", 1) as blob:
1410 actual = self.cx.execute("select b from test").fetchone()[0]
1411 self.assertEqual(actual, data)
1414 with self.assertRaisesRegex(sqlite.ProgrammingError, "closed blob"):
1417 def test_blob_context_manager_reraise_exceptions(self):
1420 with self.assertRaisesRegex(DummyException, "reraised"):
1421 with self.cx.blobopen("test", "b", 1) as blob:
1425 def test_blob_closed(self):
1433 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1435 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1437 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1439 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1441 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1443 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1445 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1447 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1449 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1451 with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
1454 def test_blob_closed_db_read(self):
1460 self.assertRaisesRegex(sqlite.ProgrammingError,
1464 def test_blob_32bit_rowid(self):
1475 def setUp(self):
1476 self.con = sqlite.connect(":memory:")
1477 self.cur = self.con.cursor()
1478 self.cur.execute("create table test(name text, b blob)")
1479 self.cur.execute("insert into test values('blob', zeroblob(1))")
1481 def tearDown(self):
1482 self.cur.close()
1483 self.con.close()
1486 def _run_test(self, fn, *args, **kwds):
1501 self.fail("\n".join(err))
1503 def test_check_connection_thread(self):
1505 lambda: self.con.cursor(),
1506 lambda: self.con.commit(),
1507 lambda: self.con.rollback(),
1508 lambda: self.con.close(),
1509 lambda: self.con.set_trace_callback(None),
1510 lambda: self.con.set_authorizer(None),
1511 lambda: self.con.create_collation("foo", None),
1512 lambda: self.con.setlimit(sqlite.SQLITE_LIMIT_LENGTH, -1),
1513 lambda: self.con.getlimit(sqlite.SQLITE_LIMIT_LENGTH),
1514 lambda: self.con.blobopen("test", "b", 1),
1517 fns.append(lambda: self.con.serialize())
1518 fns.append(lambda: self.con.deserialize(b""))
1520 fns.append(lambda: self.con.create_window_function("foo", 0, None))
1523 with self.subTest(fn=fn):
1524 self._run_test(fn)
1526 def test_check_cursor_thread(self):
1528 lambda: self.cur.execute("insert into test(name) values('a')"),
1529 lambda: self.cur.close(),
1530 lambda: self.cur.execute("select name from test"),
1531 lambda: self.cur.fetchone(),
1534 with self.subTest(fn=fn):
1535 self._run_test(fn)
1539 def test_dont_check_same_thread(self):
1551 self.assertEqual(len(err), 0, "\n".join(err))
1555 def test_date(self):
1558 def test_time(self):
1561 def test_timestamp(self):
1564 def test_date_from_ticks(self):
1567 def test_time_from_ticks(self):
1570 def test_timestamp_from_ticks(self):
1573 def test_binary(self):
1577 def test_script_string_sql(self):
1588 self.assertEqual(res, 5)
1590 def test_script_syntax_error(self):
1593 with self.assertRaises(sqlite.OperationalError):
1596 def test_script_error_normal(self):
1599 with self.assertRaises(sqlite.OperationalError):
1602 def test_cursor_executescript_as_bytes(self):
1605 with self.assertRaises(TypeError):
1608 def test_cursor_executescript_with_null_characters(self):
1611 with self.assertRaises(ValueError):
1617 def test_cursor_executescript_with_surrogates(self):
1620 with self.assertRaises(UnicodeEncodeError):
1626 def test_cursor_executescript_too_large_script(self):
1630 with self.assertRaisesRegex(sqlite.DataError, msg):
1633 def test_cursor_executescript_tx_control(self):
1636 self.assertTrue(con.in_transaction)
1638 self.assertFalse(con.in_transaction)
1640 def test_connection_execute(self):
1643 self.assertEqual(result, 5, "Basic test of Connection.execute")
1645 def test_connection_executemany(self):
1650 self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany")
1651 self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany")
1653 def test_connection_executescript(self):
1657 self.assertEqual(result, 5, "Basic test of Connection.executescript")
1660 def test_closed_con_cursor(self):
1663 with self.assertRaises(sqlite.ProgrammingError):
1666 def test_closed_con_commit(self):
1669 with self.assertRaises(sqlite.ProgrammingError):
1672 def test_closed_con_rollback(self):
1675 with self.assertRaises(sqlite.ProgrammingError):
1678 def test_closed_cur_execute(self):
1682 with self.assertRaises(sqlite.ProgrammingError):
1685 def test_closed_create_function(self):
1689 with self.assertRaises(sqlite.ProgrammingError):
1692 def test_closed_create_aggregate(self):
1696 def __init__(self):
1698 def step(self, x):
1700 def finalize(self):
1702 with self.assertRaises(sqlite.ProgrammingError):
1705 def test_closed_set_authorizer(self):
1710 with self.assertRaises(sqlite.ProgrammingError):
1713 def test_closed_set_progress_callback(self):
1717 with self.assertRaises(sqlite.ProgrammingError):
1720 def test_closed_call(self):
1723 with self.assertRaises(sqlite.ProgrammingError):
1727 def test_closed(self):
1740 with self.assertRaises(sqlite.ProgrammingError):
1752 def setUp(self):
1753 self.cx = sqlite.connect(":memory:")
1754 self.cu = self.cx.cursor()
1755 self.cu.execute("""
1761 def tearDown(self):
1762 self.cu.close()
1763 self.cx.close()
1765 def test_on_conflict_rollback_with_explicit_transaction(self):
1766 self.cx.isolation_level = None # autocommit mode
1767 self.cu = self.cx.cursor()
1769 self.cu.execute("BEGIN")
1770 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
1771 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
1772 with self.assertRaises(sqlite.IntegrityError):
1773 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
1775 self.cx.commit()
1776 self.cu.execute("SELECT name, unique_name from test")
1778 self.assertEqual(self.cu.fetchall(), [])
1780 def test_on_conflict_abort_raises_with_explicit_transactions(self):
1783 self.cx.isolation_level = None # autocommit mode
1784 self.cu = self.cx.cursor()
1786 self.cu.execute("BEGIN")
1787 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
1788 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
1789 with self.assertRaises(sqlite.IntegrityError):
1790 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
1791 self.cx.commit()
1792 self.cu.execute("SELECT name, unique_name FROM test")
1794 self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)])
1796 def test_on_conflict_rollback_without_transaction(self):
1798 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
1799 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
1800 with self.assertRaises(sqlite.IntegrityError):
1801 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
1802 self.cu.execute("SELECT name, unique_name FROM test")
1804 self.assertEqual(self.cu.fetchall(), [])
1806 def test_on_conflict_abort_raises_without_transactions(self):
1809 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
1810 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
1811 with self.assertRaises(sqlite.IntegrityError):
1812 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
1814 self.cu.execute("SELECT name, unique_name FROM test")
1815 self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)])
1817 def test_on_conflict_fail(self):
1818 self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')")
1819 with self.assertRaises(sqlite.IntegrityError):
1820 self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')")
1821 self.assertEqual(self.cu.fetchall(), [])
1823 def test_on_conflict_ignore(self):
1824 self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')")
1826 self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')")
1827 self.cu.execute("SELECT unique_name FROM test")
1828 self.assertEqual(self.cu.fetchall(), [('foo',)])
1830 def test_on_conflict_replace(self):
1831 self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Data!', 'foo')")
1833 self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Very different data!', 'foo')")
1834 self.cu.execute("SELECT name, unique_name FROM test")
1835 self.assertEqual(self.cu.fetchall(), [('Very different data!', 'foo')])
1842 def tearDown(self):
1845 def test_ctx_mgr_rollback_if_commit_failed(self):
1853 cx = sqlite3.connect("{TESTFN}", timeout={self.CONNECTION_TIMEOUT})
1883 self.addCleanup(proc.communicate)
1886 self.assertEqual("started", proc.stdout.readline().strip())
1888 cx = sqlite.connect(TESTFN, timeout=self.CONNECTION_TIMEOUT)
1900 self.assertIsNone(proc.returncode)
1907 self.assertEqual(proc.returncode, 0)