manage session

This commit is contained in:
Emmanuel Garette 2018-09-11 20:11:13 +02:00
parent 4e5ffb201d
commit c80adedc02
4 changed files with 304 additions and 299 deletions

View file

@ -17,286 +17,286 @@ def test_non_persistent():
Config(o, session_id='test_non_persistent') Config(o, session_id='test_non_persistent')
#def test_list(): def test_list():
# b = BoolOption('b', '') b = BoolOption('b', '')
# o = OptionDescription('od', '', [b]) o = OptionDescription('od', '', [b])
# c = Config(o, session_id='test_non_persistent') c = Config(o, session_id='test_non_persistent')
# c.cfgimpl_get_settings().remove('cache') c.option('b').value.set(True)
# c.b = True assert 'test_non_persistent' in list_sessions('config')
# assert 'test_non_persistent' in list_sessions('config') del(c)
# del(c) assert 'test_non_persistent' not in list_sessions('config')
# assert 'test_non_persistent' not in list_sessions('config')
def test_create_persistent():
b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
Config(o, session_id='test_persistent', persistent=True)
delete_session('test_persistent')
except ValueError:
# storage is not persistent
pass
def test_create_delete_not_persistent():
b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
Config(o, session_id='test_persistent', persistent=True)
delete_session('test_persistent')
except ValueError:
raises(ValueError, "delete_session('test_persistent')")
def test_list_sessions_persistent():
b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
c = Config(o, session_id='test_persistent', persistent=True)
c.option('b').value.set(True)
except ValueError:
# storage is not persistent
pass
else:
assert 'test_persistent' in list_sessions('config')
delete_session('test_persistent')
def test_delete_session_persistent():
b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
Config(o, session_id='test_persistent', persistent=True)
except ValueError:
# storage is not persistent
pass
else:
assert 'test_persistent' in list_sessions('config')
delete_session('test_persistent')
assert 'test_persistent' not in list_sessions('config')
def test_create_persistent_retrieve():
b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
c = Config(o, session_id='test_persistent', persistent=True)
except ValueError:
# storage is not persistent
pass
else:
assert c.option('b').value.get() is None
c.option('b').value.set(True)
assert c.option('b').value.get() is True
del c
c = Config(o, session_id='test_persistent', persistent=True)
assert c.option('b').value.get() is True
assert 'test_persistent' in list_sessions('config')
delete_session(c.config.name())
del c
c = Config(o, session_id='test_persistent', persistent=True)
assert c.option('b').value.get() is None
delete_session(c.config.name())
del c
def test_two_persistent():
b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
c = Config(o, session_id='test_persistent', persistent=True)
except ValueError:
# storage is not persistent
pass
else:
c2 = Config(o, session_id='test_persistent', persistent=True)
c2.property.pop('cache')
assert c.option('b').value.get() is None
assert c2.option('b').value.get() is None
# #
c.option('b').value.set(False)
assert c.option('b').value.get() is False
assert c2.option('b').value.get() is False
# #
#def test_create_persistent(): c.option('b').value.set(True)
# b = BoolOption('b', '') assert c.option('b').value.get() is True
# o = OptionDescription('od', '', [b]) assert c2.option('b').value.get() is True
# try: delete_session('test_persistent')
# Config(o, session_id='test_persistent', persistent=True)
# except ValueError:
# # storage is not persistent def test_create_persistent_retrieve_owner():
# pass b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
c = Config(o, session_id='test_persistent', persistent=True)
except ValueError:
# storage is not persistent
pass
else:
assert c.option('b').owner.isdefault()
c.option('b').value.set(True)
assert c.option('b').value.get()
assert c.option('b').owner.get() == 'user'
##owners.addowner('persistentowner')
c.option('b').owner.set('persistentowner')
assert c.option('b').owner.get() == 'persistentowner'
del c
# #
c = Config(o, session_id='test_persistent', persistent=True)
c.option('b').owner.set('persistentowner')
delete_session(c.config.name())
del c
# #
#def test_create_delete_not_persistent(): c = Config(o, session_id='test_persistent', persistent=True)
# b = BoolOption('b', '') assert c.option('b').value.get() is None
# o = OptionDescription('od', '', [b]) assert c.option('b').owner.isdefault()
# try: delete_session(c.config.name())
# Config(o, session_id='test_persistent', persistent=True) del c
# except ValueError:
# raises(ValueError, "delete_session('option', 'test_persistent')")
# def test_create_persistent_retrieve_owner_masterslaves():
# a = BoolOption('a', '', multi=True)
#def test_list_sessions_persistent(): b = BoolOption('b', '', multi=True)
# b = BoolOption('b', '') o = MasterSlaves('a', '', [a, b])
# o = OptionDescription('od', '', [b]) #o.impl_set_group_type(groups.master)
# try: o1 = OptionDescription('a', '', [o])
# c = Config(o, session_id='test_persistent', persistent=True) try:
# c.b = True c = Config(o1, session_id='test_persistent', persistent=True)
# except ValueError: except ValueError:
# # storage is not persistent # storage is not persistent
# pass pass
# else: else:
# assert 'test_persistent' in list_sessions('config') assert c.option('a.a').owner.isdefault()
# c.option('a.a').value.set([True, False])
# c.option('a.b', 1).value.set(True)
#def test_delete_session_persistent(): assert c.option('a.a').owner.get() == 'user'
# b = BoolOption('b', '') assert c.option('a.b', 0).owner.isdefault()
# o = OptionDescription('od', '', [b]) assert c.option('a.b', 1).owner.get() == 'user'
# try:
# Config(o, session_id='test_persistent', persistent=True)
# except ValueError:
# # storage is not persistent
# pass
# else:
# assert 'test_persistent' in list_sessions('config')
# delete_session('config', 'test_persistent')
# assert 'test_persistent' not in list_sessions('config')
#
#
#def test_create_persistent_retrieve():
# b = BoolOption('b', '')
# o = OptionDescription('od', '', [b])
# try:
# c = Config(o, session_id='test_persistent', persistent=True)
# except ValueError:
# # storage is not persistent
# pass
# else:
# assert c.b is None
# c.b = True
# assert c.b is True
# del(c)
# c = Config(o, session_id='test_persistent', persistent=True)
# assert c.b is True
# assert 'test_persistent' in list_sessions('config')
# delete_session('config', c.impl_getsessionid())
# del(c)
# c = Config(o, session_id='test_persistent', persistent=True)
# assert c.b is None
# delete_session('config', c.impl_getsessionid())
# del(c)
#
#
#def test_two_persistent():
# b = BoolOption('b', '')
# o = OptionDescription('od', '', [b])
# try:
# c = Config(o, session_id='test_persistent', persistent=True)
# except ValueError:
# # storage is not persistent
# pass
# else:
# c.cfgimpl_get_settings().remove('cache')
# c2 = Config(o, session_id='test_persistent', persistent=True)
# c2.cfgimpl_get_settings().remove('cache')
# assert c.b is None
# assert c2.b is None
# c.b = False
# assert c.b is False
# assert c2.b is False
# c2.b = True
# assert c.b is True
# assert c2.b is True
# delete_session('config', 'test_persistent')
#
#
#def test_create_persistent_retrieve_owner():
# b = BoolOption('b', '')
# o = OptionDescription('od', '', [b])
# try:
# c = Config(o, session_id='test_persistent', persistent=True)
# except ValueError:
# # storage is not persistent
# pass
# else:
# assert c.getowner(b) == owners.default
# c.b = True
# assert c.b is True
# assert c.getowner(b) == owners.user
# owners.addowner('persistentowner')
# c.cfgimpl_get_values().setowner(b, owners.persistentowner)
# assert c.getowner(b) == owners.persistentowner
# del(c)
# #
# c = Config(o, session_id='test_persistent', persistent=True)
# c.cfgimpl_get_values().setowner(b, owners.persistentowner)
# delete_session('config', c.impl_getsessionid())
# del(c)
# #
# c = Config(o, session_id='test_persistent', persistent=True)
# assert c.b is None
# assert c.getowner(b) == owners.default
# delete_session('config', c.impl_getsessionid())
# del(c)
#
#
#def test_create_persistent_retrieve_owner_masterslaves():
# a = BoolOption('a', '', multi=True)
# b = BoolOption('b', '', multi=True)
# o = MasterSlaves('a', '', [a, b])
# #o.impl_set_group_type(groups.master)
# o1 = OptionDescription('a', '', [o])
# try:
# c = Config(o1, session_id='test_persistent', persistent=True)
# except ValueError:
# # storage is not persistent
# pass
# else:
# assert c.getowner(a) == owners.default
# assert c.getowner(b) == owners.default
# c.a.a = [True]
# c.a.a.append(False)
# c.a.b[1] = True
# assert c.getowner(a) == owners.user
# assert c.getowner(b, 0) == owners.default
# assert c.getowner(b, 1) == owners.user
#owners.addowner('persistentowner2') #owners.addowner('persistentowner2')
# c.cfgimpl_get_values().setowner(b, owners.persistentowner2, 1) c.option('a.b', 1).owner.set('persistentowner2')
# c.a.b[0] = True c.option('a.b', 0).value.set(True)
# assert c.getowner(b, 0) == owners.user assert c.option('a.b', 0).owner.get() == 'user'
# assert c.getowner(b, 1) == owners.persistentowner2 assert c.option('a.b', 1).owner.get() == 'persistentowner2'
# del(c) assert c.option('a.a').value.get() == [True, False]
# # del c
# c = Config(o1, session_id='test_persistent', persistent=True)
# assert c.getowner(b, 0) == owners.user
# assert c.getowner(b, 1) == owners.persistentowner2
# delete_session('config', c.impl_getsessionid())
# del(c)
# #
# c = Config(o1, session_id='test_persistent', persistent=True)
# assert c.a.b == []
# assert c.getowner(b) == owners.default
# delete_session('config', c.impl_getsessionid())
# del(c)
# #
c = Config(o1, session_id='test_persistent', persistent=True)
assert c.option('a.a').value.get() == [True, False]
assert c.option('a.b', 0).owner.get() == 'user'
assert c.option('a.b', 1).owner.get() == 'persistentowner2'
delete_session(c.config.name())
del c
# #
#def test_two_persistent_owner(): c = Config(o1, session_id='test_persistent', persistent=True)
# b = BoolOption('b', '') assert c.option('a.a').value.get() == []
# o = OptionDescription('od', '', [b]) delete_session(c.config.name())
# try: del c
# c = Config(o, session_id='test_persistent', persistent=True)
# c.cfgimpl_get_settings().remove('cache')
# except ValueError: def test_two_persistent_owner():
# # storage is not persistent b = BoolOption('b', '')
# pass o = OptionDescription('od', '', [b])
# else: try:
# c2 = Config(o, session_id='test_persistent', persistent=True) c = Config(o, session_id='test_persistent', persistent=True)
# c2.cfgimpl_get_settings().remove('cache') c.property.pop('cache')
# assert c.getowner(b) == owners.default except ValueError:
# assert c2.getowner(b) == owners.default # storage is not persistent
# c.b = False pass
# assert c.getowner(b) == owners.user else:
# assert c2.getowner(b) == owners.user c2 = Config(o, session_id='test_persistent', persistent=True)
# owners.addowner('persistent') c2.property.pop('cache')
# c.cfgimpl_get_values().setowner(b, owners.persistent) assert c.option('b').owner.isdefault()
# assert c.getowner(b) == owners.persistent assert c2.option('b').owner.isdefault()
# assert c2.getowner(b) == owners.persistent c.option('b').value.set(False)
# delete_session('config', 'test_persistent') assert c.option('b').owner.get() == 'user'
assert c2.option('b').owner.get() == 'user'
c.option('b').owner.set('persistent')
assert c.option('b').owner.get() == 'persistent'
assert c2.option('b').owner.get() == 'persistent'
delete_session('test_persistent')
def test_create_persistent_retrieve_information():
b = BoolOption('b', '')
o = OptionDescription('od', '', [b])
try:
c = Config(o, session_id='test_persistent', persistent=True)
except ValueError:
# storage is not persistent
pass
else:
c.information.set('info', 'string')
assert c.information.get('info') == 'string'
del c
# #
c = Config(o, session_id='test_persistent', persistent=True)
assert c.information.get('info') == 'string'
delete_session(c.config.name())
del c
# #
#def test_create_persistent_retrieve_information(): c = Config(o, session_id='test_persistent', persistent=True)
# b = BoolOption('b', '') assert c.information.get('info', None) is None
# o = OptionDescription('od', '', [b]) delete_session(c.config.name())
# try: del c
# c = Config(o, session_id='test_persistent', persistent=True)
# except ValueError:
# # storage is not persistent def test_two_persistent_information():
# pass b = BoolOption('b', '')
# else: o = OptionDescription('od', '', [b])
# c.impl_set_information('info', 'string') try:
# assert c.impl_get_information('info') == 'string' c = Config(o, session_id='test_persistent', persistent=True)
# del(c) c.property.pop('cache')
# # except ValueError:
# c = Config(o, session_id='test_persistent', persistent=True) # storage is not persistent
# assert c.impl_get_information('info') == 'string' pass
# delete_session('config', c.impl_getsessionid()) else:
# del(c) c.information.set('info', 'string')
# # assert c.information.get('info') == 'string'
# c = Config(o, session_id='test_persistent', persistent=True) c2 = Config(o, session_id='test_persistent', persistent=True)
# assert c.impl_get_information('info', None) == None c2.property.pop('cache')
# delete_session('config', c.impl_getsessionid()) assert c2.information.get('info') == 'string'
# del(c) delete_session('test_persistent')
#
#
#def test_two_persistent_information(): def test_two_different_persistents():
# b = BoolOption('b', '') b = BoolOption('b', '')
# o = OptionDescription('od', '', [b]) o = OptionDescription('od', '', [b])
# try: try:
# c = Config(o, session_id='test_persistent', persistent=True) c = Config(o, session_id='test_persistent', persistent=True)
# c.cfgimpl_get_settings().remove('cache') c.property.pop('cache')
# except ValueError: d = Config(o, session_id='test_persistent2', persistent=True)
# # storage is not persistent d.property.pop('cache')
# pass except ValueError:
# else: # storage is not persistent
# c.impl_set_information('info', 'string') pass
# assert c.impl_get_information('info') == 'string' else:
# c2 = Config(o, session_id='test_persistent', persistent=True) c.option('b').property.add('test')
# c2.cfgimpl_get_settings().remove('cache') assert c.option('b').property.get() == {'test'}
# c2.cfgimpl_get_settings().remove('cache') assert d.option('b').property.get() == set()
# assert c2.impl_get_information('info') == 'string' assert c.option('b').value.get() is None
# delete_session('config', 'test_persistent') assert d.option('b').value.get() is None
# c.option('b').value.set(True)
# assert c.option('b').value.get() == True
#def test_two_different_persistents(): assert d.option('b').value.get() is None
# b = BoolOption('b', '')
# o = OptionDescription('od', '', [b]) delete_session('test_persistent')
# try: delete_session('test_persistent2')
# c = Config(o, session_id='test_persistent', persistent=True)
# c.cfgimpl_get_settings().remove('cache')
# d = Config(o, session_id='test_persistent2', persistent=True) def test_two_different_information():
# d.cfgimpl_get_settings().remove('cache') b = BoolOption('b', '')
# except ValueError: o = OptionDescription('od', '', [b])
# # storage is not persistent try:
# pass c = Config(o, session_id='test_persistent', persistent=True)
# else: c.information.set('a', 'a')
# c.cfgimpl_get_settings()[b].append('test') d = Config(o, session_id='test_persistent2', persistent=True)
# assert str(c.cfgimpl_get_settings()[b]) in ["['test']", "[u'test']"] d.information.set('a', 'b')
# assert str(d.cfgimpl_get_settings()[b]) == "[]" except ValueError:
# assert c.b is None # storage is not persistent
# assert d.b is None pass
# c.b = True else:
# assert c.b == True assert c.information.get('a') == 'a'
# assert d.b is None assert d.information.get('a') == 'b'
#
# delete_session('config', 'test_persistent') delete_session('test_persistent')
# delete_session('config', 'test_persistent2') delete_session('test_persistent2')
#
#
#def test_two_different_information():
# b = BoolOption('b', '')
# o = OptionDescription('od', '', [b])
# try:
# c = Config(o, session_id='test_persistent', persistent=True)
# c.impl_set_information('a', 'a')
# d = Config(o, session_id='test_persistent2', persistent=True)
# d.impl_set_information('a', 'b')
# except ValueError:
# # storage is not persistent
# pass
# else:
# assert c.impl_get_information('a') == 'a'
# assert d.impl_get_information('a') == 'b'
#
# delete_session('config', 'test_persistent')
# delete_session('config', 'test_persistent2')

