prehistoric-systems/beeref
1
0
Fork 0
mirror of https://github.com/rbreu/beeref.git synced 2026-03-11 08:54:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Error handling for reading/writing bee files

Browse source
This commit is contained in:
Rebecca Breu 2021-03-27 21:55:47 +01:00
parent e42ed2d598
commit aa75b30094
7 changed files with 214 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
beeref/__main__.py
View file

@ -37,13 +37,13 @@ class BeeRefMainWindow(QtWidgets.QWidget):
logo = os.path.join(root, 'assets', 'logo.png')
logger.debug(f'Loading icon {logo}')
self.setWindowIcon(QtGui.QIcon(logo))
view = BeeGraphicsView(app, self, filename)
layout = QtWidgets.QVBoxLayout()
layout.setContentsMargins(QtCore.QMargins(0, 0, 0, 0))
layout.addWidget(view)
self.setLayout(layout)
self.resize(500, 300)
self.show()
view = BeeGraphicsView(app, self, filename)
layout.addWidget(view)
def safe_timer(timeout, func, *args, **kwargs):

5
beeref/fileio/__init__.py
View file

@ -15,16 +15,17 @@
import logging
from beeref.fileio.errors import BeeFileIOError
from beeref.fileio.sql import SQLiteIO
__all__ = ['load', 'save']
__all__ = ['load', 'save', 'BeeFileIOError']
logger = logging.getLogger('BeeRef')
def load(filename, scene):
logger.info(f'Loading from file {filename}...')
io = SQLiteIO(filename, scene)
io = SQLiteIO(filename, scene, readonly=True)
return io.read()

19
beeref/fileio/errors.py Normal file
View file

@ -0,0 +1,19 @@
# This file is part of BeeRef.
#
# BeeRef is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# BeeRef is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with BeeRef. If not, see <https://www.gnu.org/licenses/>.
class BeeFileIOError(Exception):
def __init__(self, msg, filename):
self.msg = msg
self.filename = filename

77
beeref/fileio/sql.py
View file

