test: add 'nodb' test runner
ensure things work as expected if sqlalchemy is not installed
This commit is contained in:
Lance Edgar
parent
132073177c
commit
f5825e964c
5 changed files with 193 additions and 148 deletions
|
|
@ -5,128 +5,131 @@ import shutil
|
|||
import tempfile
|
||||
from unittest import TestCase
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import orm
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy.pool import NullPool
|
||||
|
||||
from wuttjamaican.db import conf
|
||||
from wuttjamaican.conf import WuttaConfig
|
||||
|
||||
try:
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import orm
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy.pool import NullPool
|
||||
from wuttjamaican.db import conf
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
|
||||
class TestGetEngines(TestCase):
|
||||
class TestGetEngines(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.tempdir = tempfile.mkdtemp()
|
||||
def setUp(self):
|
||||
self.tempdir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tempdir)
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tempdir)
|
||||
|
||||
def write_file(self, filename, content):
|
||||
path = os.path.join(self.tempdir, filename)
|
||||
with open(path, 'wt') as f:
|
||||
f.write(content)
|
||||
return path
|
||||
def write_file(self, filename, content):
|
||||
path = os.path.join(self.tempdir, filename)
|
||||
with open(path, 'wt') as f:
|
||||
f.write(content)
|
||||
return path
|
||||
|
||||
def test_no_default(self):
|
||||
myfile = self.write_file('my.conf', '')
|
||||
config = WuttaConfig([myfile])
|
||||
self.assertEqual(conf.get_engines(config, 'wuttadb'), {})
|
||||
def test_no_default(self):
|
||||
myfile = self.write_file('my.conf', '')
|
||||
config = WuttaConfig([myfile])
|
||||
self.assertEqual(conf.get_engines(config, 'wuttadb'), {})
|
||||
|
||||
def test_default(self):
|
||||
myfile = self.write_file('my.conf', """\
|
||||
[wuttadb]
|
||||
default.url = sqlite://
|
||||
""")
|
||||
config = WuttaConfig([myfile])
|
||||
result = conf.get_engines(config, 'wuttadb')
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertIn('default', result)
|
||||
engine = result['default']
|
||||
self.assertEqual(engine.dialect.name, 'sqlite')
|
||||
def test_default(self):
|
||||
myfile = self.write_file('my.conf', """\
|
||||
[wuttadb]
|
||||
default.url = sqlite://
|
||||
""")
|
||||
config = WuttaConfig([myfile])
|
||||
result = conf.get_engines(config, 'wuttadb')
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertIn('default', result)
|
||||
engine = result['default']
|
||||
self.assertEqual(engine.dialect.name, 'sqlite')
|
||||
|
||||
def test_default_fallback(self):
|
||||
myfile = self.write_file('my.conf', """\
|
||||
[wuttadb]
|
||||
sqlalchemy.url = sqlite://
|
||||
""")
|
||||
config = WuttaConfig([myfile])
|
||||
result = conf.get_engines(config, 'wuttadb')
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertIn('default', result)
|
||||
engine = result['default']
|
||||
self.assertEqual(engine.dialect.name, 'sqlite')
|
||||
def test_default_fallback(self):
|
||||
myfile = self.write_file('my.conf', """\
|
||||
[wuttadb]
|
||||
sqlalchemy.url = sqlite://
|
||||
""")
|
||||
config = WuttaConfig([myfile])
|
||||
result = conf.get_engines(config, 'wuttadb')
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertIn('default', result)
|
||||
engine = result['default']
|
||||
self.assertEqual(engine.dialect.name, 'sqlite')
|
||||
|
||||
def test_other(self):
|
||||
myfile = self.write_file('my.conf', """\
|
||||
[otherdb]
|
||||
keys = first, second
|
||||
first.url = sqlite://
|
||||
second.url = sqlite://
|
||||
""")
|
||||
config = WuttaConfig([myfile])
|
||||
result = conf.get_engines(config, 'otherdb')
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertIn('first', result)
|
||||
self.assertIn('second', result)
|
||||
def test_other(self):
|
||||
myfile = self.write_file('my.conf', """\
|
||||
[otherdb]
|
||||
keys = first, second
|
||||
first.url = sqlite://
|
||||
second.url = sqlite://
|
||||
""")
|
||||
config = WuttaConfig([myfile])
|
||||
result = conf.get_engines(config, 'otherdb')
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertIn('first', result)
|
||||
self.assertIn('second', result)
|
||||
|
||||
|
||||
class TestGetSetting(TestCase):
|
||||
class TestGetSetting(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
Session = orm.sessionmaker()
|
||||
engine = sa.create_engine('sqlite://')
|
||||
self.session = Session(bind=engine)
|
||||
self.session.execute(sa.text("""
|
||||
create table setting (
|
||||
name varchar(255) primary key,
|
||||
value text
|
||||
);
|
||||
"""))
|
||||
def setUp(self):
|
||||
Session = orm.sessionmaker()
|
||||
engine = sa.create_engine('sqlite://')
|
||||
self.session = Session(bind=engine)
|
||||
self.session.execute(sa.text("""
|
||||
create table setting (
|
||||
name varchar(255) primary key,
|
||||
value text
|
||||
);
|
||||
"""))
|
||||
|
||||
def tearDown(self):
|
||||
self.session.close()
|
||||
def tearDown(self):
|
||||
self.session.close()
|
||||
|
||||
def test_basic_value(self):
|
||||
self.session.execute(sa.text("insert into setting values ('foo', 'bar');"))
|
||||
value = conf.get_setting(self.session, 'foo')
|
||||
self.assertEqual(value, 'bar')
|
||||
def test_basic_value(self):
|
||||
self.session.execute(sa.text("insert into setting values ('foo', 'bar');"))
|
||||
value = conf.get_setting(self.session, 'foo')
|
||||
self.assertEqual(value, 'bar')
|
||||
|
||||
def test_missing_value(self):
|
||||
value = conf.get_setting(self.session, 'foo')
|
||||
self.assertIsNone(value)
|
||||
def test_missing_value(self):
|
||||
value = conf.get_setting(self.session, 'foo')
|
||||
self.assertIsNone(value)
|
||||
|
||||
|
||||
class TestMakeEngineFromConfig(TestCase):
|
||||
class TestMakeEngineFromConfig(TestCase):
|
||||
|
||||
def test_basic(self):
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
})
|
||||
self.assertIsInstance(engine, Engine)
|
||||
def test_basic(self):
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
})
|
||||
self.assertIsInstance(engine, Engine)
|
||||
|
||||
def test_poolclass(self):
|
||||
def test_poolclass(self):
|
||||
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
})
|
||||
self.assertNotIsInstance(engine.pool, NullPool)
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
})
|
||||
self.assertNotIsInstance(engine.pool, NullPool)
|
||||
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
'sqlalchemy.poolclass': 'sqlalchemy.pool:NullPool',
|
||||
})
|
||||
self.assertIsInstance(engine.pool, NullPool)
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
'sqlalchemy.poolclass': 'sqlalchemy.pool:NullPool',
|
||||
})
|
||||
self.assertIsInstance(engine.pool, NullPool)
|
||||
|
||||
def test_pool_pre_ping(self):
|
||||
def test_pool_pre_ping(self):
|
||||
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
})
|
||||
self.assertFalse(engine.pool._pre_ping)
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
})
|
||||
self.assertFalse(engine.pool._pre_ping)
|
||||
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
'sqlalchemy.pool_pre_ping': 'true',
|
||||
})
|
||||
self.assertTrue(engine.pool._pre_ping)
|
||||
engine = conf.make_engine_from_config({
|
||||
'sqlalchemy.url': 'sqlite://',
|
||||
'sqlalchemy.pool_pre_ping': 'true',
|
||||
})
|
||||
self.assertTrue(engine.pool._pre_ping)
|
||||
|
|
|
|||
|
|
@ -3,52 +3,55 @@
|
|||
from unittest import TestCase
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import orm
|
||||
|
||||
from wuttjamaican.db import sess
|
||||
from wuttjamaican.conf import WuttaConfig
|
||||
|
||||
try:
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import orm
|
||||
from wuttjamaican.db import sess
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
|
||||
class TestShortSession(TestCase):
|
||||
class TestShortSession(TestCase):
|
||||
|
||||
def test_none(self):
|
||||
with sess.short_session() as s:
|
||||
self.assertIsInstance(s, sess.Session.class_)
|
||||
def test_none(self):
|
||||
with sess.short_session() as s:
|
||||
self.assertIsInstance(s, sess.Session.class_)
|
||||
|
||||
def test_factory(self):
|
||||
TestSession = orm.sessionmaker()
|
||||
with sess.short_session(factory=TestSession) as s:
|
||||
self.assertIsInstance(s, TestSession.class_)
|
||||
def test_factory(self):
|
||||
TestSession = orm.sessionmaker()
|
||||
with sess.short_session(factory=TestSession) as s:
|
||||
self.assertIsInstance(s, TestSession.class_)
|
||||
|
||||
def test_instance(self):
|
||||
# nb. nothing really happens if we provide the session instance
|
||||
session = MagicMock()
|
||||
with sess.short_session(session=session) as s:
|
||||
pass
|
||||
session.commit.assert_not_called()
|
||||
session.close.assert_not_called()
|
||||
def test_instance(self):
|
||||
# nb. nothing really happens if we provide the session instance
|
||||
session = MagicMock()
|
||||
with sess.short_session(session=session) as s:
|
||||
pass
|
||||
session.commit.assert_not_called()
|
||||
session.close.assert_not_called()
|
||||
|
||||
def test_config(self):
|
||||
config = MagicMock()
|
||||
TestSession = orm.sessionmaker()
|
||||
config.get_app.return_value.make_session = TestSession
|
||||
# nb. config may be first arg (or kwarg)
|
||||
with sess.short_session(config) as s:
|
||||
self.assertIsInstance(s, TestSession.class_)
|
||||
def test_config(self):
|
||||
config = MagicMock()
|
||||
TestSession = orm.sessionmaker()
|
||||
config.get_app.return_value.make_session = TestSession
|
||||
# nb. config may be first arg (or kwarg)
|
||||
with sess.short_session(config) as s:
|
||||
self.assertIsInstance(s, TestSession.class_)
|
||||
|
||||
def test_without_commit(self):
|
||||
session = MagicMock()
|
||||
TestSession = MagicMock(return_value=session)
|
||||
with sess.short_session(factory=TestSession, commit=False) as s:
|
||||
pass
|
||||
session.commit.assert_not_called()
|
||||
session.close.assert_called_once_with()
|
||||
def test_without_commit(self):
|
||||
session = MagicMock()
|
||||
TestSession = MagicMock(return_value=session)
|
||||
with sess.short_session(factory=TestSession, commit=False) as s:
|
||||
pass
|
||||
session.commit.assert_not_called()
|
||||
session.close.assert_called_once_with()
|
||||
|
||||
def test_with_commit(self):
|
||||
session = MagicMock()
|
||||
TestSession = MagicMock(return_value=session)
|
||||
with sess.short_session(factory=TestSession, commit=True) as s:
|
||||
pass
|
||||
session.commit.assert_called_once_with()
|
||||
session.close.assert_called_once_with()
|
||||
def test_with_commit(self):
|
||||
session = MagicMock()
|
||||
TestSession = MagicMock(return_value=session)
|
||||
with sess.short_session(factory=TestSession, commit=True) as s:
|
||||
pass
|
||||
session.commit.assert_called_once_with()
|
||||
session.close.assert_called_once_with()
|
||||
|
|
|
|||
|
|
@ -7,10 +7,9 @@ import warnings
|
|||
from unittest import TestCase
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import orm
|
||||
import pytest
|
||||
|
||||
from wuttjamaican import app, db
|
||||
from wuttjamaican import app
|
||||
from wuttjamaican.conf import WuttaConfig
|
||||
|
||||
|
||||
|
|
@ -46,6 +45,11 @@ class TestAppHandler(TestCase):
|
|||
shutil.rmtree(tempdir)
|
||||
|
||||
def test_make_session(self):
|
||||
try:
|
||||
from wuttjamaican import db
|
||||
except ImportError:
|
||||
pytest.skip("test is not relevant without sqlalchemy")
|
||||
|
||||
session = self.app.make_session()
|
||||
self.assertIsInstance(session, db.Session.class_)
|
||||
|
||||
|
|
@ -60,6 +64,12 @@ class TestAppHandler(TestCase):
|
|||
foo='bar', factory=self.app.make_session)
|
||||
|
||||
def test_get_setting(self):
|
||||
try:
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import orm
|
||||
except ImportError:
|
||||
pytest.skip("test is not relevant without sqlalchemy")
|
||||
|
||||
Session = orm.sessionmaker()
|
||||
engine = sa.create_engine('sqlite://')
|
||||
session = Session(bind=engine)
|
||||
|
|
|
|||
|
|
@ -5,12 +5,10 @@ import os
|
|||
from unittest import TestCase
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import sqlalchemy as sa
|
||||
import pytest
|
||||
|
||||
from wuttjamaican import conf
|
||||
from wuttjamaican.exc import ConfigurationError
|
||||
from wuttjamaican.db import Session
|
||||
from wuttjamaican.db.conf import make_engine_from_config
|
||||
from wuttjamaican.app import AppHandler
|
||||
from wuttjamaican.testing import FileConfigTestCase
|
||||
|
||||
|
|
@ -133,6 +131,13 @@ require = %(here)s/first.conf
|
|||
self.assertEqual(config.get('foo'), 'bar')
|
||||
|
||||
def test_constructor_db_flags(self):
|
||||
try:
|
||||
# nb. we don't need this import but the test will not
|
||||
# behave correctly unless the lib is installed
|
||||
import sqlalchemy
|
||||
except ImportError:
|
||||
pytest.skip("test is not relevant without sqlalchemy")
|
||||
|
||||
myfile = self.write_file('my.conf', """\
|
||||
[wutta.config]
|
||||
usedb = true
|
||||
|
|
@ -155,6 +160,12 @@ preferdb = true
|
|||
self.assertTrue(config.preferdb)
|
||||
|
||||
def test_constructor_db_not_supported(self):
|
||||
try:
|
||||
# nb. we don't need this import but the test will not
|
||||
# behave correctly unless the lib is installed
|
||||
import sqlalchemy
|
||||
except ImportError:
|
||||
pytest.skip("test is not relevant without sqlalchemy")
|
||||
|
||||
# flags are off by default
|
||||
config = conf.WuttaConfig()
|
||||
|
|
@ -269,6 +280,12 @@ configure_logging = true
|
|||
self.assertEqual(config.get('foo', default='bar'), 'bar')
|
||||
|
||||
def test_get_from_db(self):
|
||||
try:
|
||||
import sqlalchemy as sa
|
||||
from wuttjamaican.db import Session
|
||||
except ImportError:
|
||||
pytest.skip("test is not relevant without sqlalchemy")
|
||||
|
||||
# minimal config, but at least it needs db cxn info
|
||||
config = conf.WuttaConfig(defaults={'wutta.db.default.url': 'sqlite://'})
|
||||
|
||||
|
|
@ -317,6 +334,11 @@ configure_logging = true
|
|||
self.assertIn("makin stuff up", str(error))
|
||||
|
||||
def test_get_preferdb(self):
|
||||
try:
|
||||
import sqlalchemy as sa
|
||||
from wuttjamaican.db import Session
|
||||
except ImportError:
|
||||
pytest.skip("test is not relevant without sqlalchemy")
|
||||
|
||||
# start out with a default value
|
||||
config = conf.WuttaConfig(defaults={'wutta.db.default.url': 'sqlite://',
|
||||
|
|
@ -403,6 +425,10 @@ configure_logging = true
|
|||
self.assertIsInstance(app, CustomAppHandler)
|
||||
|
||||
def test_get_engine_maker(self):
|
||||
try:
|
||||
from wuttjamaican.db.conf import make_engine_from_config
|
||||
except ImportError:
|
||||
pytest.skip("test is not relevant without sqlalchemy")
|
||||
|
||||
# default func
|
||||
config = conf.WuttaConfig()
|
||||
|
|
@ -421,7 +447,7 @@ class CustomAppHandler(AppHandler):
|
|||
pass
|
||||
|
||||
|
||||
def custom_make_engine_from_config(*args, **kwargs):
|
||||
def custom_make_engine_from_config():
|
||||
pass
|
||||
|
||||
|
||||
|
|
|
|||
5
tox.ini
5
tox.ini
|
|
@ -1,6 +1,6 @@
|
|||
|
||||
[tox]
|
||||
envlist = py36, py37, py38, py39, py310, py311
|
||||
envlist = py36, py37, py38, py39, py310, py311, nodb
|
||||
|
||||
# TODO: can remove this when we drop py36 support
|
||||
# nb. need this for testing older python versions
|
||||
|
|
@ -15,6 +15,9 @@ commands = pytest {posargs}
|
|||
# nb. newer coverage is causing segfault for this one, so must avoid that
|
||||
deps = coverage<6.5
|
||||
|
||||
[testenv:nodb]
|
||||
extras = tests
|
||||
|
||||
[testenv:coverage]
|
||||
basepython = python3.11
|
||||
extras = db,tests
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue