mirror
/
qutebrowser
mirror of https://github.com/qutebrowser/qutebrowser.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'history-cleanup'

This commit is contained in:
Florian Bruhin 2021-01-15 17:16:14 +01:00
parent 21e8aad793 485a317379
commit b10c7ee1b7
7 changed files with 353 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
qutebrowser/app.py
View File

@ -66,7 +66,7 @@ from qutebrowser.misc import (ipc, savemanager, sessions, crashsignal,
earlyinit, sql, cmdhistory, backendproblem,
objects, quitter)
from qutebrowser.utils import (log, version, message, utils, urlutils, objreg,
usertypes, standarddir, error, qtutils)
usertypes, standarddir, error, qtutils, debug)
# pylint: disable=unused-import
# We import those to run the cmdutils.register decorators.
from qutebrowser.mainwindow.statusbar import command
@ -445,17 +445,18 @@ def _init_modules(*, args):
downloads.init()
quitter.instance.shutting_down.connect(downloads.shutdown)
try:
log.init.debug("Initializing SQL...")
sql.init(os.path.join(standarddir.data(), 'history.sqlite'))
with debug.log_time("init", "Initializing SQL/history"):
try:
log.init.debug("Initializing SQL...")
sql.init(os.path.join(standarddir.data(), 'history.sqlite'))
log.init.debug("Initializing web history...")
history.init(objects.qapp)
except sql.KnownError as e:
error.handle_fatal_exc(e, 'Error initializing SQL',
pre_text='Error initializing SQL',
no_err_windows=args.no_err_windows)
sys.exit(usertypes.Exit.err_init)
log.init.debug("Initializing web history...")
history.init(objects.qapp)
except sql.KnownError as e:
error.handle_fatal_exc(e, 'Error initializing SQL',
pre_text='Error initializing SQL',
no_err_windows=args.no_err_windows)
sys.exit(usertypes.Exit.err_init)
log.init.debug("Initializing command history...")
cmdhistory.init()

154
qutebrowser/browser/history.py
View File

