diff --git a/tests/helpers/server_utils.py b/tests/helpers/server_utils.py index 0e4463e..8e1a584 100644 --- a/tests/helpers/server_utils.py +++ b/tests/helpers/server_utils.py @@ -8,91 +8,82 @@ import tempfile from ankisyncd.sync_app import SyncApp, SyncCollectionHandler, SyncMediaHandler -class ServerUtils(object): - def clean_up(self): - shutil.rmtree(self.dir) +def create_server_paths(): + """ + Creates temporary files and dirs for our app to use during tests. + """ + dir = tempfile.mkdtemp(prefix="ServerUtils") + os.mkdir(os.path.join(dir, "data")) - def create_server_paths(self): - """ - Creates temporary files and dirs for our app to use during tests. - """ - dir = tempfile.mkdtemp(prefix="ServerUtils") - self.dir = dir - os.mkdir(os.path.join(dir, "data")) + return { + "auth_db": os.path.join(dir, "auth.db"), + "session_db": os.path.join(dir, "session.db"), + "data_root": os.path.join(dir, "data"), + } - return { - "auth_db": os.path.join(dir, "auth.db"), - "session_db": os.path.join(dir, "session.db"), - "data_root": os.path.join(dir, "data"), - } +def create_sync_app(server_paths, config_path): + config = ConfigParser.SafeConfigParser() + config.read(config_path) - @staticmethod - def create_sync_app(server_paths, config_path): - config = ConfigParser.SafeConfigParser() - config.read(config_path) + # Use custom files and dirs in settings. + config.set("sync_app", "auth_db_path", server_paths["auth_db"]) + config.set("sync_app", "session_db_path", server_paths["session_db"]) + config.set("sync_app", "data_root", server_paths["data_root"]) - # Use custom files and dirs in settings. - config.set("sync_app", "auth_db_path", server_paths["auth_db"]) - config.set("sync_app", "session_db_path", server_paths["session_db"]) - config.set("sync_app", "data_root", server_paths["data_root"]) + return SyncApp(config) - return SyncApp(config) +def get_session_for_hkey(server, hkey): + return server.session_manager.load(hkey) - def get_session_for_hkey(self, server, hkey): - return server.session_manager.load(hkey) +def get_thread_for_hkey(server, hkey): + session = get_session_for_hkey(server, hkey) + thread = session.get_thread() + return thread - def get_thread_for_hkey(self, server, hkey): - session = self.get_session_for_hkey(server, hkey) - thread = session.get_thread() - return thread +def get_col_wrapper_for_hkey(server, hkey): + print("getting col wrapper for hkey " + hkey) + print("all session keys: " + str(server.session_manager.sessions.keys())) + thread = get_thread_for_hkey(server, hkey) + col_wrapper = thread.wrapper + return col_wrapper - def get_col_wrapper_for_hkey(self, server, hkey): - print("getting col wrapper for hkey " + hkey) - print("all session keys: " + str(server.session_manager.sessions.keys())) - thread = self.get_thread_for_hkey(server, hkey) - col_wrapper = thread.wrapper - return col_wrapper +def get_col_for_hkey(server, hkey): + col_wrapper = get_col_wrapper_for_hkey(server, hkey) + col_wrapper.open() # Make sure the col is opened. + return col_wrapper._CollectionWrapper__col - def get_col_for_hkey(self, server, hkey): - col_wrapper = self.get_col_wrapper_for_hkey(server, hkey) - col_wrapper.open() # Make sure the col is opened. - return col_wrapper._CollectionWrapper__col +def get_col_db_path_for_hkey(server, hkey): + col = get_col_for_hkey(server, hkey) + return col.db._path - def get_col_db_path_for_hkey(self, server, hkey): - col = self.get_col_for_hkey(server, hkey) - return col.db._path +def get_syncer_for_hkey(server, hkey, syncer_type='collection'): + col = get_col_for_hkey(server, hkey) - def get_syncer_for_hkey(self, server, hkey, syncer_type='collection'): - col = self.get_col_for_hkey(server, hkey) + session = get_session_for_hkey(server, hkey) - session = self.get_session_for_hkey(server, hkey) + syncer_type = syncer_type.lower() + if syncer_type == 'collection': + handler_method = SyncCollectionHandler.operations[0] + elif syncer_type == 'media': + handler_method = SyncMediaHandler.operations[0] - syncer_type = syncer_type.lower() - if syncer_type == 'collection': - handler_method = SyncCollectionHandler.operations[0] - elif syncer_type == 'media': - handler_method = SyncMediaHandler.operations[0] + return session.get_handler_for_operation(handler_method, col) - return session.get_handler_for_operation(handler_method, col) +def add_files_to_mediasyncer(media_syncer, filepaths, + update_db=False, bump_last_usn=False): + """ + If bumpLastUsn is True, the media syncer's lastUsn will be incremented + once for each added file. Use this when adding files to the server. + """ - def add_files_to_mediasyncer(self, - media_syncer, - filepaths, - update_db=False, - bump_last_usn=False): - """ - If bumpLastUsn is True, the media syncer's lastUsn will be incremented - once for each added file. Use this when adding files to the server. - """ + for filepath in filepaths: + logging.debug("Adding file '{}' to mediaSyncer".format(filepath)) + # Import file into media dir. + media_syncer.col.media.addFile(filepath) + if bump_last_usn: + # Need to bump lastUsn once for each file. + media_manager = media_syncer.col.media + media_manager.setLastUsn(media_syncer.col.media.lastUsn() + 1) - for filepath in filepaths: - logging.debug("Adding file '{}' to mediaSyncer".format(filepath)) - # Import file into media dir. - media_syncer.col.media.addFile(filepath) - if bump_last_usn: - # Need to bump lastUsn once for each file. - media_manager = media_syncer.col.media - media_manager.setLastUsn(media_syncer.col.media.lastUsn() + 1) - - if update_db: - media_syncer.col.media.findChanges() # Write changes to db. + if update_db: + media_syncer.col.media.findChanges() # Write changes to db. diff --git a/tests/sync_app_functional_test_base.py b/tests/sync_app_functional_test_base.py index ca32a70..a985d2e 100644 --- a/tests/sync_app_functional_test_base.py +++ b/tests/sync_app_functional_test_base.py @@ -3,11 +3,11 @@ import os import unittest from webtest import TestApp +import helpers.server_utils from ankisyncd.users import SqliteUserManager from helpers.collection_utils import CollectionUtils from helpers.mock_servers import MockRemoteServer from helpers.monkey_patches import monkeypatch_db, unpatch_db -from helpers.server_utils import ServerUtils class SyncAppFunctionalTestBase(unittest.TestCase): @@ -15,21 +15,17 @@ class SyncAppFunctionalTestBase(unittest.TestCase): @classmethod def setUpClass(cls): cls.colutils = CollectionUtils() - cls.serverutils = ServerUtils() @classmethod def tearDownClass(cls): cls.colutils.clean_up() cls.colutils = None - cls.serverutils.clean_up() - cls.serverutils = None - def setUp(self): monkeypatch_db() # Create temporary files and dirs the server will use. - self.server_paths = self.serverutils.create_server_paths() + self.server_paths = helpers.server_utils.create_server_paths() # Add a test user to the temp auth db the server will use. self.user_manager = SqliteUserManager(self.server_paths['auth_db'], @@ -44,7 +40,7 @@ class SyncAppFunctionalTestBase(unittest.TestCase): # Create SyncApp instance using the dev ini file and the temporary # paths. - self.server_app = self.serverutils.create_sync_app(self.server_paths, + self.server_app = helpers.server_utils.create_sync_app(self.server_paths, ini_file_path) # Wrap the SyncApp object in TestApp instance for testing. diff --git a/tests/test_web_media.py b/tests/test_web_media.py index a39afaa..3311040 100644 --- a/tests/test_web_media.py +++ b/tests/test_web_media.py @@ -6,6 +6,7 @@ import os import shutil import helpers.file_utils +import helpers.server_utils import helpers.db_utils from anki.sync import MediaSyncer from helpers.mock_servers import MockRemoteMediaServer @@ -94,7 +95,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): the identical file in their media directories and media databases. """ client = self.client_syncer - server = self.serverutils.get_syncer_for_hkey(self.server_app, + server = helpers.server_utils.get_syncer_for_hkey(self.server_app, self.hkey, 'media') @@ -102,7 +103,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): temp_file_path = helpers.file_utils.create_named_file(u"foo.jpg", "hello") # Add the test file to the server's collection. - self.serverutils.add_files_to_mediasyncer(server, + helpers.server_utils.add_files_to_mediasyncer(server, [temp_file_path], update_db=True, bump_last_usn=True) @@ -126,7 +127,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): """ join = os.path.join client = self.client_syncer - server = self.serverutils.get_syncer_for_hkey(self.server_app, + server = helpers.server_utils.get_syncer_for_hkey(self.server_app, self.hkey, 'media') @@ -134,7 +135,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): temp_file_path = helpers.file_utils.create_named_file(u"foo.jpg", "hello") # Add the test file to the client's media collection. - self.serverutils.add_files_to_mediasyncer(client, + helpers.server_utils.add_files_to_mediasyncer(client, [temp_file_path], update_db=True, bump_last_usn=False) @@ -165,7 +166,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): join = os.path.join isfile = os.path.isfile client = self.client_syncer - server = self.serverutils.get_syncer_for_hkey(self.server_app, + server = helpers.server_utils.get_syncer_for_hkey(self.server_app, self.hkey, 'media') @@ -173,10 +174,10 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): file_for_client = helpers.file_utils.create_named_file(u"foo.jpg", "hello") file_for_server = helpers.file_utils.create_named_file(u"bar.jpg", "goodbye") - self.serverutils.add_files_to_mediasyncer(client, + helpers.server_utils.add_files_to_mediasyncer(client, [file_for_client], update_db=True) - self.serverutils.add_files_to_mediasyncer(server, + helpers.server_utils.add_files_to_mediasyncer(server, [file_for_server], update_db=True, bump_last_usn=True) @@ -212,7 +213,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): join = os.path.join isfile = os.path.isfile client = self.client_syncer - server = self.serverutils.get_syncer_for_hkey(self.server_app, + server = helpers.server_utils.get_syncer_for_hkey(self.server_app, self.hkey, 'media') @@ -221,10 +222,10 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): file_for_client = helpers.file_utils.create_named_file(u"foo.jpg", "hello") file_for_server = helpers.file_utils.create_named_file(u"foo.jpg", "goodbye") - self.serverutils.add_files_to_mediasyncer(client, + helpers.server_utils.add_files_to_mediasyncer(client, [file_for_client], update_db=True) - self.serverutils.add_files_to_mediasyncer(server, + helpers.server_utils.add_files_to_mediasyncer(server, [file_for_server], update_db=True, bump_last_usn=True) @@ -258,7 +259,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): join = os.path.join isfile = os.path.isfile client = self.client_syncer - server = self.serverutils.get_syncer_for_hkey(self.server_app, + server = helpers.server_utils.get_syncer_for_hkey(self.server_app, self.hkey, 'media') @@ -266,7 +267,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): temp_file_path = helpers.file_utils.create_named_file(u"foo.jpg", "hello") # Add the test file to client's media collection. - self.serverutils.add_files_to_mediasyncer(client, + helpers.server_utils.add_files_to_mediasyncer(client, [temp_file_path], update_db=True, bump_last_usn=False) @@ -313,7 +314,7 @@ class SyncAppFunctionalMediaTest(SyncAppFunctionalTestBase): # findChanges(), only during syncs. support_file = helpers.file_utils.get_asset_path(u'blue.jpg') self.assertTrue(os.path.isfile(support_file)) - self.serverutils.add_files_to_mediasyncer(client, + helpers.server_utils.add_files_to_mediasyncer(client, [support_file], update_db=False)