3
0
Fork 0

feat: add basic batch feature, data model and partial handler

hopefully data model is complete enough for now, but handler does not
yet have all methods, e.g. execute()
This commit is contained in:
Lance Edgar 2024-12-13 20:38:00 -06:00
parent 51accc5a93
commit a514d9cfba
9 changed files with 813 additions and 1 deletions

View file

@ -0,0 +1,52 @@
# -*- coding: utf-8; -*-
import uuid as _uuid
from wuttjamaican.testing import DataTestCase
try:
import sqlalchemy as sa
from wuttjamaican.db import model
from wuttjamaican.db.model import batch as mod
except ImportError:
pass
else:
class TestBatchMixin(DataTestCase):
def test_basic(self):
class MyBatch(mod.BatchMixin, model.Base):
__tablename__ = 'testing_mybatch'
model.Base.metadata.create_all(bind=self.session.bind)
metadata = sa.MetaData()
metadata.reflect(self.session.bind)
self.assertIn('testing_mybatch', metadata.tables)
batch = MyBatch(id=42, uuid=_uuid.UUID('0675cdac-ffc9-7690-8000-6023de1c8cfd'))
self.assertEqual(repr(batch), "MyBatch(uuid=UUID('0675cdac-ffc9-7690-8000-6023de1c8cfd'))")
self.assertEqual(str(batch), "00000042")
class TestBatchRowMixin(DataTestCase):
def test_basic(self):
class MyBatch2(mod.BatchMixin, model.Base):
__tablename__ = 'testing_mybatch2'
class MyBatchRow2(mod.BatchRowMixin, model.Base):
__tablename__ = 'testing_mybatch_row2'
__batch_class__ = MyBatch2
model.Base.metadata.create_all(bind=self.session.bind)
metadata = sa.MetaData()
metadata.reflect(self.session.bind)
self.assertIn('testing_mybatch2', metadata.tables)
self.assertIn('testing_mybatch_row2', metadata.tables)
# nb. this gives coverage but doesn't really test much
batch = MyBatch2(id=42, uuid=_uuid.UUID('0675cdac-ffc9-7690-8000-6023de1c8cfd'))
row = MyBatchRow2()
batch.rows.append(row)

70
tests/test_batch.py Normal file
View file

@ -0,0 +1,70 @@
# -*- coding: utf-8; -*-
from wuttjamaican import batch as mod
try:
import sqlalchemy as sa
from wuttjamaican.db import model
from wuttjamaican.testing import DataTestCase
except ImportError:
pass
else:
class MockBatch(model.BatchMixin, model.Base):
__tablename__ = 'testing_batch_mock'
class MockBatchRow(model.BatchRowMixin, model.Base):
__tablename__ = 'testing_batch_mock_row'
__batch_class__ = MockBatch
class MockBatchHandler(mod.BatchHandler):
model_class = MockBatch
class TestBatchHandler(DataTestCase):
def make_handler(self, **kwargs):
return MockBatchHandler(self.config, **kwargs)
def test_model_class(self):
handler = mod.BatchHandler(self.config)
self.assertRaises(NotImplementedError, getattr, handler, 'model_class')
def test_make_batch(self):
handler = self.make_handler()
batch = handler.make_batch(self.session)
self.assertIsInstance(batch, MockBatch)
def test_consume_batch_id(self):
handler = self.make_handler()
first = handler.consume_batch_id(self.session)
second = handler.consume_batch_id(self.session)
self.assertEqual(second, first + 1)
third = handler.consume_batch_id(self.session, as_str=True)
self.assertEqual(third, f'{first + 2:08d}')
def test_should_populate(self):
handler = self.make_handler()
batch = handler.make_batch(self.session)
self.assertFalse(handler.should_populate(batch))
def test_do_populate(self):
handler = self.make_handler()
batch = handler.make_batch(self.session)
# nb. coverage only; tests nothing
handler.do_populate(batch)
def test_make_row(self):
handler = self.make_handler()
row = handler.make_row()
self.assertIsInstance(row, MockBatchRow)
def test_add_row(self):
handler = self.make_handler()
batch = handler.make_batch(self.session)
self.session.add(batch)
row = handler.make_row()
self.assertIsNone(batch.row_count)
handler.add_row(batch, row)
self.assertEqual(batch.row_count, 1)