View file

@ -135,18 +135,15 @@ def list_sessions(type_): # pragma: optional cover
return storage_type.get().list_sessions() return storage_type.get().list_sessions()
def delete_session(type_, session_id): # pragma: optional cover def delete_session(session_id): # pragma: optional cover
"""Delete a selected session, be careful, you can deleted a session """Delete a selected session, be careful, you can deleted a session
use by an other instance use by an other instance
:params session_id: id of session to delete :params session_id: id of session to delete
""" """
storage_module = storage_type.get() storage_module = storage_type.get()
session = storage_module.storage.getsession() session = storage_module.storage.getsession()
#if type_ == 'option': storage_module.value.delete_session(session_id)
# storage_option_type.get().delete_session(session_id, session) storage_module.storage.delete_session(session_id)
#else:
storage_module.value.delete_session(session_id, session)
storage_module.storage.delete_session(session_id, session)
if session: if session:
session.commit() session.commit()
del(session) del(session)

View file

@ -32,7 +32,7 @@ def list_sessions(): # pragma: optional cover
return _list_sessions return _list_sessions
def delete_session(session_id, session): # pragma: optional cover def delete_session(session_id): # pragma: optional cover
raise ConfigError(_('dictionary storage cannot delete session')) raise ConfigError(_('dictionary storage cannot delete session'))

