cco/py-scopes
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

work in progress: topics (as example for concepts)

Browse source
This commit is contained in:
Helmut Merz 2024-03-15 21:16:45 +01:00
parent 20520a70c5
commit 606284791d
4 changed files with 107 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
scopes/storage/concept.py
View file

@ -52,6 +52,13 @@ class Predicates(Concepts):
tableName = 'preds' tableName = 'preds'
defaultPredicate = 'standard'
def storePredicate(storage, name):
preds = storage.getContainer('pred')
preds.save(Predicate(name))
class Triple(Track): class Triple(Track):
headFields = ['first', 'second', 'predicate'] headFields = ['first', 'second', 'predicate']
@ -76,8 +83,6 @@ class Rels(Container):
tableName = 'rels' tableName = 'rels'
insertOnChange = False insertOnChange = False
defaultPredicate = 'standard'
# types stuff # types stuff

18
scopes/storage/topic.py Normal file
View file

@ -0,0 +1,18 @@
# scopes.storage.topic
from scopes.storage.common import registerContainerClass
from scopes.storage import concept
class Topic(concept.Concept):
prefix = 'tpc'
@registerContainerClass
class Topics(concept.Concepts):
itemFactory = Topic
tableName = 'topics'

3
tests/test_standard.py
View file

@ -24,6 +24,9 @@ class Test(unittest.TestCase):
def test_003_type(self): def test_003_type(self):
tlib_storage.test_type(self, config) tlib_storage.test_type(self, config)
def test_004_topic(self):
tlib_storage.test_topic(self, config)
def test_013_server(self): def test_013_server(self):
tlib_server.test_app(self, config) tlib_server.test_app(self, config)

143
tests/tlib_storage.py
View file