@ -33,8 +33,6 @@ from qutebrowser.utils import utils, log, usertypes, message, qtutils
from qutebrowser.misc import objects, sql
# increment to indicate that HistoryCompletion must be regenerated
_USER_VERSION = 2
web_history = cast('WebHistory', None)
@ -51,16 +49,23 @@ class HistoryProgress:
self._progress = None
self._value = 0
def start(self, text, maximum):
def start(self, text):
"""Start showing a progress dialog."""
self._progress = QProgressDialog()
self._progress.setMinimumDuration(500)
self._progress.setMaximum(0) # unknown
self._progress.setMinimumDuration(0)
self._progress.setLabelText(text)
self._progress.setMaximum(maximum)
self._progress.setCancelButton(None)
self._progress.setAutoClose(False)
self._progress.show()
QApplication.processEvents()
def set_maximum(self, maximum):
"""Set the progress maximum as soon as we know about it."""
assert self._progress is not None
self._progress.setMaximum(maximum)
QApplication.processEvents()
def tick(self):
"""Increase the displayed progress value."""
self._value += 1
@ -69,7 +74,10 @@ class HistoryProgress:
QApplication.processEvents()
def finish(self):
"""Finish showing the progress dialog."""
"""Finish showing the progress dialog.
After this is called, the object can be reused.
"""
if self._progress is not None:
self._progress.hide()
@ -85,17 +93,20 @@ class CompletionMetaInfo(sql.SqlTable):
def __init__(self, parent=None):
super().__init__("CompletionMetaInfo", ['key', 'value'],
constraints={'key': 'PRIMARY KEY'})
for key, default in self.KEYS.items():
if key not in self:
self[key] = default
# force_rebuild is not in use anymore
self.delete('key', 'force_rebuild', optional=True)
if sql.user_version_changed():
self._init_default_values()
# force_rebuild is not in use anymore
self.delete('key', 'force_rebuild', optional=True)
def _check_key(self, key):
if key not in self.KEYS:
raise KeyError(key)
def _init_default_values(self):
for key, default in self.KEYS.items():
if key not in self:
self[key] = default
def __contains__(self, key):
self._check_key(key)
query = self.contains_query('key')
@ -133,9 +144,6 @@ class WebHistory(sql.SqlTable):
completion: A CompletionHistory instance.
metainfo: A CompletionMetaInfo instance.
_progress: A HistoryProgress instance.
Class attributes:
_PROGRESS_THRESHOLD: When to start showing progress dialogs.
"""
# All web history cleared
@ -143,8 +151,6 @@ class WebHistory(sql.SqlTable):
# one url cleared
url_cleared = pyqtSignal(QUrl)
_PROGRESS_THRESHOLD = 1000
def __init__(self, progress, parent=None):
super().__init__("History", ['url', 'title', 'atime', 'redirect'],
constraints={'url': 'NOT NULL',
@ -159,8 +165,18 @@ class WebHistory(sql.SqlTable):
self.completion = CompletionHistory(parent=self)
self.metainfo = CompletionMetaInfo(parent=self)
if sql.Query('pragma user_version').run().value() < _USER_VERSION:
self.completion.delete_all()
rebuild_completion = False
if sql.user_version_changed():
# If the DB user version changed, run a full cleanup and rebuild the
# completion history.
#
# In the future, this could be improved to only be done when actually needed
# - but version changes happen very infrequently, rebuilding everything
# gives us less corner-cases to deal with, and we can run a VACUUM to make
# things smaller.
self._cleanup_history()
rebuild_completion = True
# Get a string of all patterns
patterns = config.instance.get_str('completion.web_history.exclude')
@ -168,10 +184,11 @@ class WebHistory(sql.SqlTable):
# If patterns changed, update them in database and rebuild completion
if self.metainfo['excluded_patterns'] != patterns:
self.metainfo['excluded_patterns'] = patterns
self.completion.delete_all()
rebuild_completion = True
if not self.completion:
# either the table is out-of-date or the user wiped it manually
if rebuild_completion and self.completion:
# If no completion history exists, we don't need to spawn a dialog for
# cleaning it up.
self._rebuild_completion()
self.create_index('HistoryIndex', 'url')
@ -202,42 +219,92 @@ class WebHistory(sql.SqlTable):
try:
yield
except sql.KnownError as e:
message.error("Failed to write history: {}".format(e.text()))
message.error(f"Failed to write history: {e.text()}")
def _is_excluded(self, url):
def _is_excluded_from_completion(self, url):
"""Check if the given URL is excluded from the completion."""
patterns = config.cache['completion.web_history.exclude']
return any(pattern.matches(url) for pattern in patterns)
def _is_excluded_entirely(self, url):
"""Check if the given URL is excluded from the entire history.
This is the case for URLs which can't be visited at a later point; or which are
usually excessively long.
NOTE: If you add new filters here, it might be a good idea to adjust the
_USER_VERSION code and _cleanup_history so that older histories get cleaned up
accordingly as well.
"""
return (
url.scheme() in {'data', 'view-source'} or
(url.scheme() == 'qute' and url.host() in {'back', 'pdfjs'})
)
def _cleanup_history(self):
"""Do a one-time cleanup of the entire history.
This is run only once after the v2.0.0 upgrade, based on the database's
user_version.
"""
terms = [
'data:%',
'view-source:%',
'qute://back%',
'qute://pdfjs%',
]
where_clause = ' OR '.join(f"url LIKE '{term}'" for term in terms)
q = sql.Query(f'DELETE FROM History WHERE {where_clause}')
entries = q.run()
log.sql.debug(f"Cleanup removed {entries.rows_affected()} items")
def _rebuild_completion(self):
data: Mapping[str, MutableSequence[str]] = {
'url': [],
'title': [],
'last_atime': []
}
# select the latest entry for each url
q = sql.Query('SELECT url, title, max(atime) AS atime FROM History '
'WHERE NOT redirect and url NOT LIKE "qute://back%" '
'GROUP BY url ORDER BY atime asc')
entries = list(q.run())
if len(entries) > self._PROGRESS_THRESHOLD:
self._progress.start("Rebuilding completion...", len(entries))
self._progress.start(
"<b>Rebuilding completion...</b><br>"
"This is a one-time operation and happens because the database version "
"or <i>completion.web_history.exclude</i> was changed."
)
# Delete old entries
self.completion.delete_all()
QApplication.processEvents()
# Select the latest entry for each url
q = sql.Query('SELECT url, title, max(atime) AS atime FROM History '
'WHERE NOT redirect '
'GROUP BY url ORDER BY atime asc')
result = q.run()
QApplication.processEvents()
entries = list(result)
self._progress.set_maximum(len(entries))
for entry in entries:
self._progress.tick()
url = QUrl(entry.url)
if self._is_excluded(url):
if self._is_excluded_from_completion(url):
continue
data['url'].append(self._format_completion_url(url))
data['title'].append(entry.title)
data['last_atime'].append(entry.atime)
self._progress.finish()
self._progress.set_maximum(0)
# We might have caused fragmentation - let's clean up.
sql.Query('VACUUM').run()
QApplication.processEvents()
self.completion.insert_batch(data, replace=True)
sql.Query('pragma user_version = {}'.format(_USER_VERSION)).run()
QApplication.processEvents()
self._progress.finish()
def get_recent(self):
"""Get the most recent history entries."""
@ -289,9 +356,7 @@ class WebHistory(sql.SqlTable):
@pyqtSlot(QUrl, QUrl, str)
def add_from_tab(self, url, requested_url, title):
"""Add a new history entry as slot, called from a BrowserTab."""
if any(url.scheme() in ('data', 'view-source') or
(url.scheme(), url.host()) == ('qute', 'back')
for url in (url, requested_url)):
if self._is_excluded_entirely(url) or self._is_excluded_entirely(requested_url):
return
if url.isEmpty():
# things set via setHtml
@ -331,7 +396,7 @@ class WebHistory(sql.SqlTable):
'atime': atime,
'redirect': redirect})
if redirect or self._is_excluded(url):
if redirect or self._is_excluded_from_completion(url):
return
self.completion.insert({
@ -374,20 +439,15 @@ def debug_dump_history(dest):
"""
dest = os.path.expanduser(dest)
lines = ('{}{} {} {}'.format(int(x.atime),
'-r' * x.redirect,
x.url,
x.title)
for x in web_history.select(sort_by='atime',
sort_order='asc'))
lines = (f'{int(x.atime)}{"-r" * x.redirect} {x.url} {x.title}'
for x in web_history.select(sort_by='atime', sort_order='asc'))
try:
with open(dest, 'w', encoding='utf-8') as f:
f.write('\n'.join(lines))
message.info("Dumped history to {}".format(dest))
message.info(f"Dumped history to {dest}")
except OSError as e:
raise cmdutils.CommandError('Could not write history: {}'
.format(e))
raise cmdutils.CommandError(f'Could not write history: {e}')
def init(parent=None):