View file

@ -46,7 +46,8 @@ def list_sessions():
return names return names
def delete_session(session_id, _session_id=None): def delete_session(session_id,
_session_id=None):
cursor = CONN.cursor() cursor = CONN.cursor()
if _session_id is None: if _session_id is None:
_session_id = cursor.execute("SELECT session_id FROM session WHERE session = ?", _session_id = cursor.execute("SELECT session_id FROM session WHERE session = ?",
@ -60,6 +61,7 @@ def delete_session(session_id, _session_id=None):
cursor.execute("DELETE FROM information WHERE session_id = ?", (_session_id,)) cursor.execute("DELETE FROM information WHERE session_id = ?", (_session_id,))
cursor.execute("DELETE FROM session WHERE session_id = ?", (_session_id,)) cursor.execute("DELETE FROM session WHERE session_id = ?", (_session_id,))
CONN.commit() CONN.commit()
cursor.close()
global CONN global CONN
CONN = None CONN = None
@ -80,7 +82,8 @@ class Storage(object):
self._cursor = self._conn.cursor() self._cursor = self._conn.cursor()
self.session_name = session_id self.session_name = session_id
if init: if init:
session_table = 'CREATE TABLE IF NOT EXISTS session(session_id INTEGER, session TEXT UNIQUE, PRIMARY KEY(session_id))' session_table = 'CREATE TABLE IF NOT EXISTS session(session_id INTEGER, '
session_table += 'session TEXT UNIQUE, persistent BOOL, PRIMARY KEY(session_id))'
settings_table = 'CREATE TABLE IF NOT EXISTS property(path TEXT,' settings_table = 'CREATE TABLE IF NOT EXISTS property(path TEXT,'
settings_table += 'properties text, session_id INTEGER, PRIMARY KEY(path, session_id), ' settings_table += 'properties text, session_id INTEGER, PRIMARY KEY(path, session_id), '
settings_table += 'FOREIGN KEY(session_id) REFERENCES session(session_id))' settings_table += 'FOREIGN KEY(session_id) REFERENCES session(session_id))'
@ -100,8 +103,15 @@ class Storage(object):
self.execute(informations_table, commit=False) self.execute(informations_table, commit=False)
self.execute(settings_table, commit=False) self.execute(settings_table, commit=False)
self.execute(permissives_table, commit=False) self.execute(permissives_table, commit=False)
self.session_id = None
if self.persistent:
select = self.select("SELECT session_id FROM session WHERE session = ?", (session_id,))
if select is not None:
self.session_id = select[0]
if self.session_id is None:
try: try:
self.execute('INSERT INTO session(session) VALUES (?)', (session_id,)) self.execute('INSERT INTO session(session, persistent) VALUES (?, ?)',
(session_id, persistent))
except sqlite3.IntegrityError: except sqlite3.IntegrityError:
raise ConflictError(_('session "{}" already used').format(session_id)) raise ConflictError(_('session "{}" already used').format(session_id))
self.session_id = self._cursor.lastrowid self.session_id = self._cursor.lastrowid
@ -126,8 +136,6 @@ class Storage(object):
def __del__(self): def __del__(self):
self._cursor.close() self._cursor.close()
#FIXME
#self._conn.close()
if not self.persistent: if not self.persistent:
if delete_session is not None: if delete_session is not None:
session_id = getattr(self, 'session_id', None) session_id = getattr(self, 'session_id', None)