131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
# py-scopes: tests.test_storage
|
|
|
|
""" Tests for the (PostgreSQL) storage-related functionalities"""
|
|
|
|
from scopes import testing
|
|
import config
|
|
|
|
from datetime import datetime
|
|
import pytest
|
|
|
|
from scopes.api import main
|
|
from scopes.storage import concept, folder, message, topic, tracking
|
|
from scopes.util import keywords as k
|
|
|
|
# predefined keywords used as domain, action, class head values in messages
|
|
k('session system')
|
|
|
|
@pytest.fixture(scope='module')
|
|
def storage():
|
|
storage = config.StorageFactory(config)(config.dbschema)
|
|
storage.dropTable('tracks')
|
|
storage.dropTable('folders')
|
|
storage.dropTable('types')
|
|
storage.dropTable('preds')
|
|
storage.dropTable('rels')
|
|
storage.dropTable('topics')
|
|
storage.dropTable('messages')
|
|
storage.commit()
|
|
return storage
|
|
|
|
def test_tracking(storage):
|
|
tracks = storage.create(tracking.Container)
|
|
tr01 = tracking.Track('t01', 'john')
|
|
assert tr01.head == {'taskId': 't01', 'userName': 'john'}
|
|
assert tr01.taskId == 't01'
|
|
assert tr01.userName == 'john'
|
|
tr01.update(dict(activity='testing'))
|
|
trid01 = tracks.save(tr01)
|
|
assert trid01 > 0
|
|
|
|
tr01a = tracks[f'{trid01:07}']
|
|
assert tr01a.head == tr01.head
|
|
assert tr01a.trackId == trid01
|
|
assert tr01a.data.get('activity') == 'testing'
|
|
tr01a.update(dict(text='Set up unit tests.'))
|
|
tr01a.timeStamp = None
|
|
assert tracks.save(tr01a) > 0
|
|
|
|
tr01b = tracks.queryLast(taskId='t01')
|
|
assert tr01b.head == tr01.head
|
|
assert tr01b.trackId != trid01
|
|
assert tr01b.data.get('activity') == 'testing'
|
|
assert tr01b.data.get('text') == 'Set up unit tests.'
|
|
|
|
tr02 = tracking.Track('t02', 'jim', trackId=31, timeStamp=datetime(2023, 11, 30),
|
|
data=dict(activity='concept'))
|
|
trid02 = tracks.upsert(tr02)
|
|
assert trid02 == 31
|
|
assert tr02.uid == 'rec-31'
|
|
tr02.trackId = trid01
|
|
trid021 = tracks.upsert(tr02)
|
|
assert trid021 == trid01
|
|
assert tr02.uid == 'rec-' + str(trid01)
|
|
|
|
tr03 = storage.getItem('rec-31')
|
|
assert tr03.trackId == 31
|
|
n = tracks.remove(tr03)
|
|
assert n == 1
|
|
assert tracks.get(31) == None
|
|
|
|
storage.commit()
|
|
|
|
def test_folder(storage):
|
|
root = folder.Root(storage)
|
|
assert list(root.keys()) == []
|
|
root['top'] = folder.Folder()
|
|
assert list(root.keys()) == ['top']
|
|
top = root['top']
|
|
top['child1'] = folder.Folder(data=dict(title='First Child'))
|
|
assert list(top.keys()) == ['child1']
|
|
ch1 = top['child1']
|
|
assert ch1.parent == top.rid
|
|
assert list(top.keys()) == ['child1']
|
|
ch1.set('name', 'level2-item1')
|
|
ch1.storeTrack()
|
|
storage.commit()
|
|
|
|
def test_type(storage):
|
|
concept.setupCoreTypes(storage)
|
|
types = storage.getContainer(concept.Type)
|
|
tps = list(types.query())
|
|
assert len(tps) == 7
|
|
tfolder = types.queryLast(name='folder')
|
|
fldrs = list(tfolder.values())
|
|
assert len(fldrs) == 2
|
|
assert fldrs[0].name == 'top'
|
|
storage.commit()
|
|
|
|
def test_topic(storage):
|
|
rels = storage.getContainer(concept.Triple)
|
|
topics = storage.getContainer(topic.Topic)
|
|
types = storage.getContainer(concept.Type)
|
|
concept.storePredicate(storage, concept.defaultPredicate)
|
|
root = folder.Root(storage)
|
|
root['top']['topics'] = ftopics = folder.Folder()
|
|
ttopic = types.queryLast(name='topic')
|
|
assert ttopic.name == 'topic'
|
|
ftopics.setTarget(ttopic)
|
|
assert ftopics.getTarget().name == 'topic'
|
|
tp_itc = topic.Topic('itc', data=dict(
|
|
title='ITC', description='Information and Communication Technology'))
|
|
topics.save(tp_itc)
|
|
tp_proglang = topic.Topic('prog_lang', data=dict(
|
|
title='Programming Languages',
|
|
description='Programming Languages'))
|
|
topics.save(tp_proglang)
|
|
tp_itc.addChild(tp_proglang)
|
|
c = list(tp_itc.children())
|
|
assert c[0].name == 'prog_lang'
|
|
storage.commit()
|
|
|
|
def test_message(storage):
|
|
messages = storage.create(message.Messages)
|
|
m01 = message.Message('system', 'data', 'session', 'V1_317784226621611853')
|
|
m01.update(dict(userid='tst9'))
|
|
mid01 = messages.save(m01)
|
|
msgs = list(messages.query())
|
|
assert msgs[0].domain is k.system
|
|
assert len(msgs) == 1
|
|
storage.commit()
|
|
|