128
qutebrowser/misc/sql.py
View File

@ -21,12 +21,59 @@
import collections
import attr
from PyQt5.QtCore import QObject, pyqtSignal
from PyQt5.QtSql import QSqlDatabase, QSqlQuery, QSqlError
from qutebrowser.utils import log, debug
@attr.s
class UserVersion:
"""The version of data stored in the history database.
When we originally started using user_version, we only used it to signify that the
completion database should be regenerated. However, sometimes there are
backwards-incompatible changes.
Instead, we now (ab)use the fact that the user_version in sqlite is a 32-bit integer
to store both a major and a minor part. If only the minor part changed, we can deal
with it (there are only new URLs to clean up or somesuch). If the major part
changed, there are backwards-incompatible changes in how the database works, so
newer databases are not compatible with older qutebrowser versions.
"""
major: int = attr.ib()
minor: int = attr.ib()
@classmethod
def from_int(cls, num):
"""Parse a number from sqlite into a major/minor user version."""
assert 0 <= num <= 0x7FFF_FFFF, num # signed integer, but shouldn't be negative
major = (num & 0x7FFF_0000) >> 16
minor = num & 0x0000_FFFF
return cls(major, minor)
def to_int(self):
"""Get a sqlite integer from a major/minor user version."""
assert 0 <= self.major <= 0x7FFF # signed integer
assert 0 <= self.minor <= 0xFFFF
return self.major << 16 | self.minor
def __str__(self):
return f'{self.major}.{self.minor}'
_db_user_version = None # The user version we got from the database
_USER_VERSION = UserVersion(0, 3) # The current / newest user version
def user_version_changed():
"""Whether the version stored in the database is different from the current one."""
return _db_user_version != _USER_VERSION
class SqliteErrorCode:
"""Error codes as used by sqlite.
@ -134,10 +181,29 @@ def init(db_path):
error.text())
raise_sqlite_error(msg, error)
# Enable write-ahead-logging and reduce disk write frequency
# see https://sqlite.org/pragma.html and issues #2930 and #3507
Query("PRAGMA journal_mode=WAL").run()
Query("PRAGMA synchronous=NORMAL").run()
global _db_user_version
version_int = Query('pragma user_version').run().value()
_db_user_version = UserVersion.from_int(version_int)
if _db_user_version.major > _USER_VERSION.major:
raise KnownError(
"Database is too new for this qutebrowser version (database version "
f"{_db_user_version}, but {_USER_VERSION.major}.x is supported)")
if user_version_changed():
log.sql.debug(f"Migrating from version {_db_user_version} to {_USER_VERSION}")
# Note we're *not* updating the _db_user_version global here. We still want
# user_version_changed() to return True, as other modules (such as history.py)
# use it to create the initial table structure.
Query(f'PRAGMA user_version = {_USER_VERSION.to_int()}').run()
# Enable write-ahead-logging and reduce disk write frequency
# see https://sqlite.org/pragma.html and issues #2930 and #3507
#
# We might already have done this (without a migration) in earlier versions, but
# as those are idempotent, let's make sure we run them once again.
Query("PRAGMA journal_mode=WAL").run()
Query("PRAGMA synchronous=NORMAL").run()
def close():
@ -172,7 +238,7 @@ class Query:
"""
self.query = QSqlQuery(QSqlDatabase.database())
log.sql.debug('Preparing SQL query: "{}"'.format(querystr))
log.sql.vdebug(f'Preparing: {querystr}') # type: ignore[attr-defined]
ok = self.query.prepare(querystr)
self._check_ok('prepare', ok)
self.query.setForwardOnly(forward_only)
@ -200,16 +266,20 @@ class Query:
def _bind_values(self, values):
for key, val in values.items():
self.query.bindValue(':{}'.format(key), val)
if any(val is None for val in self.bound_values().values()):
bound_values = self.bound_values()
if None in bound_values.values():
raise BugError("Missing bound values!")
return bound_values
def run(self, **values):
"""Execute the prepared query."""
log.sql.debug('Running SQL query: "{}"'.format(
self.query.lastQuery()))
log.sql.debug(self.query.lastQuery())
self._bind_values(values)
log.sql.debug('query bindings: {}'.format(self.bound_values()))
bound_values = self._bind_values(values)
if bound_values:
log.sql.debug(f' {bound_values}')
ok = self.query.exec()
self._check_ok('exec', ok)
@ -245,7 +315,12 @@ class Query:
return self.query.record().value(0)
def rows_affected(self):
return self.query.numRowsAffected()
"""Return how many rows were affected by a non-SELECT query."""
assert not self.query.isSelect(), self
assert self.query.isActive(), self
rows = self.query.numRowsAffected()
assert rows != -1
return rows
def bound_values(self):
return self.query.boundValues()
@ -265,9 +340,7 @@ class SqlTable(QObject):
changed = pyqtSignal()
def __init__(self, name, fields, constraints=None, parent=None):
"""Create a new table in the SQL database.
Does nothing if the table already exists.
"""Wrapper over a table in the SQL database.
Args:
name: Name of the table.
@ -276,22 +349,34 @@ class SqlTable(QObject):
"""
super().__init__(parent)
self._name = name
self._create_table(fields, constraints)
def _create_table(self, fields, constraints):
"""Create the table if the database is uninitialized.
If the table already exists, this does nothing, so it can e.g. be called on
every user_version change.
"""
if not user_version_changed():
return
constraints = constraints or {}
column_defs = ['{} {}'.format(field, constraints.get(field, ''))
for field in fields]
q = Query("CREATE TABLE IF NOT EXISTS {name} ({column_defs})"
.format(name=name, column_defs=', '.join(column_defs)))
.format(name=self._name, column_defs=', '.join(column_defs)))
q.run()
def create_index(self, name, field):
"""Create an index over this table.
"""Create an index over this table if the database is uninitialized.
Args:
name: Name of the index, should be unique.
field: Name of the field to index.
"""
if not user_version_changed():
return
q = Query("CREATE INDEX IF NOT EXISTS {name} ON {table} ({field})"
.format(name=name, table=self._name, field=field))
q.run()
@ -318,6 +403,12 @@ class SqlTable(QObject):
q.run()
return q.value()
def __bool__(self):
"""Check whether there's any data in the table."""
q = Query(f"SELECT 1 FROM {self._name} LIMIT 1")
q.run()
return q.query.next()
def delete(self, field, value, *, optional=False):
"""Remove all rows for which `field` equals `value`.
@ -329,8 +420,7 @@ class SqlTable(QObject):
Return:
The number of rows deleted.
"""
q = Query("DELETE FROM {table} where {field} = :val"
.format(table=self._name, field=field))
q = Query(f"DELETE FROM {self._name} where {field} = :val")
q.run(val=value)
if not q.rows_affected():
if optional:

