diff --git a/test/test_storage.py b/test/test_storage.py index 6e3f573..54eab0e 100644 --- a/test/test_storage.py +++ b/test/test_storage.py @@ -17,286 +17,286 @@ def test_non_persistent(): Config(o, session_id='test_non_persistent') -#def test_list(): -# b = BoolOption('b', '') -# o = OptionDescription('od', '', [b]) -# c = Config(o, session_id='test_non_persistent') -# c.cfgimpl_get_settings().remove('cache') -# c.b = True -# assert 'test_non_persistent' in list_sessions('config') -# del(c) -# 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) -# 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) -# except ValueError: -# raises(ValueError, "delete_session('option', '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.b = True -# except ValueError: -# # storage is not persistent -# pass -# else: -# assert 'test_persistent' in list_sessions('config') -# -# -#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('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') -# c.cfgimpl_get_values().setowner(b, owners.persistentowner2, 1) -# c.a.b[0] = True -# assert c.getowner(b, 0) == owners.user -# assert c.getowner(b, 1) == owners.persistentowner2 -# 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) -# -# -#def test_two_persistent_owner(): -# b = BoolOption('b', '') -# o = OptionDescription('od', '', [b]) -# try: -# c = Config(o, session_id='test_persistent', persistent=True) -# c.cfgimpl_get_settings().remove('cache') -# except ValueError: -# # storage is not persistent -# pass -# else: -# c2 = Config(o, session_id='test_persistent', persistent=True) -# c2.cfgimpl_get_settings().remove('cache') -# assert c.getowner(b) == owners.default -# assert c2.getowner(b) == owners.default -# c.b = False -# assert c.getowner(b) == owners.user -# assert c2.getowner(b) == owners.user -# owners.addowner('persistent') -# c.cfgimpl_get_values().setowner(b, owners.persistent) -# assert c.getowner(b) == owners.persistent -# assert c2.getowner(b) == owners.persistent -# delete_session('config', '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.impl_set_information('info', 'string') -# assert c.impl_get_information('info') == 'string' -# del(c) -# # -# c = Config(o, session_id='test_persistent', persistent=True) -# assert c.impl_get_information('info') == 'string' -# delete_session('config', c.impl_getsessionid()) -# del(c) -# # -# c = Config(o, session_id='test_persistent', persistent=True) -# assert c.impl_get_information('info', None) == None -# delete_session('config', c.impl_getsessionid()) -# del(c) -# -# -#def test_two_persistent_information(): -# b = BoolOption('b', '') -# o = OptionDescription('od', '', [b]) -# try: -# c = Config(o, session_id='test_persistent', persistent=True) -# c.cfgimpl_get_settings().remove('cache') -# except ValueError: -# # storage is not persistent -# pass -# else: -# c.impl_set_information('info', 'string') -# assert c.impl_get_information('info') == 'string' -# c2 = Config(o, session_id='test_persistent', persistent=True) -# c2.cfgimpl_get_settings().remove('cache') -# c2.cfgimpl_get_settings().remove('cache') -# assert c2.impl_get_information('info') == 'string' -# delete_session('config', 'test_persistent') -# -# -#def test_two_different_persistents(): -# b = BoolOption('b', '') -# o = OptionDescription('od', '', [b]) -# try: -# c = Config(o, session_id='test_persistent', persistent=True) -# c.cfgimpl_get_settings().remove('cache') -# d = Config(o, session_id='test_persistent2', persistent=True) -# d.cfgimpl_get_settings().remove('cache') -# except ValueError: -# # storage is not persistent -# pass -# else: -# c.cfgimpl_get_settings()[b].append('test') -# assert str(c.cfgimpl_get_settings()[b]) in ["['test']", "[u'test']"] -# assert str(d.cfgimpl_get_settings()[b]) == "[]" -# assert c.b is None -# assert d.b is None -# c.b = True -# assert c.b == True -# assert d.b is None -# -# delete_session('config', 'test_persistent') -# delete_session('config', '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') +def test_list(): + b = BoolOption('b', '') + o = OptionDescription('od', '', [b]) + c = Config(o, session_id='test_non_persistent') + c.option('b').value.set(True) + assert 'test_non_persistent' in list_sessions('config') + del(c) + 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 + # + c.option('b').value.set(True) + assert c.option('b').value.get() is True + assert c2.option('b').value.get() is True + delete_session('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.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 + # + c = Config(o, session_id='test_persistent', persistent=True) + assert c.option('b').value.get() is None + assert c.option('b').owner.isdefault() + delete_session(c.config.name()) + 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.option('a.a').owner.isdefault() + c.option('a.a').value.set([True, False]) + c.option('a.b', 1).value.set(True) + assert c.option('a.a').owner.get() == 'user' + assert c.option('a.b', 0).owner.isdefault() + assert c.option('a.b', 1).owner.get() == 'user' + #owners.addowner('persistentowner2') + c.option('a.b', 1).owner.set('persistentowner2') + c.option('a.b', 0).value.set(True) + assert c.option('a.b', 0).owner.get() == 'user' + assert c.option('a.b', 1).owner.get() == 'persistentowner2' + assert c.option('a.a').value.get() == [True, False] + 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 + # + c = Config(o1, session_id='test_persistent', persistent=True) + assert c.option('a.a').value.get() == [] + delete_session(c.config.name()) + del c + + +def test_two_persistent_owner(): + b = BoolOption('b', '') + o = OptionDescription('od', '', [b]) + try: + c = Config(o, session_id='test_persistent', persistent=True) + c.property.pop('cache') + 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').owner.isdefault() + assert c2.option('b').owner.isdefault() + c.option('b').value.set(False) + 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 + # + c = Config(o, session_id='test_persistent', persistent=True) + assert c.information.get('info', None) is None + delete_session(c.config.name()) + del c + + +def test_two_persistent_information(): + b = BoolOption('b', '') + o = OptionDescription('od', '', [b]) + try: + c = Config(o, session_id='test_persistent', persistent=True) + c.property.pop('cache') + except ValueError: + # storage is not persistent + pass + else: + c.information.set('info', 'string') + assert c.information.get('info') == 'string' + c2 = Config(o, session_id='test_persistent', persistent=True) + c2.property.pop('cache') + assert c2.information.get('info') == 'string' + delete_session('test_persistent') + + +def test_two_different_persistents(): + b = BoolOption('b', '') + o = OptionDescription('od', '', [b]) + try: + c = Config(o, session_id='test_persistent', persistent=True) + c.property.pop('cache') + d = Config(o, session_id='test_persistent2', persistent=True) + d.property.pop('cache') + except ValueError: + # storage is not persistent + pass + else: + c.option('b').property.add('test') + assert c.option('b').property.get() == {'test'} + assert d.option('b').property.get() == set() + assert c.option('b').value.get() is None + assert d.option('b').value.get() is None + c.option('b').value.set(True) + assert c.option('b').value.get() == True + assert d.option('b').value.get() is None + + delete_session('test_persistent') + 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.information.set('a', 'a') + d = Config(o, session_id='test_persistent2', persistent=True) + d.information.set('a', 'b') + except ValueError: + # storage is not persistent + pass + else: + assert c.information.get('a') == 'a' + assert d.information.get('a') == 'b' + + delete_session('test_persistent') + delete_session('test_persistent2') diff --git a/tiramisu/storage/__init__.py b/tiramisu/storage/__init__.py index f67a4b9..3ecd8f0 100644 --- a/tiramisu/storage/__init__.py +++ b/tiramisu/storage/__init__.py @@ -135,18 +135,15 @@ def list_sessions(type_): # pragma: optional cover 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 use by an other instance :params session_id: id of session to delete """ storage_module = storage_type.get() session = storage_module.storage.getsession() - #if type_ == 'option': - # storage_option_type.get().delete_session(session_id, session) - #else: - storage_module.value.delete_session(session_id, session) - storage_module.storage.delete_session(session_id, session) + storage_module.value.delete_session(session_id) + storage_module.storage.delete_session(session_id) if session: session.commit() del(session) diff --git a/tiramisu/storage/dictionary/storage.py b/tiramisu/storage/dictionary/storage.py index 9758c01..32c5747 100644 --- a/tiramisu/storage/dictionary/storage.py +++ b/tiramisu/storage/dictionary/storage.py @@ -32,7 +32,7 @@ def list_sessions(): # pragma: optional cover 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')) diff --git a/tiramisu/storage/sqlite3/storage.py b/tiramisu/storage/sqlite3/storage.py index 0be0262..040e04c 100644 --- a/tiramisu/storage/sqlite3/storage.py +++ b/tiramisu/storage/sqlite3/storage.py @@ -46,7 +46,8 @@ def list_sessions(): return names -def delete_session(session_id, _session_id=None): +def delete_session(session_id, + _session_id=None): cursor = CONN.cursor() if _session_id is None: _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 session WHERE session_id = ?", (_session_id,)) CONN.commit() + cursor.close() global CONN CONN = None @@ -80,7 +82,8 @@ class Storage(object): self._cursor = self._conn.cursor() self.session_name = session_id 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 += 'properties text, session_id INTEGER, PRIMARY KEY(path, session_id), ' settings_table += 'FOREIGN KEY(session_id) REFERENCES session(session_id))' @@ -100,11 +103,18 @@ class Storage(object): self.execute(informations_table, commit=False) self.execute(settings_table, commit=False) self.execute(permissives_table, commit=False) - try: - self.execute('INSERT INTO session(session) VALUES (?)', (session_id,)) - except sqlite3.IntegrityError: - raise ConflictError(_('session "{}" already used').format(session_id)) - self.session_id = self._cursor.lastrowid + 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: + self.execute('INSERT INTO session(session, persistent) VALUES (?, ?)', + (session_id, persistent)) + except sqlite3.IntegrityError: + raise ConflictError(_('session "{}" already used').format(session_id)) + self.session_id = self._cursor.lastrowid def commit(self): self._conn.commit() @@ -126,8 +136,6 @@ class Storage(object): def __del__(self): self._cursor.close() - #FIXME - #self._conn.close() if not self.persistent: if delete_session is not None: session_id = getattr(self, 'session_id', None)