wutta/wuttjamaican
3
0
Fork 0
Code Activity

fix: add basic execution methods for batch handler

Browse source 
also logic for batch data files, and deletion
This commit is contained in:
Lance Edgar 2024-12-14 19:39:02 -06:00
parent a514d9cfba
commit 3585eca65b
8 changed files with 402 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

8
tests/test_app.py
View file

@ -434,6 +434,14 @@ app_title = WuttaTest
dt = datetime.datetime(2024, 12, 11, 8, 30, tzinfo=datetime.timezone.utc)
self.assertEqual(self.app.render_datetime(dt), '2024-12-11 08:30+0000')
def test_render_time_ago(self):
with patch.object(mod, 'humanize') as humanize:
humanize.naturaltime.return_value = 'now'
now = datetime.datetime.now()
result = self.app.render_time_ago(now)
self.assertEqual(result, 'now')
humanize.naturaltime.assert_called_once_with(now)
def test_get_person(self):
people = self.app.get_people_handler()
with patch.object(people, 'get_person') as get_person:

128
tests/test_batch.py
View file

@ -1,5 +1,8 @@
# -*- coding: utf-8; -*-
import os
from unittest.mock import patch
from wuttjamaican import batch as mod
try:
@ -29,6 +32,11 @@ else:
handler = mod.BatchHandler(self.config)
self.assertRaises(NotImplementedError, getattr, handler, 'model_class')
def test_batch_type(self):
with patch.object(mod.BatchHandler, 'model_class', new=MockBatch):
handler = mod.BatchHandler(self.config)
self.assertEqual(handler.batch_type, 'testing_batch_mock')
def test_make_batch(self):
handler = self.make_handler()
batch = handler.make_batch(self.session)
@ -44,6 +52,44 @@ else:
third = handler.consume_batch_id(self.session, as_str=True)
self.assertEqual(third, f'{first + 2:08d}')
def test_get_data_path(self):
model = self.app.model
user = model.User(username='barney')
self.session.add(user)
with patch.object(mod.BatchHandler, 'model_class', new=MockBatch):
handler = self.make_handler()
# root storage (default)
with patch.object(self.app, 'get_appdir', return_value=self.tempdir):
path = handler.get_data_path()
self.assertEqual(path, os.path.join(self.tempdir, 'data', 'batch', 'testing_batch_mock'))
# root storage (configured)
self.config.setdefault('wutta.batch.storage_path', self.tempdir)
path = handler.get_data_path()
self.assertEqual(path, os.path.join(self.tempdir, 'testing_batch_mock'))
batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch)
self.session.flush()
# batch-specific
path = handler.get_data_path(batch)
uuid = batch.uuid.hex
final = os.path.join(uuid[-2:], uuid[:-2])
self.assertEqual(path, os.path.join(self.tempdir, 'testing_batch_mock', final))
# with filename
path = handler.get_data_path(batch, 'input.csv')
self.assertEqual(path, os.path.join(self.tempdir, 'testing_batch_mock', final, 'input.csv'))
# makedirs
path = handler.get_data_path(batch)
self.assertFalse(os.path.exists(path))
path = handler.get_data_path(batch, makedirs=True)
self.assertTrue(os.path.exists(path))
def test_should_populate(self):
handler = self.make_handler()
batch = handler.make_batch(self.session)
@ -68,3 +114,85 @@ else:
self.assertIsNone(batch.row_count)
handler.add_row(batch, row)
self.assertEqual(batch.row_count, 1)
def test_do_execute(self):
model = self.app.model
user = model.User(username='barney')
self.session.add(user)
handler = self.make_handler()
batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch)
self.session.flush()
# error if execution not allowed
with patch.object(handler, 'why_not_execute', return_value="bad batch"):
self.assertRaises(RuntimeError, handler.do_execute, batch, user)
# nb. coverage only; tests nothing
self.assertIsNone(batch.executed)
self.assertIsNone(batch.executed_by)
handler.do_execute(batch, user)
self.assertIsNotNone(batch.executed)
self.assertIs(batch.executed_by, user)
# error if execution already happened
self.assertRaises(ValueError, handler.do_execute, batch, user)
def test_do_delete(self):
model = self.app.model
handler = self.make_handler()
user = model.User(username='barney')
self.session.add(user)
# simple delete
batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch)
self.session.flush()
self.assertEqual(self.session.query(MockBatch).count(), 1)
handler.do_delete(batch, user)
self.assertEqual(self.session.query(MockBatch).count(), 0)
# delete w/ rows
batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch)
for i in range(5):
row = handler.make_row()
handler.add_row(batch, row)
self.session.flush()
self.assertEqual(self.session.query(MockBatch).count(), 1)
handler.do_delete(batch, user)
self.assertEqual(self.session.query(MockBatch).count(), 0)
# delete w/ files
self.config.setdefault('wutta.batch.storage_path', self.tempdir)
batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch)
self.session.flush()
path = handler.get_data_path(batch, 'data.txt', makedirs=True)
with open(path, 'wt') as f:
f.write('foo=bar')
self.assertEqual(self.session.query(MockBatch).count(), 1)
path = handler.get_data_path(batch)
self.assertTrue(os.path.exists(path))
handler.do_delete(batch, user)
self.assertEqual(self.session.query(MockBatch).count(), 0)
self.assertFalse(os.path.exists(path))
# delete w/ files (dry-run)
self.config.setdefault('wutta.batch.storage_path', self.tempdir)
batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch)
self.session.flush()
path = handler.get_data_path(batch, 'data.txt', makedirs=True)
with open(path, 'wt') as f:
f.write('foo=bar')
self.assertEqual(self.session.query(MockBatch).count(), 1)
path = handler.get_data_path(batch)
self.assertTrue(os.path.exists(path))
handler.do_delete(batch, user, dry_run=True)
# nb. batch appears missing from session even in dry-run
self.assertEqual(self.session.query(MockBatch).count(), 0)
# nb. but its files remain intact
self.assertTrue(os.path.exists(path))