15
tests/end2end/features/history.feature
View File

@ -57,21 +57,6 @@ Feature: Page history
And I run :click-element id open-invalid
Then "load status for * LoadStatus.success" should be logged
Scenario: History with data URL
When I open data/data_link.html
And I run :click-element id link
And I wait until data:;base64,cXV0ZWJyb3dzZXI= is loaded
Then the history should contain:
http://localhost:(port)/data/data_link.html data: link
@qtwebkit_skip
Scenario: History with view-source URL
When I open data/title.html
And I run :view-source
And I wait for regex "Changing title for idx \d+ to 'view-source:(http://)?localhost:\d+/data/title.html'" in the log
Then the history should contain:
http://localhost:(port)/data/title.html Test title
Scenario: Clearing history
When I run :tab-only
And I open data/title.html

5
tests/helpers/stubs.py
View File

@ -628,9 +628,12 @@ class FakeHistoryProgress:
self._finished = False
self._value = 0
def start(self, _text, _maximum):
def start(self, _text):
self._started = True
def set_maximum(self, _maximum):
pass
def tick(self):
self._value += 1

76
tests/unit/browser/test_history.py
View File

@ -210,9 +210,20 @@ class TestAdd:
(logging.DEBUG, 'a.com', 'b.com', [('a.com', 'title', 12345, False),
('b.com', 'title', 12345, True)]),
(logging.WARNING, 'a.com', '', [('a.com', 'title', 12345, False)]),
(logging.WARNING, '', '', []),
(logging.WARNING, 'data:foo', '', []),
(logging.WARNING, 'a.com', 'data:foo', []),
(logging.WARNING, 'view-source:foo', '', []),
(logging.WARNING, 'a.com', 'view-source:foo', []),
(logging.WARNING, 'qute://back', '', []),
(logging.WARNING, 'a.com', 'qute://back', []),
(logging.WARNING, 'qute://pdfjs/', '', []),
(logging.WARNING, 'a.com', 'qute://pdfjs/', []),
])
def test_from_tab(self, web_history, caplog, mock_time,
level, url, req_url, expected):
@ -357,33 +368,12 @@ class TestDump:
class TestRebuild:
def test_delete(self, web_history, stubs):
web_history.insert({'url': 'example.com/1', 'title': 'example1',
'redirect': False, 'atime': 1})
web_history.insert({'url': 'example.com/1', 'title': 'example1',
'redirect': False, 'atime': 2})
web_history.insert({'url': 'example.com/2%203', 'title': 'example2',
'redirect': False, 'atime': 3})
web_history.insert({'url': 'example.com/3', 'title': 'example3',
'redirect': True, 'atime': 4})
web_history.insert({'url': 'example.com/2 3', 'title': 'example2',
'redirect': False, 'atime': 5})
web_history.completion.delete_all()
hist2 = history.WebHistory(progress=stubs.FakeHistoryProgress())
assert list(hist2.completion) == [
('example.com/1', 'example1', 2),
('example.com/2 3', 'example2', 5),
]
def test_no_rebuild(self, web_history, stubs):
"""Ensure that completion is not regenerated unless empty."""
web_history.add_url(QUrl('example.com/1'), redirect=False, atime=1)
web_history.add_url(QUrl('example.com/2'), redirect=False, atime=2)
web_history.completion.delete('url', 'example.com/2')
hist2 = history.WebHistory(progress=stubs.FakeHistoryProgress())
assert list(hist2.completion) == [('example.com/1', '', 1)]
# FIXME: Some of those tests might be a bit misleading, as creating a new
# history.WebHistory will regenerate the completion either way with the SQL changes
# in v2.0.0 (because the user version changed from 0 -> 3).
#
# They should be revisited once we can actually create two independent sqlite
# databases and copy the data over, for a "real" test.
def test_user_version(self, web_history, stubs, monkeypatch):
"""Ensure that completion is regenerated if user_version changes."""
@ -391,11 +381,12 @@ class TestRebuild:
web_history.add_url(QUrl('example.com/2'), redirect=False, atime=2)
web_history.completion.delete('url', 'example.com/2')
hist2 = history.WebHistory(progress=stubs.FakeHistoryProgress())
assert list(hist2.completion) == [('example.com/1', '', 1)]
# User version always changes, so this won't work
# hist2 = history.WebHistory(progress=stubs.FakeHistoryProgress())
# assert list(hist2.completion) == [('example.com/1', '', 1)]
monkeypatch.setattr(sql, 'user_version_changed', lambda: True)
monkeypatch.setattr(history, '_USER_VERSION',
history._USER_VERSION + 1)
hist3 = history.WebHistory(progress=stubs.FakeHistoryProgress())
assert list(hist3.completion) == [
('example.com/1', '', 1),
@ -439,22 +430,18 @@ class TestRebuild:
('http://example.org', '', 2)
]
@pytest.mark.parametrize('patch_threshold', [True, False])
def test_progress(self, web_history, config_stub, monkeypatch, stubs,
patch_threshold):
def test_progress(self, monkeypatch, web_history, config_stub, stubs):
web_history.add_url(QUrl('example.com/1'), redirect=False, atime=1)
web_history.add_url(QUrl('example.com/2'), redirect=False, atime=2)
# Change cached patterns to trigger a completion rebuild
web_history.metainfo['excluded_patterns'] = 'http://example.org'
if patch_threshold:
monkeypatch.setattr(history.WebHistory, '_PROGRESS_THRESHOLD', 1)
# Trigger a completion rebuild
monkeypatch.setattr(sql, 'user_version_changed', lambda: True)
progress = stubs.FakeHistoryProgress()
history.WebHistory(progress=progress)
assert progress._value == 2
assert progress._started
assert progress._finished
assert progress._started == patch_threshold
class TestCompletionMetaInfo:
@ -501,12 +488,12 @@ class TestHistoryProgress:
def test_no_start(self, progress):
"""Test calling tick/finish without start."""
progress.tick()
assert progress._value == 1
progress.finish()
assert progress._progress is None
assert progress._value == 1
def test_gui(self, qtbot, progress):
progress.start("Hello World", 42)
progress.start("Hello World")
dialog = progress._progress
qtbot.add_widget(dialog)
progress.tick()
@ -514,9 +501,12 @@ class TestHistoryProgress:
assert dialog.isVisible()
assert dialog.labelText() == "Hello World"
assert dialog.minimum() == 0
assert dialog.maximum() == 42
assert dialog.value() == 1
assert dialog.minimumDuration() == 500
assert dialog.minimumDuration() == 0
assert dialog.maximum() == 0
progress.set_maximum(42)
assert dialog.maximum() == 42
progress.finish()
assert not dialog.isVisible()

