testing improvements; work in progress: folders traversable by zope.publisher
This commit is contained in:
parent
fd6006899e
commit
419902ae3f
7 changed files with 49 additions and 31 deletions
|
@ -2,17 +2,19 @@
|
||||||
|
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
from os import getenv
|
from os import getenv
|
||||||
from scopes.server.app import demo_app, zope_app
|
from scopes.server.app import zope_app_factory
|
||||||
|
|
||||||
load_dotenv()
|
load_dotenv()
|
||||||
|
|
||||||
server_port = getenv('SERVER_PORT', '8099')
|
server_port = getenv('SERVER_PORT', '8099')
|
||||||
|
|
||||||
app = zope_app
|
app_factory = zope_app_factory
|
||||||
|
|
||||||
# storage settings
|
# storage settings
|
||||||
|
from scopes.storage.db.postgres import StorageFactory
|
||||||
dbengine = 'postgresql+psycopg'
|
dbengine = 'postgresql+psycopg'
|
||||||
dbname = getenv('DBNAME', 'demo')
|
dbname = getenv('DBNAME', 'demo')
|
||||||
dbuser = getenv('DBUSER', 'demo')
|
dbuser = getenv('DBUSER', 'demo')
|
||||||
dbpassword = getenv('DBPASSWORD', 'secret')
|
dbpassword = getenv('DBPASSWORD', 'secret')
|
||||||
|
dbschema = getenv('DBSCHEMA', 'demo')
|
||||||
|
|
||||||
|
|
|
@ -14,6 +14,7 @@ def run(app, config):
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
import config
|
import config
|
||||||
run(config.app, config)
|
#run(config.app, config)
|
||||||
#run(config.app_factory(), config)
|
app = config.app_factory(config)
|
||||||
|
run(app, config)
|
||||||
# see zope.app.wsgi.getWSGIApplication
|
# see zope.app.wsgi.getWSGIApplication
|
||||||
|
|
|
@ -5,6 +5,8 @@ from zope.publisher.browser import BrowserRequest
|
||||||
from zope.publisher.publish import publish
|
from zope.publisher.publish import publish
|
||||||
from zope.traversing.publicationtraverse import PublicationTraverser
|
from zope.traversing.publicationtraverse import PublicationTraverser
|
||||||
|
|
||||||
|
from scopes.storage.folder import Root
|
||||||
|
|
||||||
|
|
||||||
def demo_app(environ, start_response):
|
def demo_app(environ, start_response):
|
||||||
print(f'*** environ {environ}.')
|
print(f'*** environ {environ}.')
|
||||||
|
@ -14,18 +16,26 @@ def demo_app(environ, start_response):
|
||||||
return ['Hello World'.encode()]
|
return ['Hello World'.encode()]
|
||||||
|
|
||||||
|
|
||||||
def zope_app(environ, start_response):
|
def zope_app_factory(config):
|
||||||
|
config.storageFactory = config.StorageFactory(config)
|
||||||
|
def zope_app(environ, start_response):
|
||||||
request = BrowserRequest(environ['wsgi.input'], environ)
|
request = BrowserRequest(environ['wsgi.input'], environ)
|
||||||
request.setPublication(DefaultPublication(AppRoot()))
|
storage = config.storageFactory(config.dbschema)
|
||||||
|
appRoot = Root(storage, config)
|
||||||
|
request.setPublication(DefaultPublication(appRoot))
|
||||||
request = publish(request, False)
|
request = publish(request, False)
|
||||||
response = request.response
|
response = request.response
|
||||||
start_response(response.getStatusString(), response.getHeaders())
|
start_response(response.getStatusString(), response.getHeaders())
|
||||||
return response.consumeBodyIter()
|
return response.consumeBodyIter()
|
||||||
|
return zope_app
|
||||||
|
|
||||||
|
|
||||||
class AppRoot:
|
class AppRoot:
|
||||||
"""Zope Demo AppRoot"""
|
"""Zope Demo AppRoot"""
|
||||||
|
|
||||||
|
def __init__(self, config):
|
||||||
|
self.config = config
|
||||||
|
|
||||||
def __call__(self):
|
def __call__(self):
|
||||||
"""calling AppRoot"""
|
"""calling AppRoot"""
|
||||||
return 'At root'
|
return 'At root'
|
||||||
|
|
|
@ -5,6 +5,7 @@ from scopes.storage.tracking import Container, Track
|
||||||
|
|
||||||
|
|
||||||
class Folder(Track):
|
class Folder(Track):
|
||||||
|
"""Needs docstring to be traversable."""
|
||||||
|
|
||||||
headFields = ['parent', 'name', 'ref']
|
headFields = ['parent', 'name', 'ref']
|
||||||
prefix = 'fldr'
|
prefix = 'fldr'
|
||||||
|
@ -30,12 +31,16 @@ class Folder(Track):
|
||||||
value.set('name', key)
|
value.set('name', key)
|
||||||
self.container.save(value)
|
self.container.save(value)
|
||||||
|
|
||||||
|
def __call__(self, request=None):
|
||||||
|
return 'folder: %s; keys: %s' % (self.name, list(self.keys()))
|
||||||
|
|
||||||
|
|
||||||
class Root(Folder):
|
class Root(Folder):
|
||||||
"""A dummy (virtual) root folder for creating real folders
|
"""A dummy (virtual) root folder for creating real folders
|
||||||
using the Folder API."""
|
using the Folder API."""
|
||||||
|
|
||||||
def __init__(self, storage):
|
def __init__(self, storage, config=None):
|
||||||
|
self.config = config
|
||||||
cont = storage.create(Folders)
|
cont = storage.create(Folders)
|
||||||
super(Root, self).__init__(container=cont)
|
super(Root, self).__init__(container=cont)
|
||||||
|
|
||||||
|
|
|
@ -3,27 +3,25 @@
|
||||||
"""Tests for the 'scopes.storage' package - using PostgreSQL."""
|
"""Tests for the 'scopes.storage' package - using PostgreSQL."""
|
||||||
|
|
||||||
import unittest
|
import unittest
|
||||||
|
import tlib_storage
|
||||||
|
|
||||||
|
from scopes.storage.db.postgres import StorageFactory
|
||||||
import config
|
import config
|
||||||
config.dbengine = 'postgresql+psycopg'
|
config.dbengine = 'postgresql+psycopg'
|
||||||
config.dbname = 'testdb'
|
config.dbname = 'testdb'
|
||||||
config.dbuser = 'testuser'
|
config.dbuser = 'testuser'
|
||||||
config.dbpassword = 'secret'
|
config.dbpassword = 'secret'
|
||||||
|
config.dbschema = 'testing'
|
||||||
|
config.storageFactory = StorageFactory(config)
|
||||||
|
|
||||||
# PostgreSQL-specific settings
|
|
||||||
from scopes.storage.db.postgres import StorageFactory
|
|
||||||
factory = StorageFactory(config)
|
|
||||||
storage = factory(schema='testing')
|
|
||||||
|
|
||||||
import tlib
|
|
||||||
|
|
||||||
class Test(unittest.TestCase):
|
class Test(unittest.TestCase):
|
||||||
|
|
||||||
def test_001_tracking(self):
|
def test_001_tracking(self):
|
||||||
tlib.test_tracking(self, storage)
|
tlib_storage.test_tracking(self, config)
|
||||||
|
|
||||||
def test_002_folder(self):
|
def test_002_folder(self):
|
||||||
tlib.test_folder(self, storage)
|
tlib_storage.test_folder(self, config)
|
||||||
|
|
||||||
def suite():
|
def suite():
|
||||||
return unittest.TestSuite((
|
return unittest.TestSuite((
|
||||||
|
|
|
@ -3,24 +3,23 @@
|
||||||
"""Tests for the 'scopes.storage' package."""
|
"""Tests for the 'scopes.storage' package."""
|
||||||
|
|
||||||
import unittest
|
import unittest
|
||||||
|
import tlib_storage
|
||||||
|
|
||||||
|
from scopes.storage.common import StorageFactory
|
||||||
import config
|
import config
|
||||||
config.dbengine = 'sqlite'
|
config.dbengine = 'sqlite'
|
||||||
config.dbname = 'var/test.db'
|
config.dbname = 'var/test.db'
|
||||||
|
config.dbschema = None
|
||||||
|
config.storageFactory = StorageFactory(config)
|
||||||
|
|
||||||
from scopes.storage.common import StorageFactory
|
|
||||||
factory = StorageFactory(config)
|
|
||||||
storage = factory(schema=None)
|
|
||||||
|
|
||||||
import tlib
|
|
||||||
|
|
||||||
class Test(unittest.TestCase):
|
class Test(unittest.TestCase):
|
||||||
|
|
||||||
def test_001_tracking(self):
|
def test_001_tracking(self):
|
||||||
tlib.test_tracking(self, storage)
|
tlib_storage.test_tracking(self, config)
|
||||||
|
|
||||||
def test_002_folder(self):
|
def test_002_folder(self):
|
||||||
tlib.test_folder(self, storage)
|
tlib_storage.test_folder(self, config)
|
||||||
|
|
||||||
def suite():
|
def suite():
|
||||||
return unittest.TestSuite((
|
return unittest.TestSuite((
|
||||||
|
|
|
@ -1,10 +1,12 @@
|
||||||
"""The real test implementations"""
|
# tests/tlib_storage.py
|
||||||
|
"""Test implementation for the `scopes.storage` package."""
|
||||||
|
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from scopes.storage import folder, tracking
|
from scopes.storage import folder, tracking
|
||||||
|
|
||||||
|
|
||||||
def test_tracking(self, storage):
|
def test_tracking(self, config):
|
||||||
|
storage = config.storageFactory(config.dbschema)
|
||||||
storage.dropTable('tracks')
|
storage.dropTable('tracks')
|
||||||
tracks = storage.create(tracking.Container)
|
tracks = storage.create(tracking.Container)
|
||||||
|
|
||||||
|
@ -54,9 +56,10 @@ def test_tracking(self, storage):
|
||||||
storage.commit()
|
storage.commit()
|
||||||
|
|
||||||
|
|
||||||
def test_folder(self, storage):
|
def test_folder(self, config):
|
||||||
|
storage = config.storageFactory(config.dbschema)
|
||||||
storage.dropTable('folders')
|
storage.dropTable('folders')
|
||||||
root = folder.Root(storage)
|
root = folder.Root(storage, config)
|
||||||
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'])
|
Loading…
Add table
Reference in a new issue