# -*- coding: utf-8; -*- import datetime from unittest.mock import patch, MagicMock from sqlalchemy import orm from pyramid.httpexceptions import HTTPFound from wuttjamaican.db import model from wuttjamaican.batch import BatchHandler from wuttaweb.views import MasterView, batch as mod from wuttaweb.progress import SessionProgress from wuttaweb.testing 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 MockBatch.__row_class__ = MockBatchRow class MockBatchHandler(BatchHandler): model_class = MockBatch class TestBatchMasterView(WebTestCase): def setUp(self): self.setup_web() # nb. create MockBatch, MockBatchRow model.Base.metadata.create_all(bind=self.session.bind) def make_handler(self): return MockBatchHandler(self.config) def make_view(self): return mod.BatchMasterView(self.request) 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_get_fallback_templates(self): handler = MockBatchHandler(self.config) with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = self.make_view() templates = view.get_fallback_templates('view') self.assertEqual(templates, [ '/batch/view.mako', '/master/view.mako', ]) def test_render_to_response(self): model = self.app.model handler = MockBatchHandler(self.config) user = model.User(username='barney') self.session.add(user) batch = handler.make_batch(self.session, created_by=user) self.session.add(batch) self.session.flush() with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): with patch.object(MasterView, 'render_to_response') as render_to_response: view = self.make_view() response = view.render_to_response('view', {'instance': batch}) self.assertTrue(render_to_response.called) context = render_to_response.call_args[0][1] self.assertIs(context['batch'], batch) self.assertIs(context['batch_handler'], handler) 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_delete_instance(self): model = self.app.model handler = self.make_handler() user = model.User(username='barney') self.session.add(user) batch = handler.make_batch(self.session, created_by=user) self.session.add(batch) self.session.flush() with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = self.make_view() self.assertEqual(self.session.query(MockBatch).count(), 1) view.delete_instance(batch) self.assertEqual(self.session.query(MockBatch).count(), 0) 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) def test_execute(self): self.pyramid_config.add_route('mock_batches.view', '/batch/mock/{uuid}') model = self.app.model handler = MockBatchHandler(self.config) user = model.User(username='barney') self.session.add(user) batch = handler.make_batch(self.session, created_by=user) self.session.add(batch) self.session.commit() with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch, route_prefix='mock_batches', get_batch_handler=MagicMock(return_value=handler), get_instance=MagicMock(return_value=batch)): view = self.make_view() # batch executes okay response = view.execute() self.assertEqual(response.status_code, 302) # redirect to "view batch" self.assertFalse(self.request.session.peek_flash('error')) # but cannot be executed again response = view.execute() self.assertEqual(response.status_code, 302) # redirect to "view batch" # nb. flash has error this time self.assertTrue(self.request.session.peek_flash('error')) def test_get_row_model_class(self): handler = MockBatchHandler(self.config) with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = self.make_view() self.assertRaises(AttributeError, view.get_row_model_class) # row class determined from batch class with patch.object(mod.BatchMasterView, 'model_class', new=MockBatch, create=True): cls = view.get_row_model_class() self.assertIs(cls, MockBatchRow) self.assertRaises(AttributeError, view.get_row_model_class) # view may specify row class with patch.object(mod.BatchMasterView, 'row_model_class', new=MockBatchRow, create=True): cls = view.get_row_model_class() self.assertIs(cls, MockBatchRow) def test_get_row_grid_data(self): handler = MockBatchHandler(self.config) model = self.app.model user = model.User(username='barney') self.session.add(user) batch = handler.make_batch(self.session, created_by=user) self.session.add(batch) row = handler.make_row() handler.add_row(batch, row) self.session.flush() with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): view = self.make_view() self.assertRaises(AttributeError, view.get_row_grid_data, batch) Session = MagicMock(return_value=self.session) Session.query.side_effect = lambda m: self.session.query(m) with patch.multiple(mod.BatchMasterView, create=True, Session=Session, model_class=MockBatch): view = self.make_view() data = view.get_row_grid_data(batch) self.assertIsInstance(data, orm.Query) self.assertEqual(data.count(), 1) def test_configure_row_grid(self): handler = MockBatchHandler(self.config) model = self.app.model user = model.User(username='barney') self.session.add(user) batch = handler.make_batch(self.session, created_by=user) self.session.add(batch) row = handler.make_row() handler.add_row(batch, row) self.session.flush() with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=handler): Session = MagicMock(return_value=self.session) Session.query.side_effect = lambda m: self.session.query(m) with patch.multiple(mod.BatchMasterView, create=True, Session=Session, model_class=MockBatch): with patch.object(self.request, 'matchdict', new={'uuid': batch.uuid}): view = self.make_view() grid = view.make_row_model_grid(batch) self.assertIn('sequence', grid.labels) self.assertEqual(grid.labels['sequence'], "Seq.") def test_render_row_status(self): with patch.object(mod.BatchMasterView, 'get_batch_handler', return_value=None): view = self.make_view() row = MagicMock(foo=1, STATUS={1: 'bar'}) self.assertEqual(view.render_row_status(row, 'foo', 1), 'bar') def test_defaults(self): # nb. coverage only with patch.object(mod.BatchMasterView, 'model_class', new=MockBatch, create=True): mod.BatchMasterView.defaults(self.pyramid_config)