124 lines
4.4 KiB
Python
124 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import tempfile
|
|
import sqlite3
|
|
import unittest
|
|
import configparser
|
|
|
|
from ankisyncd.sessions import SimpleSessionManager
|
|
from ankisyncd.sessions import SqliteSessionManager
|
|
from ankisyncd.sessions import get_session_manager
|
|
|
|
from ankisyncd.sync_app import SyncUserSession
|
|
|
|
import helpers.server_utils
|
|
|
|
class FakeSessionManager(SimpleSessionManager):
|
|
def __init__(self, config):
|
|
pass
|
|
|
|
class BadSessionManager:
|
|
pass
|
|
|
|
class SessionManagerFactoryTest(unittest.TestCase):
|
|
def test_get_session_manager(self):
|
|
# Get absolute path to development ini file.
|
|
script_dir = os.path.dirname(os.path.realpath(__file__))
|
|
ini_file_path = os.path.join(script_dir,
|
|
"assets",
|
|
"test.conf")
|
|
|
|
# Create temporary files and dirs the server will use.
|
|
server_paths = helpers.server_utils.create_server_paths()
|
|
|
|
config = configparser.ConfigParser()
|
|
config.read(ini_file_path)
|
|
|
|
# Use custom files and dirs in settings. Should be SqliteSessionManager
|
|
config['sync_app'].update(server_paths)
|
|
self.assertTrue(type(get_session_manager(config['sync_app']) == SqliteSessionManager))
|
|
|
|
# No value defaults to SimpleSessionManager
|
|
config.remove_option("sync_app", "session_db_path")
|
|
self.assertTrue(type(get_session_manager(config['sync_app'])) == SimpleSessionManager)
|
|
|
|
# A conf-specified SessionManager is loaded
|
|
config.set("sync_app", "session_manager", 'test_sessions.FakeSessionManager')
|
|
self.assertTrue(type(get_session_manager(config['sync_app'])) == FakeSessionManager)
|
|
|
|
# Should fail at load time if the class doesn't inherit from SimpleSessionManager
|
|
config.set("sync_app", "session_manager", 'test_sessions.BadSessionManager')
|
|
with self.assertRaises(TypeError):
|
|
sm = get_session_manager(config['sync_app'])
|
|
|
|
# Add the session_db_path back, it should take precedence over BadSessionManager
|
|
config['sync_app'].update(server_paths)
|
|
self.assertTrue(type(get_session_manager(config['sync_app'])) == SqliteSessionManager)
|
|
|
|
|
|
class SimpleSessionManagerTest(unittest.TestCase):
|
|
test_hkey = '1234567890'
|
|
sdir = tempfile.mkdtemp(suffix="_session")
|
|
os.rmdir(sdir)
|
|
test_session = SyncUserSession('testName', sdir, None, None)
|
|
|
|
def setUp(self):
|
|
self.sessionManager = SimpleSessionManager()
|
|
|
|
def tearDown(self):
|
|
self.sessionManager = None
|
|
|
|
def test_save(self):
|
|
self.sessionManager.save(self.test_hkey, self.test_session)
|
|
self.assertEqual(self.sessionManager.sessions[self.test_hkey].name,
|
|
self.test_session.name)
|
|
self.assertEqual(self.sessionManager.sessions[self.test_hkey].path,
|
|
self.test_session.path)
|
|
|
|
def test_delete(self):
|
|
self.sessionManager.save(self.test_hkey, self.test_session)
|
|
self.assertTrue(self.test_hkey in self.sessionManager.sessions)
|
|
|
|
self.sessionManager.delete(self.test_hkey)
|
|
|
|
self.assertTrue(self.test_hkey not in self.sessionManager.sessions)
|
|
|
|
def test_load(self):
|
|
self.sessionManager.save(self.test_hkey, self.test_session)
|
|
self.assertTrue(self.test_hkey in self.sessionManager.sessions)
|
|
|
|
loaded_session = self.sessionManager.load(self.test_hkey)
|
|
self.assertEqual(loaded_session.name, self.test_session.name)
|
|
self.assertEqual(loaded_session.path, self.test_session.path)
|
|
|
|
|
|
class SqliteSessionManagerTest(SimpleSessionManagerTest):
|
|
file_descriptor, _test_sess_db_path = tempfile.mkstemp(suffix=".db")
|
|
os.close(file_descriptor)
|
|
os.unlink(_test_sess_db_path)
|
|
|
|
def setUp(self):
|
|
self.sessionManager = SqliteSessionManager(self._test_sess_db_path)
|
|
|
|
def tearDown(self):
|
|
if os.path.exists(self._test_sess_db_path):
|
|
os.remove(self._test_sess_db_path)
|
|
|
|
def test_save(self):
|
|
SimpleSessionManagerTest.test_save(self)
|
|
self.assertTrue(os.path.exists(self._test_sess_db_path))
|
|
|
|
conn = sqlite3.connect(self._test_sess_db_path)
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT username, path FROM session WHERE hkey=?",
|
|
(self.test_hkey,))
|
|
res = cursor.fetchone()
|
|
conn.close()
|
|
|
|
self.assertEqual(res[0], self.test_session.name)
|
|
self.assertEqual(res[1], self.test_session.path)
|
|
|
|
|
|
|