@ -3,90 +3,105 @@
"""Test implementation for the `scopes.storage` package.""" """Test implementation for the `scopes.storage` package."""
from datetime import datetime from datetime import datetime
from scopes.storage import concept, folder, tracking from scopes.storage import concept, folder, topic, tracking
def test_tracking(self, config): def test_tracking(self, config):
storage = config.storageFactory(config.dbschema) storage = config.storageFactory(config.dbschema)
storage.dropTable('tracks') storage.dropTable('tracks')
tracks = storage.create(tracking.Container) tracks = storage.create(tracking.Container)
tr01 = tracking.Track('t01', 'john') tr01 = tracking.Track('t01', 'john')
tr01.update(dict(activity='testing')) tr01.update(dict(activity='testing'))
self.assertEqual(tr01.head, {'taskId': 't01', 'userName': 'john'}) self.assertEqual(tr01.head, {'taskId': 't01', 'userName': 'john'})
self.assertEqual(tr01.taskId, 't01') self.assertEqual(tr01.taskId, 't01')
self.assertEqual(tr01.userName, 'john') self.assertEqual(tr01.userName, 'john')
self.assertTrue(tracks.getTable() is not None) self.assertTrue(tracks.getTable() is not None)
trid01 = tracks.save(tr01) trid01 = tracks.save(tr01)
self.assertTrue(trid01 > 0) self.assertTrue(trid01 > 0)
#tr01a = tracks.get(trid01) #tr01a = tracks.get(trid01)
tr01a = tracks['%07i' % trid01] tr01a = tracks['%07i' % trid01]
self.assertEqual(tr01a.head, tr01.head) self.assertEqual(tr01a.head, tr01.head)
self.assertEqual(tr01a.trackId, trid01) self.assertEqual(tr01a.trackId, trid01)
self.assertEqual(tr01a.data.get('activity'), 'testing') self.assertEqual(tr01a.data.get('activity'), 'testing')
tr01a.update(dict(text='Set up unit tests.')) tr01a.update(dict(text='Set up unit tests.'))
tr01a.timeStamp = None tr01a.timeStamp = None
self.assertTrue(tracks.save(tr01a) > 0) self.assertTrue(tracks.save(tr01a) > 0)
tr01b = tracks.queryLast(taskId='t01') tr01b = tracks.queryLast(taskId='t01')
self.assertEqual(tr01b.head, tr01.head) self.assertEqual(tr01b.head, tr01.head)
self.assertNotEqual(tr01b.trackId, trid01) self.assertNotEqual(tr01b.trackId, trid01)
self.assertEqual(tr01b.data.get('activity'), 'testing') self.assertEqual(tr01b.data.get('activity'), 'testing')
tr02 = tracking.Track('t02', 'jim', trackId=31, timeStamp=datetime(2023, 11, 30), tr02 = tracking.Track('t02', 'jim', trackId=31, timeStamp=datetime(2023, 11, 30),
data=dict(activity='concept')) data=dict(activity='concept'))
trid02 = tracks.upsert(tr02) trid02 = tracks.upsert(tr02)
self.assertEqual(trid02, 31) self.assertEqual(trid02, 31)
self.assertEqual(tr02.uid, 'rec-31') self.assertEqual(tr02.uid, 'rec-31')
tr02.trackId = trid01 tr02.trackId = trid01
trid021 = tracks.upsert(tr02) trid021 = tracks.upsert(tr02)
self.assertEqual(trid021, trid01) self.assertEqual(trid021, trid01)
self.assertEqual(tr02.uid, 'rec-' + str(trid01)) self.assertEqual(tr02.uid, 'rec-' + str(trid01))
tr03 = storage.getItem('rec-31') tr03 = storage.getItem('rec-31')
self.assertEqual(tr03.trackId, 31) self.assertEqual(tr03.trackId, 31)
n = tracks.remove(31) n = tracks.remove(31)
self.assertEqual(n, 1) self.assertEqual(n, 1)
self.assertEqual(tracks.get(31), None) self.assertEqual(tracks.get(31), None)
storage.commit() storage.commit()
def test_folder(self, config): def test_folder(self, config):
storage = config.storageFactory(config.dbschema) storage = config.storageFactory(config.dbschema)
storage.dropTable('folders') storage.dropTable('folders')
root = folder.Root(storage) root = folder.Root(storage)
self.assertEqual(list(root.keys()), []) self.assertEqual(list(root.keys()), [])
root['top'] = folder.Folder() root['top'] = folder.Folder()
self.assertEqual(list(root.keys()), ['top']) self.assertEqual(list(root.keys()), ['top'])
top = root['top'] top = root['top']
top['child1'] = folder.Folder(data=dict(title='First Child')) top['child1'] = folder.Folder(data=dict(title='First Child'))
self.assertEqual(list(top.keys()), ['child1']) self.assertEqual(list(top.keys()), ['child1'])
ch1 = top['child1'] ch1 = top['child1']
self.assertEqual(ch1.parent, top.rid) self.assertEqual(ch1.parent, top.rid)
self.assertEqual(list(top.keys()), ['child1']) self.assertEqual(list(top.keys()), ['child1'])
storage.commit() storage.commit()
def test_type(self, config): def test_type(self, config):
storage = config.storageFactory(config.dbschema) storage = config.storageFactory(config.dbschema)
storage.dropTable('types') storage.dropTable('types')
concept.setupCoreTypes(storage) concept.setupCoreTypes(storage)
types = storage.getContainer(concept.Type.prefix) types = storage.getContainer(concept.Type.prefix)
tps = list(types.query()) tps = list(types.query())
self.assertEqual(len(tps), 5) self.assertEqual(len(tps), 6)
self.assertEqual(tps[0].name, 'track') self.assertEqual(tps[0].name, 'track')
tfolder = types.queryLast(name='folder') tfolder = types.queryLast(name='folder')
fldrs = list(tfolder.values()) fldrs = list(tfolder.values())
self.assertEqual(len(fldrs), 2) self.assertEqual(len(fldrs), 2)
self.assertEqual(fldrs[0].name, 'top') self.assertEqual(fldrs[0].name, 'top')
storage.commit()
def test_topic(self, config):
storage = config.storageFactory(config.dbschema)
storage.dropTable('topics')
topics = storage.getContainer(topic.Topic.prefix)
concept.storePredicate(storage, concept.defaultPredicate)
root = folder.Root(storage)
root['top']['topics'] = folder.Folder()
tp_itc = topic.Topic('itc')
topics.save(tp_itc)
storage.commit()
storage.commit()