Merge branch 'blacklist-history'
This commit is contained in:
commit
2fcdc5a0c9
@ -25,6 +25,8 @@ Added
|
||||
- When `:spawn --userscript` is called with a count, that count is now
|
||||
passed to userscripts as `$QUTE_COUNT`.
|
||||
- New `content.mouse_lock` setting to handle HTML5 pointer locking.
|
||||
- New `completion.web_history.exclude` setting which hides a list of URL
|
||||
patterns from the completion.
|
||||
|
||||
Changed
|
||||
~~~~~~~
|
||||
@ -39,6 +41,8 @@ Changed
|
||||
on macOS.
|
||||
- Using `:set option` now shows the value of the setting (like `:set option?`
|
||||
already did).
|
||||
- The `completion.web_history_max_items` setting got renamed to
|
||||
`completion.web_history.max_items`.
|
||||
|
||||
v1.4.2
|
||||
------
|
||||
|
@ -453,7 +453,7 @@ or always navigate through command history with
|
||||
:bind -m command <Down> command-history-next
|
||||
----
|
||||
|
||||
- The default for `completion.web_history_max_items` is now set to `-1`, showing
|
||||
- The default for `completion.web_history.max_items` is now set to `-1`, showing
|
||||
an unlimited number of items in the completion for `:open` as the new
|
||||
sqlite-based completion is much faster. If the `:open` completion is too slow
|
||||
on your machine, set an appropriate limit again.
|
||||
|
@ -107,7 +107,8 @@
|
||||
|<<completion.shrink,completion.shrink>>|Shrink the completion to be smaller than the configured size if there are no scrollbars.
|
||||
|<<completion.timestamp_format,completion.timestamp_format>>|Format of timestamps (e.g. for the history completion).
|
||||
|<<completion.use_best_match,completion.use_best_match>>|Execute the best-matching command on a partial match.
|
||||
|<<completion.web_history_max_items,completion.web_history_max_items>>|Number of URLs to show in the web history.
|
||||
|<<completion.web_history.exclude,completion.web_history.exclude>>|A list of patterns which should not be shown in the history.
|
||||
|<<completion.web_history.max_items,completion.web_history.max_items>>|Number of URLs to show in the web history.
|
||||
|<<confirm_quit,confirm_quit>>|Require a confirmation before quitting the application.
|
||||
|<<content.autoplay,content.autoplay>>|Automatically start playing `<video>` elements.
|
||||
|<<content.cache.appcache,content.cache.appcache>>|Enable support for the HTML 5 web application cache feature.
|
||||
@ -1447,8 +1448,19 @@ Type: <<types,Bool>>
|
||||
|
||||
Default: +pass:[false]+
|
||||
|
||||
[[completion.web_history_max_items]]
|
||||
=== completion.web_history_max_items
|
||||
[[completion.web_history.exclude]]
|
||||
=== completion.web_history.exclude
|
||||
A list of patterns which should not be shown in the history.
|
||||
This only affects the completion. Matching URLs are still saved in the history (and visible on the qute://history page), but hidden in the completion.
|
||||
Changing this setting will cause the completion history to be regenerated on the next start, which will take a short while.
|
||||
This setting requires a restart.
|
||||
|
||||
Type: <<types,List of UrlPattern>>
|
||||
|
||||
Default: empty
|
||||
|
||||
[[completion.web_history.max_items]]
|
||||
=== completion.web_history.max_items
|
||||
Number of URLs to show in the web history.
|
||||
0: no history / -1: unlimited
|
||||
|
||||
@ -3449,5 +3461,8 @@ See the setting's valid values for more information on allowed values.
|
||||
See https://sqlite.org/lang_datefunc.html for reference.
|
||||
|UniqueCharString|A string which may not contain duplicate chars.
|
||||
|Url|A URL as a string.
|
||||
|UrlPattern|A match pattern for a URL.
|
||||
|
||||
See https://developer.chrome.com/apps/match_patterns for the allowed syntax.
|
||||
|VerticalPosition|The position of the download bar.
|
||||
|==============
|
||||
|
@ -449,13 +449,10 @@ def _init_modules(args, crash_handler):
|
||||
|
||||
log.init.debug("Initializing web history...")
|
||||
history.init(qApp)
|
||||
except sql.SqlError as e:
|
||||
if e.environmental:
|
||||
error.handle_fatal_exc(e, args, 'Error initializing SQL',
|
||||
pre_text='Error initializing SQL')
|
||||
sys.exit(usertypes.Exit.err_init)
|
||||
else:
|
||||
raise
|
||||
except sql.SqlEnvironmentError as e:
|
||||
error.handle_fatal_exc(e, args, 'Error initializing SQL',
|
||||
pre_text='Error initializing SQL')
|
||||
sys.exit(usertypes.Exit.err_init)
|
||||
|
||||
log.init.debug("Initializing completion...")
|
||||
completiondelegate.init()
|
||||
|
@ -25,6 +25,7 @@ import contextlib
|
||||
|
||||
from PyQt5.QtCore import pyqtSlot, QUrl, QTimer, pyqtSignal
|
||||
|
||||
from qutebrowser.config import config
|
||||
from qutebrowser.commands import cmdutils, cmdexc
|
||||
from qutebrowser.utils import (utils, objreg, log, usertypes, message,
|
||||
debug, standarddir, qtutils)
|
||||
@ -35,6 +36,41 @@ from qutebrowser.misc import objects, sql
|
||||
_USER_VERSION = 2
|
||||
|
||||
|
||||
class CompletionMetaInfo(sql.SqlTable):
|
||||
|
||||
"""Table containing meta-information for the completion."""
|
||||
|
||||
KEYS = {
|
||||
'force_rebuild': False,
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
def _check_key(self, key):
|
||||
if key not in self.KEYS:
|
||||
raise KeyError(key)
|
||||
|
||||
def __contains__(self, key):
|
||||
self._check_key(key)
|
||||
query = self.contains_query('key')
|
||||
return query.run(val=key).value()
|
||||
|
||||
def __getitem__(self, key):
|
||||
self._check_key(key)
|
||||
query = sql.Query('SELECT value FROM CompletionMetaInfo '
|
||||
'WHERE key = :key')
|
||||
return query.run(key=key).value()
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
self._check_key(key)
|
||||
self.insert({'key': key, 'value': value}, replace=True)
|
||||
|
||||
|
||||
class CompletionHistory(sql.SqlTable):
|
||||
|
||||
"""History which only has the newest entry for each URL."""
|
||||
@ -65,11 +101,18 @@ class WebHistory(sql.SqlTable):
|
||||
'redirect': 'NOT NULL'},
|
||||
parent=parent)
|
||||
self.completion = CompletionHistory(parent=self)
|
||||
self.metainfo = CompletionMetaInfo(parent=self)
|
||||
|
||||
if sql.Query('pragma user_version').run().value() < _USER_VERSION:
|
||||
self.completion.delete_all()
|
||||
if self.metainfo['force_rebuild']:
|
||||
self.completion.delete_all()
|
||||
self.metainfo['force_rebuild'] = False
|
||||
|
||||
if not self.completion:
|
||||
# either the table is out-of-date or the user wiped it manually
|
||||
self._rebuild_completion()
|
||||
|
||||
self.create_index('HistoryIndex', 'url')
|
||||
self.create_index('HistoryAtimeIndex', 'atime')
|
||||
self._contains_query = self.contains_query('url')
|
||||
@ -87,21 +130,29 @@ class WebHistory(sql.SqlTable):
|
||||
'ORDER BY atime desc '
|
||||
'limit :limit offset :offset')
|
||||
|
||||
config.instance.changed.connect(self._on_config_changed)
|
||||
|
||||
def __repr__(self):
|
||||
return utils.get_repr(self, length=len(self))
|
||||
|
||||
def __contains__(self, url):
|
||||
return self._contains_query.run(val=url).value()
|
||||
|
||||
@config.change_filter('completion.web_history.exclude')
|
||||
def _on_config_changed(self):
|
||||
self.metainfo['force_rebuild'] = True
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _handle_sql_errors(self):
|
||||
try:
|
||||
yield
|
||||
except sql.SqlError as e:
|
||||
if e.environmental:
|
||||
message.error("Failed to write history: {}".format(e.text()))
|
||||
else:
|
||||
raise
|
||||
except sql.SqlEnvironmentError as e:
|
||||
message.error("Failed to write history: {}".format(e.text()))
|
||||
|
||||
def _is_excluded(self, url):
|
||||
"""Check if the given URL is excluded from the completion."""
|
||||
return any(pattern.matches(url)
|
||||
for pattern in config.val.completion.web_history.exclude)
|
||||
|
||||
def _rebuild_completion(self):
|
||||
data = {'url': [], 'title': [], 'last_atime': []}
|
||||
@ -110,9 +161,13 @@ class WebHistory(sql.SqlTable):
|
||||
'WHERE NOT redirect and url NOT LIKE "qute://back%" '
|
||||
'GROUP BY url ORDER BY atime asc')
|
||||
for entry in q.run():
|
||||
data['url'].append(self._format_completion_url(QUrl(entry.url)))
|
||||
url = QUrl(entry.url)
|
||||
if self._is_excluded(url):
|
||||
continue
|
||||
data['url'].append(self._format_completion_url(url))
|
||||
data['title'].append(entry.title)
|
||||
data['last_atime'].append(entry.atime)
|
||||
|
||||
self.completion.insert_batch(data, replace=True)
|
||||
sql.Query('pragma user_version = {}'.format(_USER_VERSION)).run()
|
||||
|
||||
@ -218,12 +273,15 @@ class WebHistory(sql.SqlTable):
|
||||
'title': title,
|
||||
'atime': atime,
|
||||
'redirect': redirect})
|
||||
if not redirect:
|
||||
self.completion.insert({
|
||||
'url': self._format_completion_url(url),
|
||||
'title': title,
|
||||
'last_atime': atime
|
||||
}, replace=True)
|
||||
|
||||
if redirect or self._is_excluded(url):
|
||||
return
|
||||
|
||||
self.completion.insert({
|
||||
'url': self._format_completion_url(url),
|
||||
'title': title,
|
||||
'last_atime': atime
|
||||
}, replace=True)
|
||||
|
||||
def _parse_entry(self, line):
|
||||
"""Parse a history line like '12345 http://example.com title'."""
|
||||
|
@ -42,7 +42,7 @@ class HistoryCategory(QSqlQueryModel):
|
||||
|
||||
def _atime_expr(self):
|
||||
"""If max_items is set, return an expression to limit the query."""
|
||||
max_items = config.val.completion.web_history_max_items
|
||||
max_items = config.val.completion.web_history.max_items
|
||||
# HistoryCategory should not be added to the completion in that case.
|
||||
assert max_items != 0
|
||||
|
||||
@ -84,7 +84,7 @@ class HistoryCategory(QSqlQueryModel):
|
||||
timefmt = ("strftime('{}', last_atime, 'unixepoch', 'localtime')"
|
||||
.format(timestamp_format.replace("'", "`")))
|
||||
|
||||
if not self._query or len(words) != len(self._query.boundValues()):
|
||||
if not self._query or len(words) != len(self._query.bound_values()):
|
||||
# if the number of words changed, we need to generate a new query
|
||||
# otherwise, we can reuse the prepared query for performance
|
||||
self._query = sql.Query(' '.join([
|
||||
@ -100,14 +100,14 @@ class HistoryCategory(QSqlQueryModel):
|
||||
with debug.log_time('sql', 'Running completion query'):
|
||||
self._query.run(**{
|
||||
str(i): w for i, w in enumerate(words)})
|
||||
self.setQuery(self._query)
|
||||
self.setQuery(self._query.query)
|
||||
|
||||
def removeRows(self, row, _count, _parent=None):
|
||||
"""Override QAbstractItemModel::removeRows to re-run sql query."""
|
||||
# re-run query to reload updated table
|
||||
with debug.log_time('sql', 'Re-running completion query post-delete'):
|
||||
self._query.run()
|
||||
self.setQuery(self._query)
|
||||
self.setQuery(self._query.query)
|
||||
while self.rowCount() < row:
|
||||
self.fetchMore()
|
||||
return True
|
||||
|
@ -68,7 +68,7 @@ def url(*, info):
|
||||
model.add_category(listcategory.ListCategory(
|
||||
'Bookmarks', bookmarks, delete_func=_delete_bookmark, sort=False))
|
||||
|
||||
if info.config.get('completion.web_history_max_items') != 0:
|
||||
if info.config.get('completion.web_history.max_items') != 0:
|
||||
hist_cat = histcategory.HistoryCategory(delete_func=_delete_history)
|
||||
model.add_category(hist_cat)
|
||||
return model
|
||||
|
@ -812,7 +812,27 @@ completion.timestamp_format:
|
||||
default: '%Y-%m-%d'
|
||||
desc: Format of timestamps (e.g. for the history completion).
|
||||
|
||||
completion.web_history.exclude:
|
||||
type:
|
||||
name: List
|
||||
valtype: UrlPattern
|
||||
none_ok: true
|
||||
default: []
|
||||
restart: true
|
||||
desc: >-
|
||||
A list of patterns which should not be shown in the history.
|
||||
|
||||
This only affects the completion. Matching URLs are still saved in the
|
||||
history (and visible on the qute://history page), but hidden in the
|
||||
completion.
|
||||
|
||||
Changing this setting will cause the completion history to be regenerated
|
||||
on the next start, which will take a short while.
|
||||
|
||||
completion.web_history_max_items:
|
||||
renamed: completion.web_history.max_items
|
||||
|
||||
completion.web_history.max_items:
|
||||
default: -1
|
||||
type:
|
||||
name: Int
|
||||
|
@ -61,7 +61,7 @@ from PyQt5.QtWidgets import QTabWidget, QTabBar
|
||||
|
||||
from qutebrowser.commands import cmdutils
|
||||
from qutebrowser.config import configexc
|
||||
from qutebrowser.utils import standarddir, utils, qtutils, urlutils
|
||||
from qutebrowser.utils import standarddir, utils, qtutils, urlutils, urlmatch
|
||||
from qutebrowser.keyinput import keyutils
|
||||
|
||||
|
||||
@ -1661,3 +1661,22 @@ class Key(BaseType):
|
||||
return keyutils.KeySequence.parse(value)
|
||||
except keyutils.KeyParseError as e:
|
||||
raise configexc.ValidationError(value, str(e))
|
||||
|
||||
|
||||
class UrlPattern(BaseType):
|
||||
|
||||
"""A match pattern for a URL.
|
||||
|
||||
See https://developer.chrome.com/apps/match_patterns for the allowed
|
||||
syntax.
|
||||
"""
|
||||
|
||||
def to_py(self, value):
|
||||
self._basic_py_validation(value, str)
|
||||
if not value:
|
||||
return None
|
||||
|
||||
try:
|
||||
return urlmatch.UrlPattern(value)
|
||||
except urlmatch.ParseError as e:
|
||||
raise configexc.ValidationError(value, str(e))
|
||||
|
@ -27,93 +27,107 @@ from PyQt5.QtSql import QSqlDatabase, QSqlQuery, QSqlError
|
||||
from qutebrowser.utils import log, debug
|
||||
|
||||
|
||||
class SqliteErrorCode:
|
||||
|
||||
"""Error codes as used by sqlite.
|
||||
|
||||
See https://sqlite.org/rescode.html - note we only define the codes we use
|
||||
in qutebrowser here.
|
||||
"""
|
||||
|
||||
BUSY = '5' # database is locked
|
||||
READONLY = '8' # attempt to write a readonly database
|
||||
IOERR = '10' # disk I/O error
|
||||
CORRUPT = '11' # database disk image is malformed
|
||||
FULL = '13' # database or disk is full
|
||||
CANTOPEN = '14' # unable to open database file
|
||||
CONSTRAINT = '19' # UNIQUE constraint failed
|
||||
|
||||
|
||||
class SqlError(Exception):
|
||||
|
||||
"""Raised on an error interacting with the SQL database.
|
||||
"""Base class for all SQL related errors."""
|
||||
|
||||
Attributes:
|
||||
environmental: Whether the error is likely caused by the environment
|
||||
and not a qutebrowser bug.
|
||||
"""
|
||||
|
||||
def __init__(self, msg, environmental=False):
|
||||
super().__init__(msg)
|
||||
self.environmental = environmental
|
||||
|
||||
def text(self):
|
||||
"""Get a short text to display."""
|
||||
return str(self)
|
||||
|
||||
|
||||
class SqliteError(SqlError):
|
||||
|
||||
"""A SQL error with a QSqlError available.
|
||||
|
||||
Attributes:
|
||||
error: The QSqlError object.
|
||||
"""
|
||||
|
||||
def __init__(self, msg, error):
|
||||
def __init__(self, msg, error=None):
|
||||
super().__init__(msg)
|
||||
self.error = error
|
||||
|
||||
log.sql.debug("SQL error:")
|
||||
log.sql.debug("type: {}".format(
|
||||
debug.qenum_key(QSqlError, error.type())))
|
||||
log.sql.debug("database text: {}".format(error.databaseText()))
|
||||
log.sql.debug("driver text: {}".format(error.driverText()))
|
||||
log.sql.debug("error code: {}".format(error.nativeErrorCode()))
|
||||
|
||||
# https://sqlite.org/rescode.html
|
||||
# https://github.com/qutebrowser/qutebrowser/issues/2930
|
||||
# https://github.com/qutebrowser/qutebrowser/issues/3004
|
||||
environmental_errors = [
|
||||
'5', # SQLITE_BUSY ("database is locked")
|
||||
'8', # SQLITE_READONLY
|
||||
'11', # SQLITE_CORRUPT
|
||||
'13', # SQLITE_FULL
|
||||
]
|
||||
# At least in init(), we can get errors like this:
|
||||
# > type: ConnectionError
|
||||
# > database text: out of memory
|
||||
# > driver text: Error opening database
|
||||
# > error code: -1
|
||||
environmental_strings = [
|
||||
"out of memory",
|
||||
]
|
||||
errcode = error.nativeErrorCode()
|
||||
self.environmental = (
|
||||
errcode in environmental_errors or
|
||||
(errcode == -1 and error.databaseText() in environmental_strings))
|
||||
|
||||
def text(self):
|
||||
return self.error.databaseText()
|
||||
"""Get a short text description of the error.
|
||||
|
||||
@classmethod
|
||||
def from_query(cls, what, query, error):
|
||||
"""Construct an error from a failed query.
|
||||
|
||||
Arguments:
|
||||
what: What we were doing when the error happened.
|
||||
query: The query which was executed.
|
||||
error: The QSqlError object.
|
||||
This is a string suitable to show to the user as error message.
|
||||
"""
|
||||
msg = 'Failed to {} query "{}": "{}"'.format(what, query, error.text())
|
||||
return cls(msg, error)
|
||||
if self.error is None:
|
||||
return str(self)
|
||||
else:
|
||||
return self.error.databaseText()
|
||||
|
||||
|
||||
class SqlEnvironmentError(SqlError):
|
||||
|
||||
"""Raised on an error interacting with the SQL database.
|
||||
|
||||
This is raised in conditions resulting from the environment (like a full
|
||||
disk or I/O errors), where qutebrowser isn't to blame.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class SqlBugError(SqlError):
|
||||
|
||||
"""Raised on an error interacting with the SQL database.
|
||||
|
||||
This is raised for errors resulting from a qutebrowser bug.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
def raise_sqlite_error(msg, error):
|
||||
"""Raise either a SqlBugError or SqlEnvironmentError."""
|
||||
log.sql.debug("SQL error:")
|
||||
log.sql.debug("type: {}".format(
|
||||
debug.qenum_key(QSqlError, error.type())))
|
||||
log.sql.debug("database text: {}".format(error.databaseText()))
|
||||
log.sql.debug("driver text: {}".format(error.driverText()))
|
||||
log.sql.debug("error code: {}".format(error.nativeErrorCode()))
|
||||
environmental_errors = [
|
||||
SqliteErrorCode.BUSY,
|
||||
SqliteErrorCode.READONLY,
|
||||
SqliteErrorCode.IOERR,
|
||||
SqliteErrorCode.CORRUPT,
|
||||
SqliteErrorCode.FULL,
|
||||
SqliteErrorCode.CANTOPEN,
|
||||
]
|
||||
# At least in init(), we can get errors like this:
|
||||
# > type: ConnectionError
|
||||
# > database text: out of memory
|
||||
# > driver text: Error opening database
|
||||
# > error code: -1
|
||||
environmental_strings = [
|
||||
"out of memory",
|
||||
]
|
||||
errcode = error.nativeErrorCode()
|
||||
if (errcode in environmental_errors or
|
||||
(errcode == -1 and error.databaseText() in environmental_strings)):
|
||||
raise SqlEnvironmentError(msg, error)
|
||||
else:
|
||||
raise SqlBugError(msg, error)
|
||||
|
||||
|
||||
def init(db_path):
|
||||
"""Initialize the SQL database connection."""
|
||||
database = QSqlDatabase.addDatabase('QSQLITE')
|
||||
if not database.isValid():
|
||||
raise SqlError('Failed to add database. '
|
||||
'Are sqlite and Qt sqlite support installed?',
|
||||
environmental=True)
|
||||
raise SqlEnvironmentError('Failed to add database. Are sqlite and Qt '
|
||||
'sqlite support installed?')
|
||||
database.setDatabaseName(db_path)
|
||||
if not database.open():
|
||||
error = database.lastError()
|
||||
raise SqliteError("Failed to open sqlite database at {}: {}"
|
||||
.format(db_path, error.text()), error)
|
||||
msg = "Failed to open sqlite database at {}: {}".format(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
|
||||
@ -135,11 +149,11 @@ def version():
|
||||
close()
|
||||
return ver
|
||||
return Query("select sqlite_version()").run().value()
|
||||
except SqlError as e:
|
||||
except SqlEnvironmentError as e:
|
||||
return 'UNAVAILABLE ({})'.format(e)
|
||||
|
||||
|
||||
class Query(QSqlQuery):
|
||||
class Query:
|
||||
|
||||
"""A prepared SQL Query."""
|
||||
|
||||
@ -151,39 +165,84 @@ class Query(QSqlQuery):
|
||||
forward_only: Optimization for queries that will only step forward.
|
||||
Must be false for completion queries.
|
||||
"""
|
||||
super().__init__(QSqlDatabase.database())
|
||||
self.query = QSqlQuery(QSqlDatabase.database())
|
||||
|
||||
log.sql.debug('Preparing SQL query: "{}"'.format(querystr))
|
||||
if not self.prepare(querystr):
|
||||
raise SqliteError.from_query('prepare', querystr, self.lastError())
|
||||
self.setForwardOnly(forward_only)
|
||||
ok = self.query.prepare(querystr)
|
||||
self._check_ok('prepare', ok)
|
||||
self.query.setForwardOnly(forward_only)
|
||||
|
||||
def __iter__(self):
|
||||
if not self.isActive():
|
||||
raise SqlError("Cannot iterate inactive query")
|
||||
rec = self.record()
|
||||
if not self.query.isActive():
|
||||
raise SqlBugError("Cannot iterate inactive query")
|
||||
rec = self.query.record()
|
||||
fields = [rec.fieldName(i) for i in range(rec.count())]
|
||||
rowtype = collections.namedtuple('ResultRow', fields)
|
||||
|
||||
while self.next():
|
||||
rec = self.record()
|
||||
while self.query.next():
|
||||
rec = self.query.record()
|
||||
yield rowtype(*[rec.value(i) for i in range(rec.count())])
|
||||
|
||||
def _check_ok(self, step, ok):
|
||||
if not ok:
|
||||
query = self.query.lastQuery()
|
||||
error = self.query.lastError()
|
||||
msg = 'Failed to {} query "{}": "{}"'.format(step, query,
|
||||
error.text())
|
||||
raise_sqlite_error(msg, error)
|
||||
|
||||
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()):
|
||||
raise SqlBugError("Missing bound values!")
|
||||
|
||||
def run(self, **values):
|
||||
"""Execute the prepared query."""
|
||||
log.sql.debug('Running SQL query: "{}"'.format(self.lastQuery()))
|
||||
for key, val in values.items():
|
||||
self.bindValue(':{}'.format(key), val)
|
||||
log.sql.debug('query bindings: {}'.format(self.boundValues()))
|
||||
if not self.exec_():
|
||||
raise SqliteError.from_query('exec', self.lastQuery(),
|
||||
self.lastError())
|
||||
log.sql.debug('Running SQL query: "{}"'.format(
|
||||
self.query.lastQuery()))
|
||||
|
||||
self._bind_values(values)
|
||||
log.sql.debug('query bindings: {}'.format(self.bound_values()))
|
||||
|
||||
ok = self.query.exec_()
|
||||
self._check_ok('exec', ok)
|
||||
|
||||
return self
|
||||
|
||||
def run_batch(self, values):
|
||||
"""Execute the query in batch mode."""
|
||||
log.sql.debug('Running SQL query (batch): "{}"'.format(
|
||||
self.query.lastQuery()))
|
||||
|
||||
self._bind_values(values)
|
||||
|
||||
db = QSqlDatabase.database()
|
||||
ok = db.transaction()
|
||||
self._check_ok('transaction', ok)
|
||||
|
||||
ok = self.query.execBatch()
|
||||
try:
|
||||
self._check_ok('execBatch', ok)
|
||||
except SqlError:
|
||||
# Not checking the return value here, as we're failing anyways...
|
||||
db.rollback()
|
||||
raise
|
||||
|
||||
ok = db.commit()
|
||||
self._check_ok('commit', ok)
|
||||
|
||||
def value(self):
|
||||
"""Return the result of a single-value query (e.g. an EXISTS)."""
|
||||
if not self.next():
|
||||
raise SqlError("No result for single-result query")
|
||||
return self.record().value(0)
|
||||
if not self.query.next():
|
||||
raise SqlBugError("No result for single-result query")
|
||||
return self.query.record().value(0)
|
||||
|
||||
def rows_affected(self):
|
||||
return self.query.numRowsAffected()
|
||||
|
||||
def bound_values(self):
|
||||
return self.query.boundValues()
|
||||
|
||||
|
||||
class SqlTable(QObject):
|
||||
@ -266,7 +325,7 @@ class SqlTable(QObject):
|
||||
q = Query("DELETE FROM {table} where {field} = :val"
|
||||
.format(table=self._name, field=field))
|
||||
q.run(val=value)
|
||||
if not q.numRowsAffected():
|
||||
if not q.rows_affected():
|
||||
raise KeyError('No row with {} = "{}"'.format(field, value))
|
||||
self.changed.emit()
|
||||
|
||||
@ -296,14 +355,7 @@ class SqlTable(QObject):
|
||||
replace: If true, overwrite rows with a primary key match.
|
||||
"""
|
||||
q = self._insert_query(values, replace)
|
||||
for key, val in values.items():
|
||||
q.bindValue(':{}'.format(key), val)
|
||||
|
||||
db = QSqlDatabase.database()
|
||||
db.transaction()
|
||||
if not q.execBatch():
|
||||
raise SqliteError.from_query('exec', q.lastQuery(), q.lastError())
|
||||
db.commit()
|
||||
q.run_batch(values)
|
||||
self.changed.emit()
|
||||
|
||||
def delete_all(self):
|
||||
|
@ -64,6 +64,7 @@ def whitelist_generator(): # noqa
|
||||
yield 'qutebrowser.utils.debug.qflags_key'
|
||||
yield 'qutebrowser.utils.qtutils.QtOSError.qt_errno'
|
||||
yield 'scripts.utils.bg_colors'
|
||||
yield 'qutebrowser.misc.sql.SqliteErrorCode.CONSTRAINT'
|
||||
|
||||
# Qt attributes
|
||||
yield 'PyQt5.QtWebKit.QWebPage.ErrorPageExtensionReturn().baseUrl'
|
||||
|
@ -429,7 +429,7 @@ def test_config_change(config_stub, basedir, download_stub,
|
||||
|
||||
host_blocker = adblock.HostBlocker()
|
||||
host_blocker.read_hosts()
|
||||
config_stub.set_obj('content.host_blocking.lists', None)
|
||||
config_stub.val.content.host_blocking.lists = None
|
||||
host_blocker.read_hosts()
|
||||
for str_url in URLS_TO_CHECK:
|
||||
assert not host_blocker.is_blocked(QUrl(str_url))
|
||||
|
@ -42,405 +42,488 @@ def hist(tmpdir):
|
||||
return history.WebHistory()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_time(mocker):
|
||||
m = mocker.patch('qutebrowser.browser.history.time')
|
||||
m.time.return_value = 12345
|
||||
return 12345
|
||||
class TestSpecialMethods:
|
||||
|
||||
def test_iter(self, hist):
|
||||
urlstr = 'http://www.example.com/'
|
||||
url = QUrl(urlstr)
|
||||
hist.add_url(url, atime=12345)
|
||||
|
||||
assert list(hist) == [(urlstr, '', 12345, False)]
|
||||
|
||||
def test_len(self, hist):
|
||||
assert len(hist) == 0
|
||||
|
||||
url = QUrl('http://www.example.com/')
|
||||
hist.add_url(url)
|
||||
|
||||
assert len(hist) == 1
|
||||
|
||||
def test_contains(self, hist):
|
||||
hist.add_url(QUrl('http://www.example.com/'), title='Title',
|
||||
atime=12345)
|
||||
assert 'http://www.example.com/' in hist
|
||||
assert 'www.example.com' not in hist
|
||||
assert 'Title' not in hist
|
||||
assert 12345 not in hist
|
||||
|
||||
|
||||
def test_iter(hist):
|
||||
urlstr = 'http://www.example.com/'
|
||||
url = QUrl(urlstr)
|
||||
hist.add_url(url, atime=12345)
|
||||
class TestGetting:
|
||||
|
||||
assert list(hist) == [(urlstr, '', 12345, False)]
|
||||
def test_get_recent(self, hist):
|
||||
hist.add_url(QUrl('http://www.qutebrowser.org/'), atime=67890)
|
||||
hist.add_url(QUrl('http://example.com/'), atime=12345)
|
||||
assert list(hist.get_recent()) == [
|
||||
('http://www.qutebrowser.org/', '', 67890, False),
|
||||
('http://example.com/', '', 12345, False),
|
||||
]
|
||||
|
||||
def test_entries_between(self, hist):
|
||||
hist.add_url(QUrl('http://www.example.com/1'), atime=12345)
|
||||
hist.add_url(QUrl('http://www.example.com/2'), atime=12346)
|
||||
hist.add_url(QUrl('http://www.example.com/3'), atime=12347)
|
||||
hist.add_url(QUrl('http://www.example.com/4'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/5'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/6'), atime=12349)
|
||||
hist.add_url(QUrl('http://www.example.com/7'), atime=12350)
|
||||
|
||||
times = [x.atime for x in hist.entries_between(12346, 12349)]
|
||||
assert times == [12349, 12348, 12348, 12347]
|
||||
|
||||
def test_entries_before(self, hist):
|
||||
hist.add_url(QUrl('http://www.example.com/1'), atime=12346)
|
||||
hist.add_url(QUrl('http://www.example.com/2'), atime=12346)
|
||||
hist.add_url(QUrl('http://www.example.com/3'), atime=12347)
|
||||
hist.add_url(QUrl('http://www.example.com/4'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/5'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/6'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/7'), atime=12349)
|
||||
hist.add_url(QUrl('http://www.example.com/8'), atime=12349)
|
||||
|
||||
times = [x.atime for x in
|
||||
hist.entries_before(12348, limit=3, offset=2)]
|
||||
assert times == [12348, 12347, 12346]
|
||||
|
||||
|
||||
def test_len(hist):
|
||||
assert len(hist) == 0
|
||||
class TestDelete:
|
||||
|
||||
url = QUrl('http://www.example.com/')
|
||||
hist.add_url(url)
|
||||
def test_clear(self, qtbot, tmpdir, hist, mocker):
|
||||
hist.add_url(QUrl('http://example.com/'))
|
||||
hist.add_url(QUrl('http://www.qutebrowser.org/'))
|
||||
|
||||
assert len(hist) == 1
|
||||
m = mocker.patch('qutebrowser.browser.history.message.confirm_async',
|
||||
new=mocker.Mock, spec=[])
|
||||
hist.clear()
|
||||
assert m.called
|
||||
|
||||
|
||||
def test_contains(hist):
|
||||
hist.add_url(QUrl('http://www.example.com/'), title='Title', atime=12345)
|
||||
assert 'http://www.example.com/' in hist
|
||||
assert 'www.example.com' not in hist
|
||||
assert 'Title' not in hist
|
||||
assert 12345 not in hist
|
||||
|
||||
|
||||
def test_get_recent(hist):
|
||||
hist.add_url(QUrl('http://www.qutebrowser.org/'), atime=67890)
|
||||
hist.add_url(QUrl('http://example.com/'), atime=12345)
|
||||
assert list(hist.get_recent()) == [
|
||||
('http://www.qutebrowser.org/', '', 67890, False),
|
||||
('http://example.com/', '', 12345, False),
|
||||
]
|
||||
|
||||
|
||||
def test_entries_between(hist):
|
||||
hist.add_url(QUrl('http://www.example.com/1'), atime=12345)
|
||||
hist.add_url(QUrl('http://www.example.com/2'), atime=12346)
|
||||
hist.add_url(QUrl('http://www.example.com/3'), atime=12347)
|
||||
hist.add_url(QUrl('http://www.example.com/4'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/5'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/6'), atime=12349)
|
||||
hist.add_url(QUrl('http://www.example.com/7'), atime=12350)
|
||||
|
||||
times = [x.atime for x in hist.entries_between(12346, 12349)]
|
||||
assert times == [12349, 12348, 12348, 12347]
|
||||
|
||||
|
||||
def test_entries_before(hist):
|
||||
hist.add_url(QUrl('http://www.example.com/1'), atime=12346)
|
||||
hist.add_url(QUrl('http://www.example.com/2'), atime=12346)
|
||||
hist.add_url(QUrl('http://www.example.com/3'), atime=12347)
|
||||
hist.add_url(QUrl('http://www.example.com/4'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/5'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/6'), atime=12348)
|
||||
hist.add_url(QUrl('http://www.example.com/7'), atime=12349)
|
||||
hist.add_url(QUrl('http://www.example.com/8'), atime=12349)
|
||||
|
||||
times = [x.atime for x in hist.entries_before(12348, limit=3, offset=2)]
|
||||
assert times == [12348, 12347, 12346]
|
||||
|
||||
|
||||
def test_clear(qtbot, tmpdir, hist, mocker):
|
||||
hist.add_url(QUrl('http://example.com/'))
|
||||
hist.add_url(QUrl('http://www.qutebrowser.org/'))
|
||||
|
||||
m = mocker.patch('qutebrowser.browser.history.message.confirm_async',
|
||||
new=mocker.Mock, spec=[])
|
||||
hist.clear()
|
||||
assert m.called
|
||||
|
||||
|
||||
def test_clear_force(qtbot, tmpdir, hist):
|
||||
hist.add_url(QUrl('http://example.com/'))
|
||||
hist.add_url(QUrl('http://www.qutebrowser.org/'))
|
||||
hist.clear(force=True)
|
||||
assert not len(hist)
|
||||
assert not len(hist.completion)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('raw, escaped', [
|
||||
('http://example.com/1', 'http://example.com/1'),
|
||||
('http://example.com/1 2', 'http://example.com/1%202'),
|
||||
])
|
||||
def test_delete_url(hist, raw, escaped):
|
||||
hist.add_url(QUrl('http://example.com/'), atime=0)
|
||||
hist.add_url(QUrl(escaped), atime=0)
|
||||
hist.add_url(QUrl('http://example.com/2'), atime=0)
|
||||
|
||||
before = set(hist)
|
||||
completion_before = set(hist.completion)
|
||||
|
||||
hist.delete_url(QUrl(raw))
|
||||
|
||||
diff = before.difference(set(hist))
|
||||
assert diff == {(escaped, '', 0, False)}
|
||||
|
||||
completion_diff = completion_before.difference(set(hist.completion))
|
||||
assert completion_diff == {(raw, '', 0)}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'url, atime, title, redirect, history_url, completion_url', [
|
||||
|
||||
('http://www.example.com', 12346, 'the title', False,
|
||||
'http://www.example.com', 'http://www.example.com'),
|
||||
('http://www.example.com', 12346, 'the title', True,
|
||||
'http://www.example.com', None),
|
||||
('http://www.example.com/sp ce', 12346, 'the title', False,
|
||||
'http://www.example.com/sp%20ce', 'http://www.example.com/sp ce'),
|
||||
('https://user:pass@example.com', 12346, 'the title', False,
|
||||
'https://user@example.com', 'https://user@example.com'),
|
||||
]
|
||||
)
|
||||
def test_add_url(qtbot, hist, url, atime, title, redirect, history_url,
|
||||
completion_url):
|
||||
hist.add_url(QUrl(url), atime=atime, title=title, redirect=redirect)
|
||||
assert list(hist) == [(history_url, title, atime, redirect)]
|
||||
if completion_url is None:
|
||||
def test_clear_force(self, qtbot, tmpdir, hist):
|
||||
hist.add_url(QUrl('http://example.com/'))
|
||||
hist.add_url(QUrl('http://www.qutebrowser.org/'))
|
||||
hist.clear(force=True)
|
||||
assert not len(hist)
|
||||
assert not len(hist.completion)
|
||||
else:
|
||||
assert list(hist.completion) == [(completion_url, title, atime)]
|
||||
|
||||
@pytest.mark.parametrize('raw, escaped', [
|
||||
('http://example.com/1', 'http://example.com/1'),
|
||||
('http://example.com/1 2', 'http://example.com/1%202'),
|
||||
])
|
||||
def test_delete_url(self, hist, raw, escaped):
|
||||
hist.add_url(QUrl('http://example.com/'), atime=0)
|
||||
hist.add_url(QUrl(escaped), atime=0)
|
||||
hist.add_url(QUrl('http://example.com/2'), atime=0)
|
||||
|
||||
before = set(hist)
|
||||
completion_before = set(hist.completion)
|
||||
|
||||
hist.delete_url(QUrl(raw))
|
||||
|
||||
diff = before.difference(set(hist))
|
||||
assert diff == {(escaped, '', 0, False)}
|
||||
|
||||
completion_diff = completion_before.difference(set(hist.completion))
|
||||
assert completion_diff == {(raw, '', 0)}
|
||||
|
||||
|
||||
def test_add_url_invalid(qtbot, hist, caplog):
|
||||
with caplog.at_level(logging.WARNING):
|
||||
hist.add_url(QUrl())
|
||||
assert not list(hist)
|
||||
assert not list(hist.completion)
|
||||
class TestAdd:
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_time(self, mocker):
|
||||
m = mocker.patch('qutebrowser.browser.history.time')
|
||||
m.time.return_value = 12345
|
||||
return 12345
|
||||
|
||||
@pytest.mark.parametrize('environmental', [True, False])
|
||||
@pytest.mark.parametrize('completion', [True, False])
|
||||
def test_add_url_error(monkeypatch, hist, message_mock, caplog,
|
||||
environmental, completion):
|
||||
def raise_error(url, replace=False):
|
||||
raise sql.SqlError("Error message", environmental=environmental)
|
||||
|
||||
if completion:
|
||||
monkeypatch.setattr(hist.completion, 'insert', raise_error)
|
||||
else:
|
||||
monkeypatch.setattr(hist, 'insert', raise_error)
|
||||
|
||||
if environmental:
|
||||
with caplog.at_level(logging.ERROR):
|
||||
hist.add_url(QUrl('https://www.example.org/'))
|
||||
msg = message_mock.getmsg(usertypes.MessageLevel.error)
|
||||
assert msg.text == "Failed to write history: Error message"
|
||||
else:
|
||||
with pytest.raises(sql.SqlError):
|
||||
hist.add_url(QUrl('https://www.example.org/'))
|
||||
|
||||
|
||||
@pytest.mark.parametrize('level, url, req_url, expected', [
|
||||
(logging.DEBUG, 'a.com', 'a.com', [('a.com', 'title', 12345, False)]),
|
||||
(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', []),
|
||||
])
|
||||
def test_add_from_tab(hist, level, url, req_url, expected, mock_time, caplog):
|
||||
with caplog.at_level(level):
|
||||
hist.add_from_tab(QUrl(url), QUrl(req_url), 'title')
|
||||
assert set(hist) == set(expected)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hist_interface(hist):
|
||||
# pylint: disable=invalid-name
|
||||
QtWebKit = pytest.importorskip('PyQt5.QtWebKit')
|
||||
from qutebrowser.browser.webkit import webkithistory
|
||||
QWebHistoryInterface = QtWebKit.QWebHistoryInterface
|
||||
# pylint: enable=invalid-name
|
||||
hist.add_url(url=QUrl('http://www.example.com/'), title='example')
|
||||
interface = webkithistory.WebHistoryInterface(hist)
|
||||
QWebHistoryInterface.setDefaultInterface(interface)
|
||||
yield
|
||||
QWebHistoryInterface.setDefaultInterface(None)
|
||||
|
||||
|
||||
def test_history_interface(qtbot, webview, hist_interface):
|
||||
html = b"<a href='about:blank'>foo</a>"
|
||||
url = urlutils.data_url('text/html', html)
|
||||
with qtbot.waitSignal(webview.loadFinished):
|
||||
webview.load(url)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cleanup_init():
|
||||
# prevent test_init from leaking state
|
||||
yield
|
||||
hist = objreg.get('web-history', None)
|
||||
if hist is not None:
|
||||
hist.setParent(None)
|
||||
objreg.delete('web-history')
|
||||
try:
|
||||
from PyQt5.QtWebKit import QWebHistoryInterface
|
||||
QWebHistoryInterface.setDefaultInterface(None)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.parametrize('backend', [usertypes.Backend.QtWebEngine,
|
||||
usertypes.Backend.QtWebKit])
|
||||
def test_init(backend, qapp, tmpdir, monkeypatch, cleanup_init):
|
||||
if backend == usertypes.Backend.QtWebKit:
|
||||
pytest.importorskip('PyQt5.QtWebKitWidgets')
|
||||
else:
|
||||
assert backend == usertypes.Backend.QtWebEngine
|
||||
|
||||
monkeypatch.setattr(history.objects, 'backend', backend)
|
||||
history.init(qapp)
|
||||
hist = objreg.get('web-history')
|
||||
assert hist.parent() is qapp
|
||||
|
||||
try:
|
||||
from PyQt5.QtWebKit import QWebHistoryInterface
|
||||
except ImportError:
|
||||
QWebHistoryInterface = None
|
||||
|
||||
if backend == usertypes.Backend.QtWebKit:
|
||||
default_interface = QWebHistoryInterface.defaultInterface()
|
||||
assert default_interface._history is hist
|
||||
else:
|
||||
assert backend == usertypes.Backend.QtWebEngine
|
||||
if QWebHistoryInterface is None:
|
||||
default_interface = None
|
||||
@pytest.mark.parametrize(
|
||||
'url, atime, title, redirect, history_url, completion_url', [
|
||||
('http://www.example.com', 12346, 'the title', False,
|
||||
'http://www.example.com', 'http://www.example.com'),
|
||||
('http://www.example.com', 12346, 'the title', True,
|
||||
'http://www.example.com', None),
|
||||
('http://www.example.com/sp ce', 12346, 'the title', False,
|
||||
'http://www.example.com/sp%20ce', 'http://www.example.com/sp ce'),
|
||||
('https://user:pass@example.com', 12346, 'the title', False,
|
||||
'https://user@example.com', 'https://user@example.com'),
|
||||
]
|
||||
)
|
||||
def test_add_url(self, qtbot, hist,
|
||||
url, atime, title, redirect, history_url, completion_url):
|
||||
hist.add_url(QUrl(url), atime=atime, title=title, redirect=redirect)
|
||||
assert list(hist) == [(history_url, title, atime, redirect)]
|
||||
if completion_url is None:
|
||||
assert not len(hist.completion)
|
||||
else:
|
||||
assert list(hist.completion) == [(completion_url, title, atime)]
|
||||
|
||||
def test_no_sql_history(self, hist, fake_args):
|
||||
fake_args.debug_flags = 'no-sql-history'
|
||||
hist.add_url(QUrl('https://www.example.com/'), atime=12346,
|
||||
title='Hello World', redirect=False)
|
||||
assert not list(hist)
|
||||
|
||||
def test_invalid(self, qtbot, hist, caplog):
|
||||
with caplog.at_level(logging.WARNING):
|
||||
hist.add_url(QUrl())
|
||||
assert not list(hist)
|
||||
assert not list(hist.completion)
|
||||
|
||||
@pytest.mark.parametrize('environmental', [True, False])
|
||||
@pytest.mark.parametrize('completion', [True, False])
|
||||
def test_error(self, monkeypatch, hist, message_mock, caplog,
|
||||
environmental, completion):
|
||||
def raise_error(url, replace=False):
|
||||
if environmental:
|
||||
raise sql.SqlEnvironmentError("Error message")
|
||||
else:
|
||||
raise sql.SqlBugError("Error message")
|
||||
|
||||
if completion:
|
||||
monkeypatch.setattr(hist.completion, 'insert', raise_error)
|
||||
else:
|
||||
monkeypatch.setattr(hist, 'insert', raise_error)
|
||||
|
||||
if environmental:
|
||||
with caplog.at_level(logging.ERROR):
|
||||
hist.add_url(QUrl('https://www.example.org/'))
|
||||
msg = message_mock.getmsg(usertypes.MessageLevel.error)
|
||||
assert msg.text == "Failed to write history: Error message"
|
||||
else:
|
||||
with pytest.raises(sql.SqlBugError):
|
||||
hist.add_url(QUrl('https://www.example.org/'))
|
||||
|
||||
@pytest.mark.parametrize('level, url, req_url, expected', [
|
||||
(logging.DEBUG, 'a.com', 'a.com', [('a.com', 'title', 12345, False)]),
|
||||
(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', []),
|
||||
])
|
||||
def test_from_tab(self, hist, caplog, mock_time,
|
||||
level, url, req_url, expected):
|
||||
with caplog.at_level(level):
|
||||
hist.add_from_tab(QUrl(url), QUrl(req_url), 'title')
|
||||
assert set(hist) == set(expected)
|
||||
|
||||
def test_exclude(self, hist, config_stub):
|
||||
"""Excluded URLs should be in the history but not completion."""
|
||||
config_stub.val.completion.web_history.exclude = ['*.example.org']
|
||||
url = QUrl('http://www.example.org/')
|
||||
hist.add_from_tab(url, url, 'title')
|
||||
assert list(hist)
|
||||
assert not list(hist.completion)
|
||||
|
||||
|
||||
class TestHistoryInterface:
|
||||
|
||||
@pytest.fixture
|
||||
def hist_interface(self, hist):
|
||||
# pylint: disable=invalid-name
|
||||
QtWebKit = pytest.importorskip('PyQt5.QtWebKit')
|
||||
from qutebrowser.browser.webkit import webkithistory
|
||||
QWebHistoryInterface = QtWebKit.QWebHistoryInterface
|
||||
# pylint: enable=invalid-name
|
||||
hist.add_url(url=QUrl('http://www.example.com/'), title='example')
|
||||
interface = webkithistory.WebHistoryInterface(hist)
|
||||
QWebHistoryInterface.setDefaultInterface(interface)
|
||||
yield
|
||||
QWebHistoryInterface.setDefaultInterface(None)
|
||||
|
||||
def test_history_interface(self, qtbot, webview, hist_interface):
|
||||
html = b"<a href='about:blank'>foo</a>"
|
||||
url = urlutils.data_url('text/html', html)
|
||||
with qtbot.waitSignal(webview.loadFinished):
|
||||
webview.load(url)
|
||||
|
||||
|
||||
class TestInit:
|
||||
|
||||
@pytest.fixture
|
||||
def cleanup_init(self):
|
||||
# prevent test_init from leaking state
|
||||
yield
|
||||
hist = objreg.get('web-history', None)
|
||||
if hist is not None:
|
||||
hist.setParent(None)
|
||||
objreg.delete('web-history')
|
||||
try:
|
||||
from PyQt5.QtWebKit import QWebHistoryInterface
|
||||
QWebHistoryInterface.setDefaultInterface(None)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
@pytest.mark.parametrize('backend', [usertypes.Backend.QtWebEngine,
|
||||
usertypes.Backend.QtWebKit])
|
||||
def test_init(self, backend, qapp, tmpdir, monkeypatch, cleanup_init):
|
||||
if backend == usertypes.Backend.QtWebKit:
|
||||
pytest.importorskip('PyQt5.QtWebKitWidgets')
|
||||
else:
|
||||
assert backend == usertypes.Backend.QtWebEngine
|
||||
|
||||
monkeypatch.setattr(history.objects, 'backend', backend)
|
||||
history.init(qapp)
|
||||
hist = objreg.get('web-history')
|
||||
assert hist.parent() is qapp
|
||||
|
||||
try:
|
||||
from PyQt5.QtWebKit import QWebHistoryInterface
|
||||
except ImportError:
|
||||
QWebHistoryInterface = None
|
||||
|
||||
if backend == usertypes.Backend.QtWebKit:
|
||||
default_interface = QWebHistoryInterface.defaultInterface()
|
||||
# For this to work, nothing can ever have called setDefaultInterface
|
||||
# before (so we need to test webengine before webkit)
|
||||
assert default_interface is None
|
||||
assert default_interface._history is hist
|
||||
else:
|
||||
assert backend == usertypes.Backend.QtWebEngine
|
||||
if QWebHistoryInterface is None:
|
||||
default_interface = None
|
||||
else:
|
||||
default_interface = QWebHistoryInterface.defaultInterface()
|
||||
# For this to work, nothing can ever have called
|
||||
# setDefaultInterface before (so we need to test webengine before
|
||||
# webkit)
|
||||
assert default_interface is None
|
||||
|
||||
|
||||
def test_import_txt(hist, data_tmpdir, monkeypatch, stubs):
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
# empty line is deliberate, to test skipping empty lines
|
||||
histfile.write('''12345 http://example.com/ title
|
||||
12346 http://qutebrowser.org/
|
||||
67890 http://example.com/path
|
||||
class TestImport:
|
||||
|
||||
68891-r http://example.com/path/other ''')
|
||||
def test_import_txt(self, hist, data_tmpdir, monkeypatch, stubs):
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
# empty line is deliberate, to test skipping empty lines
|
||||
histfile.write('''12345 http://example.com/ title
|
||||
12346 http://qutebrowser.org/
|
||||
67890 http://example.com/path
|
||||
|
||||
hist.import_txt()
|
||||
68891-r http://example.com/path/other ''')
|
||||
|
||||
assert list(hist) == [
|
||||
('http://example.com/', 'title', 12345, False),
|
||||
('http://qutebrowser.org/', '', 12346, False),
|
||||
('http://example.com/path', '', 67890, False),
|
||||
('http://example.com/path/other', '', 68891, True)
|
||||
]
|
||||
|
||||
assert not histfile.exists()
|
||||
assert (data_tmpdir / 'history.bak').exists()
|
||||
|
||||
|
||||
def test_import_txt_existing_backup(hist, data_tmpdir, monkeypatch, stubs):
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
bakfile = data_tmpdir / 'history.bak'
|
||||
histfile.write('12345 http://example.com/ title')
|
||||
bakfile.write('12346 http://qutebrowser.org/')
|
||||
|
||||
hist.import_txt()
|
||||
|
||||
assert list(hist) == [('http://example.com/', 'title', 12345, False)]
|
||||
|
||||
assert not histfile.exists()
|
||||
assert bakfile.read().split('\n') == ['12346 http://qutebrowser.org/',
|
||||
'12345 http://example.com/ title']
|
||||
|
||||
|
||||
@pytest.mark.parametrize('line', [
|
||||
'',
|
||||
'#12345 http://example.com/commented',
|
||||
|
||||
# https://bugreports.qt.io/browse/QTBUG-60364
|
||||
'12345 http://.com/',
|
||||
'12345 https://.com/',
|
||||
'12345 http://www..com/',
|
||||
'12345 https://www..com/',
|
||||
|
||||
# issue #2646
|
||||
'12345 data:text/html;charset=UTF-8,%3C%21DOCTYPE%20html%20PUBLIC%20%22-',
|
||||
])
|
||||
def test_import_txt_skip(hist, data_tmpdir, line, monkeypatch, stubs):
|
||||
"""import_txt should skip certain lines silently."""
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
histfile.write(line)
|
||||
|
||||
hist.import_txt()
|
||||
|
||||
assert not histfile.exists()
|
||||
assert not len(hist)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('line', [
|
||||
'xyz http://example.com/bad-timestamp',
|
||||
'12345',
|
||||
'http://example.com/no-timestamp',
|
||||
'68891-r-r http://example.com/double-flag',
|
||||
'68891-x http://example.com/bad-flag',
|
||||
'68891 http://.com',
|
||||
])
|
||||
def test_import_txt_invalid(hist, data_tmpdir, line, monkeypatch, stubs,
|
||||
caplog):
|
||||
"""import_txt should fail on certain lines."""
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
histfile.write(line)
|
||||
|
||||
with caplog.at_level(logging.ERROR):
|
||||
hist.import_txt()
|
||||
|
||||
assert any(rec.msg.startswith("Failed to import history:")
|
||||
for rec in caplog.records)
|
||||
assert list(hist) == [
|
||||
('http://example.com/', 'title', 12345, False),
|
||||
('http://qutebrowser.org/', '', 12346, False),
|
||||
('http://example.com/path', '', 67890, False),
|
||||
('http://example.com/path/other', '', 68891, True)
|
||||
]
|
||||
|
||||
assert histfile.exists()
|
||||
assert not histfile.exists()
|
||||
assert (data_tmpdir / 'history.bak').exists()
|
||||
|
||||
def test_existing_backup(self, hist, data_tmpdir, monkeypatch, stubs):
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
bakfile = data_tmpdir / 'history.bak'
|
||||
histfile.write('12345 http://example.com/ title')
|
||||
bakfile.write('12346 http://qutebrowser.org/')
|
||||
|
||||
hist.import_txt()
|
||||
|
||||
assert list(hist) == [('http://example.com/', 'title', 12345, False)]
|
||||
|
||||
assert not histfile.exists()
|
||||
assert bakfile.read().split('\n') == ['12346 http://qutebrowser.org/',
|
||||
'12345 http://example.com/ title']
|
||||
|
||||
@pytest.mark.parametrize('line', [
|
||||
'',
|
||||
'#12345 http://example.com/commented',
|
||||
|
||||
# https://bugreports.qt.io/browse/QTBUG-60364
|
||||
'12345 http://.com/',
|
||||
'12345 https://.com/',
|
||||
'12345 http://www..com/',
|
||||
'12345 https://www..com/',
|
||||
|
||||
# issue #2646
|
||||
('12345 data:text/html;'
|
||||
'charset=UTF-8,%3C%21DOCTYPE%20html%20PUBLIC%20%22-'),
|
||||
])
|
||||
def test_skip(self, hist, data_tmpdir, monkeypatch, stubs, line):
|
||||
"""import_txt should skip certain lines silently."""
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
histfile.write(line)
|
||||
|
||||
hist.import_txt()
|
||||
|
||||
assert not histfile.exists()
|
||||
assert not len(hist)
|
||||
|
||||
@pytest.mark.parametrize('line', [
|
||||
'xyz http://example.com/bad-timestamp',
|
||||
'12345',
|
||||
'http://example.com/no-timestamp',
|
||||
'68891-r-r http://example.com/double-flag',
|
||||
'68891-x http://example.com/bad-flag',
|
||||
'68891 http://.com',
|
||||
])
|
||||
def test_invalid(self, hist, data_tmpdir, monkeypatch, stubs, caplog,
|
||||
line):
|
||||
"""import_txt should fail on certain lines."""
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
histfile = data_tmpdir / 'history'
|
||||
histfile.write(line)
|
||||
|
||||
with caplog.at_level(logging.ERROR):
|
||||
hist.import_txt()
|
||||
|
||||
assert any(rec.msg.startswith("Failed to import history:")
|
||||
for rec in caplog.records)
|
||||
|
||||
assert histfile.exists()
|
||||
|
||||
def test_nonexistent(self, hist, data_tmpdir, monkeypatch, stubs):
|
||||
"""import_txt should do nothing if the history file doesn't exist."""
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
hist.import_txt()
|
||||
|
||||
|
||||
def test_import_txt_nonexistent(hist, data_tmpdir, monkeypatch, stubs):
|
||||
"""import_txt should do nothing if the history file doesn't exist."""
|
||||
monkeypatch.setattr(history, 'QTimer', stubs.InstaTimer)
|
||||
hist.import_txt()
|
||||
class TestDump:
|
||||
|
||||
|
||||
def test_debug_dump_history(hist, tmpdir):
|
||||
hist.add_url(QUrl('http://example.com/1'), title="Title1", atime=12345)
|
||||
hist.add_url(QUrl('http://example.com/2'), title="Title2", atime=12346)
|
||||
hist.add_url(QUrl('http://example.com/3'), title="Title3", atime=12347)
|
||||
hist.add_url(QUrl('http://example.com/4'), title="Title4", atime=12348,
|
||||
redirect=True)
|
||||
histfile = tmpdir / 'history'
|
||||
hist.debug_dump_history(str(histfile))
|
||||
expected = ['12345 http://example.com/1 Title1',
|
||||
'12346 http://example.com/2 Title2',
|
||||
'12347 http://example.com/3 Title3',
|
||||
'12348-r http://example.com/4 Title4']
|
||||
assert histfile.read() == '\n'.join(expected)
|
||||
|
||||
|
||||
def test_debug_dump_history_nonexistent(hist, tmpdir):
|
||||
histfile = tmpdir / 'nonexistent' / 'history'
|
||||
with pytest.raises(cmdexc.CommandError):
|
||||
def test_debug_dump_history(self, hist, tmpdir):
|
||||
hist.add_url(QUrl('http://example.com/1'), title="Title1", atime=12345)
|
||||
hist.add_url(QUrl('http://example.com/2'), title="Title2", atime=12346)
|
||||
hist.add_url(QUrl('http://example.com/3'), title="Title3", atime=12347)
|
||||
hist.add_url(QUrl('http://example.com/4'), title="Title4", atime=12348,
|
||||
redirect=True)
|
||||
histfile = tmpdir / 'history'
|
||||
hist.debug_dump_history(str(histfile))
|
||||
expected = ['12345 http://example.com/1 Title1',
|
||||
'12346 http://example.com/2 Title2',
|
||||
'12347 http://example.com/3 Title3',
|
||||
'12348-r http://example.com/4 Title4']
|
||||
assert histfile.read() == '\n'.join(expected)
|
||||
|
||||
def test_nonexistent(self, hist, tmpdir):
|
||||
histfile = tmpdir / 'nonexistent' / 'history'
|
||||
with pytest.raises(cmdexc.CommandError):
|
||||
hist.debug_dump_history(str(histfile))
|
||||
|
||||
|
||||
def test_rebuild_completion(hist):
|
||||
hist.insert({'url': 'example.com/1', 'title': 'example1',
|
||||
'redirect': False, 'atime': 1})
|
||||
hist.insert({'url': 'example.com/1', 'title': 'example1',
|
||||
'redirect': False, 'atime': 2})
|
||||
hist.insert({'url': 'example.com/2%203', 'title': 'example2',
|
||||
'redirect': False, 'atime': 3})
|
||||
hist.insert({'url': 'example.com/3', 'title': 'example3',
|
||||
'redirect': True, 'atime': 4})
|
||||
hist.insert({'url': 'example.com/2 3', 'title': 'example2',
|
||||
'redirect': False, 'atime': 5})
|
||||
hist.completion.delete_all()
|
||||
class TestRebuild:
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [
|
||||
('example.com/1', 'example1', 2),
|
||||
('example.com/2 3', 'example2', 5),
|
||||
]
|
||||
def test_delete(self, hist):
|
||||
hist.insert({'url': 'example.com/1', 'title': 'example1',
|
||||
'redirect': False, 'atime': 1})
|
||||
hist.insert({'url': 'example.com/1', 'title': 'example1',
|
||||
'redirect': False, 'atime': 2})
|
||||
hist.insert({'url': 'example.com/2%203', 'title': 'example2',
|
||||
'redirect': False, 'atime': 3})
|
||||
hist.insert({'url': 'example.com/3', 'title': 'example3',
|
||||
'redirect': True, 'atime': 4})
|
||||
hist.insert({'url': 'example.com/2 3', 'title': 'example2',
|
||||
'redirect': False, 'atime': 5})
|
||||
hist.completion.delete_all()
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [
|
||||
('example.com/1', 'example1', 2),
|
||||
('example.com/2 3', 'example2', 5),
|
||||
]
|
||||
|
||||
def test_no_rebuild(self, hist):
|
||||
"""Ensure that completion is not regenerated unless empty."""
|
||||
hist.add_url(QUrl('example.com/1'), redirect=False, atime=1)
|
||||
hist.add_url(QUrl('example.com/2'), redirect=False, atime=2)
|
||||
hist.completion.delete('url', 'example.com/2')
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [('example.com/1', '', 1)]
|
||||
|
||||
def test_user_version(self, hist, monkeypatch):
|
||||
"""Ensure that completion is regenerated if user_version changes."""
|
||||
hist.add_url(QUrl('example.com/1'), redirect=False, atime=1)
|
||||
hist.add_url(QUrl('example.com/2'), redirect=False, atime=2)
|
||||
hist.completion.delete('url', 'example.com/2')
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [('example.com/1', '', 1)]
|
||||
|
||||
monkeypatch.setattr(history, '_USER_VERSION',
|
||||
history._USER_VERSION + 1)
|
||||
hist3 = history.WebHistory()
|
||||
assert list(hist3.completion) == [
|
||||
('example.com/1', '', 1),
|
||||
('example.com/2', '', 2),
|
||||
]
|
||||
|
||||
def test_force_rebuild(self, hist):
|
||||
"""Ensure that completion is regenerated if we force a rebuild."""
|
||||
hist.add_url(QUrl('example.com/1'), redirect=False, atime=1)
|
||||
hist.add_url(QUrl('example.com/2'), redirect=False, atime=2)
|
||||
hist.completion.delete('url', 'example.com/2')
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [('example.com/1', '', 1)]
|
||||
hist2.metainfo['force_rebuild'] = True
|
||||
|
||||
hist3 = history.WebHistory()
|
||||
assert list(hist3.completion) == [
|
||||
('example.com/1', '', 1),
|
||||
('example.com/2', '', 2),
|
||||
]
|
||||
assert not hist3.metainfo['force_rebuild']
|
||||
|
||||
def test_exclude(self, config_stub, hist):
|
||||
"""Ensure that patterns in completion.web_history.exclude are ignored.
|
||||
|
||||
This setting should only be used for the completion.
|
||||
"""
|
||||
config_stub.val.completion.web_history.exclude = ['*.example.org']
|
||||
assert hist.metainfo['force_rebuild']
|
||||
|
||||
hist.add_url(QUrl('http://example.com'), redirect=False, atime=1)
|
||||
hist.add_url(QUrl('http://example.org'), redirect=False, atime=2)
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [('http://example.com', '', 1)]
|
||||
|
||||
def test_unrelated_config_change(self, config_stub, hist):
|
||||
config_stub.val.history_gap_interval = 1234
|
||||
assert not hist.metainfo['force_rebuild']
|
||||
|
||||
|
||||
def test_no_rebuild_completion(hist):
|
||||
"""Ensure that completion is not regenerated unless completely empty."""
|
||||
hist.add_url(QUrl('example.com/1'), redirect=False, atime=1)
|
||||
hist.add_url(QUrl('example.com/2'), redirect=False, atime=2)
|
||||
hist.completion.delete('url', 'example.com/2')
|
||||
class TestCompletionMetaInfo:
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [('example.com/1', '', 1)]
|
||||
@pytest.fixture
|
||||
def metainfo(self):
|
||||
return history.CompletionMetaInfo()
|
||||
|
||||
def test_contains_keyerror(self, metainfo):
|
||||
with pytest.raises(KeyError):
|
||||
'does_not_exist' in metainfo # pylint: disable=pointless-statement
|
||||
|
||||
def test_user_version(hist, monkeypatch):
|
||||
"""Ensure that completion is regenerated if user_version is incremented."""
|
||||
hist.add_url(QUrl('example.com/1'), redirect=False, atime=1)
|
||||
hist.add_url(QUrl('example.com/2'), redirect=False, atime=2)
|
||||
hist.completion.delete('url', 'example.com/2')
|
||||
def test_getitem_keyerror(self, metainfo):
|
||||
with pytest.raises(KeyError):
|
||||
metainfo['does_not_exist'] # pylint: disable=pointless-statement
|
||||
|
||||
hist2 = history.WebHistory()
|
||||
assert list(hist2.completion) == [('example.com/1', '', 1)]
|
||||
def test_setitem_keyerror(self, metainfo):
|
||||
with pytest.raises(KeyError):
|
||||
metainfo['does_not_exist'] = 42
|
||||
|
||||
monkeypatch.setattr(history, '_USER_VERSION', history._USER_VERSION + 1)
|
||||
hist3 = history.WebHistory()
|
||||
assert list(hist3.completion) == [
|
||||
('example.com/1', '', 1),
|
||||
('example.com/2', '', 2),
|
||||
]
|
||||
def test_contains(self, metainfo):
|
||||
assert 'force_rebuild' in metainfo
|
||||
|
||||
def test_modify(self, metainfo):
|
||||
assert not metainfo['force_rebuild']
|
||||
metainfo['force_rebuild'] = True
|
||||
assert metainfo['force_rebuild']
|
||||
|
@ -90,14 +90,15 @@ class TestHistoryHandler:
|
||||
for i in range(entry_count):
|
||||
entry_atime = now - i * interval
|
||||
entry = {"atime": str(entry_atime),
|
||||
"url": QUrl("www.x.com/" + str(i)),
|
||||
"url": QUrl("http://www.x.com/" + str(i)),
|
||||
"title": "Page " + str(i)}
|
||||
items.insert(0, entry)
|
||||
|
||||
return items
|
||||
|
||||
@pytest.fixture
|
||||
def fake_web_history(self, fake_save_manager, tmpdir, init_sql):
|
||||
def fake_web_history(self, fake_save_manager, tmpdir, init_sql,
|
||||
config_stub):
|
||||
"""Create a fake web-history and register it into objreg."""
|
||||
web_history = history.WebHistory()
|
||||
objreg.register('web-history', web_history)
|
||||
@ -133,6 +134,15 @@ class TestHistoryHandler:
|
||||
assert item['time'] <= start_time
|
||||
assert item['time'] > end_time
|
||||
|
||||
def test_exclude(self, fake_web_history, now, config_stub):
|
||||
"""Make sure the completion.web_history.exclude setting is not used."""
|
||||
config_stub.val.completion.web_history.exclude = ['www.x.com']
|
||||
|
||||
url = QUrl("qute://history/data?start_time={}".format(now))
|
||||
_mimetype, data = qutescheme.qute_history(url)
|
||||
items = json.loads(data)
|
||||
assert items
|
||||
|
||||
def test_qute_history_benchmark(self, fake_web_history, benchmark, now):
|
||||
r = range(100000)
|
||||
entries = {
|
||||
|
@ -30,7 +30,7 @@ from qutebrowser.completion.models import histcategory
|
||||
@pytest.fixture
|
||||
def hist(init_sql, config_stub):
|
||||
config_stub.val.completion.timestamp_format = '%Y-%m-%d'
|
||||
config_stub.val.completion.web_history_max_items = -1
|
||||
config_stub.val.completion.web_history.max_items = -1
|
||||
return sql.SqlTable('CompletionHistory', ['url', 'title', 'last_atime'])
|
||||
|
||||
|
||||
@ -165,7 +165,7 @@ def test_set_pattern_repeated(model_validator, hist):
|
||||
])
|
||||
def test_sorting(max_items, before, after, model_validator, hist, config_stub):
|
||||
"""Validate the filtering and sorting results of set_pattern."""
|
||||
config_stub.val.completion.web_history_max_items = max_items
|
||||
config_stub.val.completion.web_history.max_items = max_items
|
||||
for url, title, atime in before:
|
||||
timestamp = datetime.datetime.strptime(atime, '%Y-%m-%d').timestamp()
|
||||
hist.insert({'url': url, 'title': title, 'last_atime': timestamp})
|
||||
|
@ -172,7 +172,7 @@ def bookmarks(bookmark_manager_stub):
|
||||
def web_history(init_sql, stubs, config_stub):
|
||||
"""Fixture which provides a web-history object."""
|
||||
config_stub.val.completion.timestamp_format = '%Y-%m-%d'
|
||||
config_stub.val.completion.web_history_max_items = -1
|
||||
config_stub.val.completion.web_history.max_items = -1
|
||||
stub = history.WebHistory()
|
||||
objreg.register('web-history', stub)
|
||||
yield stub
|
||||
@ -518,7 +518,7 @@ def test_url_completion_delete_history(qtmodeltester, info,
|
||||
def test_url_completion_zero_limit(config_stub, web_history, quickmarks, info,
|
||||
bookmarks):
|
||||
"""Make sure there's no history if the limit was set to zero."""
|
||||
config_stub.val.completion.web_history_max_items = 0
|
||||
config_stub.val.completion.web_history.max_items = 0
|
||||
model = urlmodel.url(info=info)
|
||||
model.set_pattern('')
|
||||
category = model.index(2, 0) # "History" normally
|
||||
|
@ -35,7 +35,7 @@ from PyQt5.QtGui import QColor, QFont
|
||||
from PyQt5.QtNetwork import QNetworkProxy
|
||||
|
||||
from qutebrowser.config import configtypes, configexc
|
||||
from qutebrowser.utils import debug, utils, qtutils
|
||||
from qutebrowser.utils import debug, utils, qtutils, urlmatch
|
||||
from qutebrowser.browser.network import pac
|
||||
from qutebrowser.keyinput import keyutils
|
||||
from tests.helpers import utils as testutils
|
||||
@ -2099,6 +2099,21 @@ class TestKey:
|
||||
klass().to_py(val)
|
||||
|
||||
|
||||
class TestUrlPattern:
|
||||
|
||||
@pytest.fixture
|
||||
def klass(self):
|
||||
return configtypes.UrlPattern
|
||||
|
||||
def test_to_py_valid(self, klass):
|
||||
pattern = 'http://*.example.com/'
|
||||
assert klass().to_py(pattern) == urlmatch.UrlPattern(pattern)
|
||||
|
||||
def test_to_py_invalid(self, klass):
|
||||
with pytest.raises(configexc.ValidationError):
|
||||
klass().to_py('http://')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('first, second, equal', [
|
||||
(re.compile('foo'), RegexEq('foo'), True),
|
||||
(RegexEq('bar'), re.compile('bar'), True),
|
||||
|
@ -81,7 +81,7 @@ def test_changing_timer_with_messages_shown(qtbot, view, config_stub):
|
||||
config_stub.val.messages.timeout = 900000 # 15s
|
||||
view.show_message(usertypes.MessageLevel.info, 'test')
|
||||
with qtbot.waitSignal(view._clear_timer.timeout):
|
||||
config_stub.set_obj('messages.timeout', 100)
|
||||
config_stub.val.messages.timeout = 100
|
||||
|
||||
|
||||
@pytest.mark.parametrize('count, expected', [(1, 100), (3, 300),
|
||||
|
@ -29,30 +29,32 @@ from qutebrowser.misc import sql
|
||||
pytestmark = pytest.mark.usefixtures('init_sql')
|
||||
|
||||
|
||||
def test_sqlerror():
|
||||
@pytest.mark.parametrize('klass', [sql.SqlEnvironmentError, sql.SqlBugError])
|
||||
def test_sqlerror(klass):
|
||||
text = "Hello World"
|
||||
err = sql.SqlError(text, environmental=True)
|
||||
err = klass(text)
|
||||
assert str(err) == text
|
||||
assert err.text() == text
|
||||
assert err.environmental
|
||||
|
||||
|
||||
class TestSqliteError:
|
||||
class TestSqlError:
|
||||
|
||||
@pytest.mark.parametrize('error_code, environmental', [
|
||||
('5', True), # SQLITE_BUSY
|
||||
('19', False), # SQLITE_CONSTRAINT
|
||||
@pytest.mark.parametrize('error_code, exception', [
|
||||
(sql.SqliteErrorCode.BUSY, sql.SqlEnvironmentError),
|
||||
(sql.SqliteErrorCode.CONSTRAINT, sql.SqlBugError),
|
||||
])
|
||||
def test_environmental(self, error_code, environmental):
|
||||
def test_environmental(self, error_code, exception):
|
||||
sql_err = QSqlError("driver text", "db text", QSqlError.UnknownError,
|
||||
error_code)
|
||||
err = sql.SqliteError("Message", sql_err)
|
||||
assert err.environmental == environmental
|
||||
with pytest.raises(exception):
|
||||
sql.raise_sqlite_error("Message", sql_err)
|
||||
|
||||
def test_logging(self, caplog):
|
||||
sql_err = QSqlError("driver text", "db text", QSqlError.UnknownError,
|
||||
'23')
|
||||
sql.SqliteError("Message", sql_err)
|
||||
with pytest.raises(sql.SqlBugError):
|
||||
sql.raise_sqlite_error("Message", sql_err)
|
||||
|
||||
lines = [r.message for r in caplog.records]
|
||||
expected = ['SQL error:',
|
||||
'type: UnknownError',
|
||||
@ -62,21 +64,11 @@ class TestSqliteError:
|
||||
|
||||
assert lines == expected
|
||||
|
||||
def test_from_query(self):
|
||||
@pytest.mark.parametrize('klass',
|
||||
[sql.SqlEnvironmentError, sql.SqlBugError])
|
||||
def test_text(self, klass):
|
||||
sql_err = QSqlError("driver text", "db text")
|
||||
err = sql.SqliteError.from_query(
|
||||
what='test', query='SELECT * from foo;', error=sql_err)
|
||||
expected = ('Failed to test query "SELECT * from foo;": '
|
||||
'"db text driver text"')
|
||||
assert str(err) == expected
|
||||
|
||||
def test_subclass(self):
|
||||
with pytest.raises(sql.SqlError):
|
||||
raise sql.SqliteError("text", QSqlError())
|
||||
|
||||
def test_text(self):
|
||||
sql_err = QSqlError("driver text", "db text")
|
||||
err = sql.SqliteError("Message", sql_err)
|
||||
err = klass("Message", sql_err)
|
||||
assert err.text() == "db text"
|
||||
|
||||
|
||||
@ -103,7 +95,7 @@ def test_insert_replace(qtbot):
|
||||
table.insert({'name': 'one', 'val': 11, 'lucky': True}, replace=True)
|
||||
assert list(table) == [('one', 11, True)]
|
||||
|
||||
with pytest.raises(sql.SqlError):
|
||||
with pytest.raises(sql.SqlBugError):
|
||||
table.insert({'name': 'one', 'val': 11, 'lucky': True}, replace=False)
|
||||
|
||||
|
||||
@ -139,7 +131,7 @@ def test_insert_batch_replace(qtbot):
|
||||
('one', 11, True),
|
||||
('nine', 19, True)]
|
||||
|
||||
with pytest.raises(sql.SqlError):
|
||||
with pytest.raises(sql.SqlBugError):
|
||||
table.insert_batch({'name': ['one', 'nine'],
|
||||
'val': [11, 19],
|
||||
'lucky': [True, True]})
|
||||
@ -231,3 +223,80 @@ def test_delete_all(qtbot):
|
||||
|
||||
def test_version():
|
||||
assert isinstance(sql.version(), str)
|
||||
|
||||
|
||||
class TestSqlQuery:
|
||||
|
||||
def test_prepare_error(self):
|
||||
with pytest.raises(sql.SqlBugError) as excinfo:
|
||||
sql.Query('invalid')
|
||||
|
||||
expected = ('Failed to prepare query "invalid": "near "invalid": '
|
||||
'syntax error Unable to execute statement"')
|
||||
assert str(excinfo.value) == expected
|
||||
|
||||
@pytest.mark.parametrize('forward_only', [True, False])
|
||||
def test_forward_only(self, forward_only):
|
||||
q = sql.Query('SELECT 0 WHERE 0', forward_only=forward_only)
|
||||
assert q.query.isForwardOnly() == forward_only
|
||||
|
||||
def test_iter_inactive(self):
|
||||
q = sql.Query('SELECT 0')
|
||||
with pytest.raises(sql.SqlBugError,
|
||||
match='Cannot iterate inactive query'):
|
||||
next(iter(q))
|
||||
|
||||
def test_iter_empty(self):
|
||||
q = sql.Query('SELECT 0 AS col WHERE 0')
|
||||
q.run()
|
||||
with pytest.raises(StopIteration):
|
||||
next(iter(q))
|
||||
|
||||
def test_iter(self):
|
||||
q = sql.Query('SELECT 0 AS col')
|
||||
q.run()
|
||||
result = next(iter(q))
|
||||
assert result.col == 0
|
||||
|
||||
def test_iter_multiple(self):
|
||||
q = sql.Query('VALUES (1), (2), (3);')
|
||||
res = list(q.run())
|
||||
assert len(res) == 3
|
||||
assert res[0].column1 == 1
|
||||
|
||||
def test_run_binding(self):
|
||||
q = sql.Query('SELECT :answer')
|
||||
q.run(answer=42)
|
||||
assert q.value() == 42
|
||||
|
||||
def test_run_missing_binding(self):
|
||||
q = sql.Query('SELECT :answer')
|
||||
with pytest.raises(sql.SqlBugError, match='Missing bound values!'):
|
||||
q.run()
|
||||
|
||||
def test_run_batch(self):
|
||||
q = sql.Query('SELECT :answer')
|
||||
q.run_batch(values={'answer': [42]})
|
||||
assert q.value() == 42
|
||||
|
||||
def test_run_batch_missing_binding(self):
|
||||
q = sql.Query('SELECT :answer')
|
||||
with pytest.raises(sql.SqlBugError, match='Missing bound values!'):
|
||||
q.run_batch(values={})
|
||||
|
||||
def test_value_missing(self):
|
||||
q = sql.Query('SELECT 0 WHERE 0')
|
||||
q.run()
|
||||
with pytest.raises(sql.SqlBugError,
|
||||
match='No result for single-result query'):
|
||||
q.value()
|
||||
|
||||
def test_num_rows_affected(self):
|
||||
q = sql.Query('SELECT 0')
|
||||
q.run()
|
||||
assert q.rows_affected() == 0
|
||||
|
||||
def test_bound_values(self):
|
||||
q = sql.Query('SELECT :answer')
|
||||
q.run(answer=42)
|
||||
assert q.bound_values() == {':answer': 42}
|
||||
|
Loading…
Reference in New Issue
Block a user