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

Remove unnecessary class from helpers.server_utils

Browse Source
This commit is contained in:
flan
2017-11-04 00:47:34 +01:00
parent a48ad44a65
commit f5050582ba
3 changed files with 80 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
tests/helpers/server_utils.py
View File

@@ -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.