ouczbs/anki-sync-server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Make DB utils more general

Browse Source
This commit is contained in:
flan 2017-11-01 04:04:48 +01:00
parent 55bdbfacaa
commit ce3aa4a685
4 changed files with 98 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
tests/helpers/__init__.py Normal file
View File

@ -0,0 +1 @@
import db_utils

156
tests/helpers/db_utils.py
View File

@ -3,123 +3,69 @@ import os
import sqlite3
import subprocess
from helpers.file_utils import FileUtils
def from_sql(path, sql):
"""
Creates an SQLite db and executes the passed sql statements on it.
:param path: the path to the created db file
:param sql: the sql statements to execute on the newly created db
"""
connection = sqlite3.connect(path)
cursor = connection.cursor()
cursor.executescript(sql)
connection.commit()
connection.close()
class DBUtils(object):
"""Provides methods for creating and comparing sqlite databases."""
def to_sql(database):
"""
Returns a string containing the sql export of the database. Used for
debugging.
def __init__(self):
self.fileutils = FileUtils()
:param database: either the path to the SQLite db file or an open
connection to it
:return: a string representing the sql export of the database
"""
def clean_up(self):
self.fileutils.clean_up()
if type(database) == str:
connection = sqlite3.connect(database)
else:
connection = database
def create_sqlite_db_with_sql(self, sql_string):
"""
Creates an SQLite db and executes the passed sql statements on it.
res = '\n'.join(connection.iterdump())
:param sql_string: the sql statements to execute on the newly created
db
:return: the path to the created db file
"""
db_path = self.fileutils.create_file_path(suffix=".anki2")
connection = sqlite3.connect(db_path)
cursor = connection.cursor()
cursor.executescript(sql_string)
connection.commit()
if type(database) == str:
connection.close()
return db_path
return res
@staticmethod
def sqlite_db_to_sql_string(database):
"""
Returns a string containing the sql export of the database. Used for
debugging.
:param database: either the path to the SQLite db file or an open
connection to it
:return: a string representing the sql export of the database
"""
def diff(left_db_path, right_db_path):
"""
Uses the sqldiff cli tool to compare two sqlite files for equality.
Returns True if the databases differ, False if they don't.
if type(database) == str:
connection = sqlite3.connect(database)
else:
connection = database
:param left_db_path: path to the left db file
:param right_db_path: path to the right db file
:return: True if the specified databases differ, False else
"""
res = '\n'.join(connection.iterdump())
command = ["/bin/sqldiff", left_db_path, right_db_path]
if type(database) == str:
connection.close()
child_process = subprocess.Popen(command,
shell=False,
stdout=subprocess.PIPE)
stdout, stderr = child_process.communicate()
exit_code = child_process.returncode
return res
if exit_code != 0 or stderr is not None:
raise RuntimeError("Command {} encountered an error, exit "
"code: {}, stderr: {}"
.format(" ".join(command),
exit_code,
stderr))
def media_dbs_differ(self, left_db_path, right_db_path, compare_timestamps=False):
"""
Compares two media sqlite database files for equality. mtime and dirMod
timestamps are not considered when comparing.
:param left_db_path: path to the left db file
:param right_db_path: path to the right db file
:param compare_timestamps: flag determining if timestamp values
(media.mtime and meta.dirMod) are included
in the comparison
:return: True if the specified databases differ, False else
"""
if not os.path.isfile(left_db_path):
raise IOError("file '" + left_db_path + "' does not exist")
elif not os.path.isfile(right_db_path):
raise IOError("file '" + right_db_path + "' does not exist")
# Create temporary copies of the files to act on.
left_db_path = self.fileutils.create_file_copy(left_db_path)
right_db_path = self.fileutils.create_file_copy(right_db_path)
if not compare_timestamps:
# Set all timestamps that are not NULL to 0.
for dbPath in [left_db_path, right_db_path]:
connection = sqlite3.connect(dbPath)
connection.execute("""UPDATE media SET mtime=0
WHERE mtime IS NOT NULL""")
connection.execute("""UPDATE meta SET dirMod=0
WHERE rowid=1""")
connection.commit()
connection.close()
return self.__sqlite_dbs_differ(left_db_path, right_db_path)
def __sqlite_dbs_differ(self, left_db_path, right_db_path):
"""
Uses the sqldiff cli tool to compare two sqlite files for equality.
Returns True if the databases differ, False if they don't.
:param left_db_path: path to the left db file
:param right_db_path: path to the right db file
:return: True if the specified databases differ, False else
"""
command = ["/bin/sqldiff", left_db_path, right_db_path]
try:
child_process = subprocess.Popen(command,
shell=False,
stdout=subprocess.PIPE)
stdout, stderr = child_process.communicate()
exit_code = child_process.returncode
if exit_code != 0 or stderr is not None:
raise RuntimeError("Command {} encountered an error, exit "
"code: {}, stderr: {}"
.format(" ".join(command),
exit_code,
stderr))
# Any output from sqldiff means the databases differ.
return stdout != ""
except OSError as err:
raise err
# Any output from sqldiff means the databases differ.
return stdout != ""