@ -23,28 +23,75 @@ https://www.sqlite.org/appfileformat.html
https://www.sqlite.org/sqlar.html
"""
import logging
import os
import sqlite3
from PyQt6 import QtGui
from beeref.items import BeePixmapItem
from .errors import BeeFileIOError
from .schema import SCHEMA
logger = logging.getLogger('BeeRef')
def handle_sqlite_errors(func):
def wrapper(self, *args, **kwargs):
try:
func(self, *args, **kwargs)
except sqlite3.Error as e:
logger.exception(f'Error while reading/writing {self.filename}')
raise BeeFileIOError(msg=str(e), filename=self.filename) from e
return wrapper
class SQLiteIO:
USER_VERSION = 1
APPLICATION_ID = 2060242126
def __init__(self, filename, scene, create_new=False):
def __init__(self, filename, scene, create_new=False, readonly=False):
self.scene = scene
self.filename = filename
self.create_new = create_new
# TBD: error handling (different for existing files/new files...)
if self.create_new and os.path.exists(self.filename):
self.filename = filename
self.readonly = readonly
def __del__(self):
self._close_connection()
def _close_connection(self):
if hasattr(self, '_connection'):
self._connection.close()
delattr(self, '_connection')
if hasattr(self, '_cursor'):
delattr(self, '_cursor')
def _establish_connection(self):
if (self.create_new
and not self.readonly
and os.path.exists(self.filename)):
os.remove(self.filename)
self.connection = sqlite3.connect(self.filename)
self.cursor = self.connection.cursor()
if self.readonly:
self._connection = sqlite3.connect(
f'file:{self.filename}?mode=ro')
else:
self._connection = sqlite3.connect(self.filename)
self._cursor = self.connection.cursor()
@property
def connection(self):
if not hasattr(self, '_connection'):
self._establish_connection()
return self._connection
@property
def cursor(self):
if not hasattr(self, '_cursor'):
self._establish_connection()
return self._cursor
def ex(self, *args, **kwargs):
return self.cursor.execute(*args, **kwargs)
@ -70,6 +117,7 @@ class SQLiteIO:
for schema in SCHEMA:
self.ex(schema)
@handle_sqlite_errors
def read(self):
rows = self.fetchall(
'SELECT pos_x, pos_y, scale, filename, sqlar.data, items.id '
@ -83,10 +131,23 @@ class SQLiteIO:
item.setScale(row[2])
self.scene.addItem(item)
@handle_sqlite_errors
def write(self):
self.write_meta()
self.create_schema_on_new()
try:
self.write_meta()
self.create_schema_on_new()
self.write_data()
except sqlite3.Error:
if self.create_new:
# If writing to a new file fails, we can't recover
raise
else:
# Updating a file failed; try creating it from scratch instead
self.create_new = True
self._close_connection()
self.write()
def write_data(self):
to_delete = self.fetchall('SELECT id from ITEMS')
for item in self.scene.items_for_save():
if item.save_id and not self.create_new:

22
beeref/view.py
View file

@ -249,8 +249,15 @@ class BeeGraphicsView(QtWidgets.QGraphicsView):
logger.info(f'Opening file {filename}')
self.scene.clear()
self.undo_stack.clear()
fileio.load(filename, self.scene)
self.filename = filename
try:
fileio.load(filename, self.scene)
self.filename = filename
except fileio.BeeFileIOError:
QtWidgets.QMessageBox.warning(
self,
'Problem loading file',
('<p>Problem loading file %s</p>'
'<p>Not accessible or not a proper bee file</p>') % filename)
def on_action_open(self):
filename, f = QtWidgets.QFileDialog.getOpenFileName(
@ -269,8 +276,15 @@ class BeeGraphicsView(QtWidgets.QGraphicsView):
if filename:
if not filename.endswith('.bee'):
filename = f'{filename}.bee'
fileio.save(filename, self.scene, create_new=True)
self.filename = filename
try:
fileio.save(filename, self.scene, create_new=True)
self.filename = filename
except fileio.BeeFileIOError:
QtWidgets.QMessageBox.warning(
self,
'Problem saving file',
('<p>Problem saving file %s</p>'
'<p>File/directory not accessible</p>') % filename)
def on_action_save(self):
if not self.filename:

21
tests/fileio/test_init.py Normal file
View file

@ -0,0 +1,21 @@
import os.path
import tempfile
from unittest.mock import patch
from beeref import fileio
@patch('beeref.fileio.sql.SQLiteIO.write')
def test_save_create_new_false(write_mock):
with tempfile.TemporaryDirectory() as dirname:
fname = os.path.join(dirname, 'test.bee')
fileio.save(fname, 'myscene', create_new=False)
write_mock.assert_called_once()
@patch('beeref.fileio.sql.SQLiteIO.read')
def test_write(read_mock):
with tempfile.TemporaryDirectory() as dirname:
fname = os.path.join(dirname, 'test.bee')
fileio.load(fname, 'myscene')
read_mock.assert_called_once()

104
tests/fileio/test_sql.py
View file

@ -1,8 +1,11 @@
import os.path
import tempfile
from unittest.mock import MagicMock, patch
from PyQt6 import QtGui
import pytest
from beeref.fileio.errors import BeeFileIOError
from beeref.fileio.sql import SQLiteIO
from beeref.items import BeePixmapItem
from beeref.scene import BeeGraphicsScene
@ -44,6 +47,21 @@ class SQLiteIOTestCase(BeeTestCase):
'WHERE type="table" AND name NOT LIKE "sqlite_%"')
assert result[0] == 0
def test_readonly_doesnt_allow_write(self):
scene = BeeGraphicsScene(None)
with tempfile.TemporaryDirectory() as dirname:
fname = os.path.join(dirname, 'test.bee')
with open(fname, 'w') as f:
f.write('foobar')
io = SQLiteIO(fname, scene, readonly=True)
with pytest.raises(BeeFileIOError) as exinfo:
io.write()
assert exinfo.value.filename == fname
with open(fname, 'r') as f:
f.read() == 'foobar'
class SQLiteIOWriteTestCase(BeeTestCase):
@ -126,32 +144,74 @@ class SQLiteIOWriteTestCase(BeeTestCase):
assert self.io.fetchone('SELECT COUNT(*) from items') == (0,)
assert self.io.fetchone('SELECT COUNT(*) from sqlar') == (0,)
def test_update_recovers_from_borked_file(self):
item = BeePixmapItem(QtGui.QImage(), filename='bee.png')
self.scene.addItem(item)
class SQLiteIOLOadTestCase(BeeTestCase):
with tempfile.TemporaryDirectory() as dirname:
fname = os.path.join(dirname, 'test.bee')
with open(fname, 'w') as f:
f.write('foobar')
io = SQLiteIO(fname, self.scene, create_new=False)
io.write()
result = io.fetchone('SELECT COUNT(*) FROM items')
assert result[0] == 1
class SQLiteIOReadTestCase(BeeTestCase):
def setUp(self):
self.scene = BeeGraphicsScene(None)
self.io = SQLiteIO(':memory:', self.scene, create_new=True)
def test_loads(self):
def test_reads_readonly(self):
root = os.path.dirname(__file__)
filename = os.path.join(root, '..', 'assets', 'test3x3.png')
with open(filename, 'rb') as f:
imgfilename = os.path.join(root, '..', 'assets', 'test3x3.png')
with open(imgfilename, 'rb') as f:
imgdata = f.read()
self.io.create_schema_on_new()
self.io.ex(
'INSERT INTO items (type, pos_x, pos_y, scale, filename) '
'VALUES (?, ?, ?, ?, ?) ',
('pixmap', 22.2, 33.3, 3.4, 'bee.png'))
self.io.ex('INSERT INTO sqlar (item_id, data) VALUES (?, ?)',
(1, imgdata))
self.io.read()
assert len(self.scene.items()) == 1
item = self.scene.items()[0]
assert item.save_id == 1
assert item.pos().x() == 22.2
assert item.pos().y() == 33.3
assert item.scale_factor == 3.4
assert item.filename == 'bee.png'
assert item.width == 3
assert item.height == 3
with tempfile.TemporaryDirectory() as dirname:
fname = os.path.join(dirname, 'test.bee')
io = SQLiteIO(fname, self.scene, create_new=True)
io.create_schema_on_new()
io.ex('INSERT INTO items (type, pos_x, pos_y, scale, filename) '
'VALUES (?, ?, ?, ?, ?) ',
('pixmap', 22.2, 33.3, 3.4, 'bee.png'))
io.ex('INSERT INTO sqlar (item_id, data) VALUES (?, ?)',
(1, imgdata))
io.connection.commit()
del(io)
io = SQLiteIO(fname, self.scene, readonly=True)
io.read()
assert len(self.scene.items()) == 1
item = self.scene.items()[0]
assert item.save_id == 1
assert item.pos().x() == 22.2
assert item.pos().y() == 33.3
assert item.scale_factor == 3.4
assert item.filename == 'bee.png'
assert item.width == 3
assert item.height == 3
def test_raises_error_when_file_borked(self):
with tempfile.TemporaryDirectory() as dirname:
fname = os.path.join(dirname, 'test.bee')
with open(fname, 'w') as f:
f.write('foobar')
io = SQLiteIO(fname, self.scene, readonly=True)
with pytest.raises(BeeFileIOError) as exinfo:
io.read()
assert exinfo.value.filename == fname
def test_reads_raises_error_when_file_empty(self):
with tempfile.TemporaryDirectory() as dirname:
fname = os.path.join(dirname, 'test.bee')
io = SQLiteIO(fname, self.scene, readonly=True)
with pytest.raises(BeeFileIOError) as exinfo:
io.read()
assert exinfo.value.filename == fname
# should not create a file on reading!
assert os.path.isfile(fname) is False