91
tests/unit/misc/test_sql.py
View File

@ -21,6 +21,8 @@
import pytest
import hypothesis
from hypothesis import strategies
from PyQt5.QtSql import QSqlError
from qutebrowser.misc import sql
@ -29,6 +31,55 @@ from qutebrowser.misc import sql
pytestmark = pytest.mark.usefixtures('init_sql')
class TestUserVersion:
@pytest.mark.parametrize('val, major, minor', [
(0x0008_0001, 8, 1),
(0x7FFF_FFFF, 0x7FFF, 0xFFFF),
])
def test_from_int(self, val, major, minor):
version = sql.UserVersion.from_int(val)
assert version.major == major
assert version.minor == minor
@pytest.mark.parametrize('major, minor, val', [
(8, 1, 0x0008_0001),
(0x7FFF, 0xFFFF, 0x7FFF_FFFF),
])
def test_to_int(self, major, minor, val):
version = sql.UserVersion(major, minor)
assert version.to_int() == val
@pytest.mark.parametrize('val', [0x8000_0000, -1])
def test_from_int_invalid(self, val):
with pytest.raises(AssertionError):
sql.UserVersion.from_int(val)
@pytest.mark.parametrize('major, minor', [
(-1, 0),
(0, -1),
(0, 0x10000),
(0x8000, 0),
])
def test_to_int_invalid(self, major, minor):
version = sql.UserVersion(major, minor)
with pytest.raises(AssertionError):
version.to_int()
@hypothesis.given(val=strategies.integers(min_value=0, max_value=0x7FFF_FFFF))
def test_from_int_hypothesis(self, val):
version = sql.UserVersion.from_int(val)
assert version.to_int() == val
@hypothesis.given(
major=strategies.integers(min_value=0, max_value=0x7FFF),
minor=strategies.integers(min_value=0, max_value=0xFFFF)
)
def test_to_int_hypothesis(self, major, minor):
version = sql.UserVersion(major, minor)
assert version.from_int(version.to_int()) == version
@pytest.mark.parametrize('klass', [sql.KnownError, sql.BugError])
def test_sqlerror(klass):
text = "Hello World"
@ -192,6 +243,26 @@ def test_len():
assert len(table) == 3
def test_bool():
table = sql.SqlTable('Foo', ['name'])
assert not table
table.insert({'name': 'one'})
assert table
def test_bool_benchmark(benchmark):
table = sql.SqlTable('Foo', ['number'])
# Simulate a history table
table.create_index('NumberIndex', 'number')
table.insert_batch({'number': [str(i) for i in range(100_000)]})
def run():
assert table
benchmark(run)
def test_contains():
table = sql.SqlTable('Foo', ['name', 'val', 'lucky'])
table.insert({'name': 'one', 'val': 1, 'lucky': False})
@ -293,10 +364,24 @@ class TestSqlQuery:
match='No result for single-result query'):
q.value()
def test_num_rows_affected(self):
q = sql.Query('SELECT 0')
def test_num_rows_affected_not_active(self):
with pytest.raises(AssertionError):
q = sql.Query('SELECT 0')
q.rows_affected()
def test_num_rows_affected_select(self):
with pytest.raises(AssertionError):
q = sql.Query('SELECT 0')
q.run()
q.rows_affected()
@pytest.mark.parametrize('condition', [0, 1])
def test_num_rows_affected(self, condition):
table = sql.SqlTable('Foo', ['name'])
table.insert({'name': 'helloworld'})
q = sql.Query(f'DELETE FROM Foo WHERE {condition}')
q.run()
assert q.rows_affected() == 0
assert q.rows_affected() == condition
def test_bound_values(self):
q = sql.Query('SELECT :answer')