5
tests/sync_app_functional_test_base.py
View File

@ -5,7 +5,6 @@ from webtest import TestApp
from ankisyncd.users import SqliteUserManager
from helpers.collection_utils import CollectionUtils
from helpers.db_utils import DBUtils
from helpers.file_utils import FileUtils
from helpers.mock_servers import MockRemoteServer
from helpers.monkey_patches import monkeypatch_db, unpatch_db
@ -19,7 +18,6 @@ class SyncAppFunctionalTestBase(unittest.TestCase):
cls.fileutils = FileUtils()
cls.colutils = CollectionUtils()
cls.serverutils = ServerUtils()
cls.dbutils = DBUtils()
@classmethod
def tearDownClass(cls):
@ -32,9 +30,6 @@ class SyncAppFunctionalTestBase(unittest.TestCase):
cls.serverutils.clean_up()
cls.serverutils = None
cls.dbutils.clean_up()
cls.dbutils = None
def setUp(self):
monkeypatch_db()

51
tests/test_web_media.py
View File

@ -1,7 +1,10 @@
# -*- coding: utf-8 -*-
import tempfile
import filecmp
import sqlite3
import os
import helpers.db_utils
from anki.sync import MediaSyncer
from helpers.mock_servers import MockRemoteMediaServer
from helpers.monkey_patches import monkeypatch_mediamanager, unpatch_mediamanager
@ -34,6 +37,43 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase):
server=mock_remote_server)
return media_syncer
def media_dbs_differ(self, left_db_path, right_db_path, compare_timestamps=False):
"""
Compares two media sqlite database files for equality. mtime and dirMod
timestamps are not considered when comparing.
:param left_db_path: path to the left db file
:param right_db_path: path to the right db file
:param compare_timestamps: flag determining if timestamp values
(media.mtime and meta.dirMod) are included
in the comparison
:return: True if the specified databases differ, False else
"""
if not os.path.isfile(right_db_path):
raise IOError("file '" + left_db_path + "' does not exist")
elif not os.path.isfile(right_db_path):
raise IOError("file '" + right_db_path + "' does not exist")
# Create temporary copies of the files to act on.
left_db_path = self.fileutils.create_file_copy(left_db_path)
right_db_path = self.fileutils.create_file_copy(right_db_path)
if not compare_timestamps:
# Set all timestamps that are not NULL to 0.
for dbPath in [left_db_path, right_db_path]:
connection = sqlite3.connect(dbPath)
connection.execute("""UPDATE media SET mtime=0
WHERE mtime IS NOT NULL""")
connection.execute("""UPDATE meta SET dirMod=0
WHERE rowid=1""")
connection.commit()
connection.close()
return helpers.db_utils.diff(left_db_path, right_db_path)
def test_sync_empty_media_dbs(self):
# With both the client and the server having no media to sync,
# syncing should change nothing.
@ -105,8 +145,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase):
# Except for timestamps, the media databases of client and server
# should be identical.
self.assertFalse(
self.dbutils.media_dbs_differ(client.col.media.db._path,
server.col.media.db._path)
self.media_dbs_differ(client.col.media.db._path, server.col.media.db._path)
)
def test_sync_different_files(self):
@ -301,11 +340,13 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase):
CREATE INDEX idx_media_dirty on media (dirty);
""" % chksum)
temp_db_path = self.dbutils.create_sqlite_db_with_sql(sql)
_, dbpath = tempfile.mkstemp(suffix=".anki2")
helpers.db_utils.from_sql(dbpath, sql)
# Except for timestamps, the client's db after sync should be identical
# to the expected data.
self.assertFalse(self.dbutils.media_dbs_differ(
self.assertFalse(self.media_dbs_differ(
client.col.media.db._path,
temp_db_path
dbpath
))
os.unlink(dbpath)