# -*- coding: utf-8; -*- import datetime from unittest.mock import patch, MagicMock from pyramid.httpexceptions import HTTPFound from wuttjamaican.db import model from wuttjamaican.batch import BatchHandler from wuttaweb.views import batch as mod from wuttaweb.progress import SessionProgress from tests.util import WebTestCase 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(BatchHandler): model_class = MockBatch class TestBatchMasterView(WebTestCase): def test_get_batch_handler(self): self.assertRaises(NotImplementedError, mod.BatchMasterView, self.request) with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=42): view = mod.BatchMasterView(self.request) self.assertEqual(view.batch_handler, 42) def test_configure_grid(self): handler = MockBatchHandler(self.config) with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch): with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = mod.BatchMasterView(self.request) grid = view.make_model_grid() # nb. coverage only; tests nothing view.configure_grid(grid) def test_render_batch_id(self): handler = MockBatchHandler(self.config) with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = mod.BatchMasterView(self.request) batch = MockBatch(id=42) result = view.render_batch_id(batch, 'id', 42) self.assertEqual(result, '00000042') result = view.render_batch_id(batch, 'id', None) self.assertIsNone(result) def test_get_instance_title(self): handler = MockBatchHandler(self.config) with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = mod.BatchMasterView(self.request) batch = MockBatch(id=42) result = view.get_instance_title(batch) self.assertEqual(result, "00000042") batch = MockBatch(id=43, description="runnin some numbers") result = view.get_instance_title(batch) self.assertEqual(result, "00000043 runnin some numbers") def test_configure_form(self): handler = MockBatchHandler(self.config) with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch): with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = mod.BatchMasterView(self.request) # creating with patch.object(view, 'creating', new=True): form = view.make_model_form(model_instance=None) view.configure_form(form) batch = MockBatch(id=42) # viewing with patch.object(view, 'viewing', new=True): form = view.make_model_form(model_instance=batch) view.configure_form(form) # editing with patch.object(view, 'editing', new=True): form = view.make_model_form(model_instance=batch) view.configure_form(form) # deleting with patch.object(view, 'deleting', new=True): form = view.make_model_form(model_instance=batch) view.configure_form(form) # viewing (executed) batch.executed = datetime.datetime.now() with patch.object(view, 'viewing', new=True): form = view.make_model_form(model_instance=batch) view.configure_form(form) def test_objectify(self): handler = MockBatchHandler(self.config) with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch): with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): with patch.object(mod.BatchMasterView, 'Session', return_value=self.session): view = mod.BatchMasterView(self.request) # create batch with patch.object(view, 'creating', new=True): form = view.make_model_form(model_instance=None) form.validated = {} batch = view.objectify(form) self.assertIsInstance(batch.id, int) self.assertTrue(batch.id > 0) # edit batch with patch.object(view, 'editing', new=True): with patch.object(view.batch_handler, 'make_batch') as make_batch: form = view.make_model_form(model_instance=batch) form.validated = {'description': 'foo'} self.assertIsNone(batch.description) batch = view.objectify(form) self.assertEqual(batch.description, 'foo') def test_redirect_after_create(self): self.pyramid_config.add_route('mock_batches.view', '/batch/mock/{uuid}') handler = MockBatchHandler(self.config) with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch, route_prefix='mock_batches'): view = mod.BatchMasterView(self.request) batch = MockBatch(id=42) # typically redirect to view batch result = view.redirect_after_create(batch) self.assertIsInstance(result, HTTPFound) # unless populating in which case thread is launched self.request.session.id = 'abcdefghijk' with patch.object(mod, 'threading') as threading: thread = MagicMock() threading.Thread.return_value = thread with patch.object(view.batch_handler, 'should_populate', return_value=True): with patch.object(view, 'render_progress') as render_progress: view.redirect_after_create(batch) self.assertTrue(threading.Thread.called) thread.start.assert_called_once_with() self.assertTrue(render_progress.called) def test_populate_thread(self): model = self.app.model handler = MockBatchHandler(self.config) with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch): view = mod.BatchMasterView(self.request) user = model.User(username='barney') self.session.add(user) batch = MockBatch(id=42, created_by=user) self.session.add(batch) self.session.commit() # nb. use our session within thread method with patch.object(self.app, 'make_session', return_value=self.session): # nb. prevent closing our session with patch.object(self.session, 'close') as close: # without progress view.populate_thread(batch.uuid) close.assert_called_once_with() close.reset_mock() # with progress self.request.session.id = 'abcdefghijk' view.populate_thread(batch.uuid, progress=SessionProgress(self.request, 'populate_mock_batch')) close.assert_called_once_with() close.reset_mock() # failure to populate, without progress with patch.object(view.batch_handler, 'do_populate', side_effect=RuntimeError): view.populate_thread(batch.uuid) close.assert_called_once_with() close.reset_mock() # failure to populate, with progress with patch.object(view.batch_handler, 'do_populate', side_effect=RuntimeError): view.populate_thread(batch.uuid, progress=SessionProgress(self.request, 'populate_mock_batch')) close.assert_called_once_with() close.reset_mock() # failure for batch to appear self.session.delete(batch) self.session.commit() # nb. should give up waiting after 1 second self.assertRaises(RuntimeError, view.populate_thread, batch.uuid)