Changeset - 98a062541871
[Not reviewed]
0 2 0
Lance Edgar (lance) - 11 years ago 2013-08-29 17:03:37
lance@edbob.org
Added some `CustomerEmailAddress` tests, removed some unused tests.
2 files changed with 51 insertions and 89 deletions:
0 comments (0 inline, 0 general)
tests/db/test_changes.py
Show inline comments
 
@@ -89,137 +89,48 @@ class TestChangeRecorder(TestCase):
 
        with patch('rattail.db.changes.Change') as MockChange:
 
            new_change = Mock()
 
            MockChange.return_value = new_change
 
            self.assertTrue(recorder.record_change(session, Product()))
 
            session.add.assert_called_once_with(new_change)
 

	
 
    @patch.multiple('rattail.db.changes', get_uuid=DEFAULT, object_mapper=DEFAULT)
 
    def test_ensure_uuid(self, get_uuid, object_mapper):
 
        recorder = changes.ChangeRecorder()
 
        uuid_column = Mock()
 
        object_mapper.return_value.columns.__getitem__.return_value = uuid_column
 

	
 
        # uuid already present
 
        product = Product(uuid='some_uuid')
 
        recorder.ensure_uuid(product)
 
        self.assertEqual(product.uuid, 'some_uuid')
 
        self.assertFalse(get_uuid.called)
 

	
 
        # no uuid yet, auto-generate
 
        uuid_column.foreign_keys = False
 
        get_uuid.return_value = 'another_uuid'
 
        product = Product()
 
        self.assertIsNone(product.uuid)
 
        recorder.ensure_uuid(product)
 
        get_uuid.assert_called_once_with()
 
        self.assertEqual(product.uuid, 'another_uuid')
 

	
 
        # some heavy mocking for following tests
 
        uuid_column.foreign_keys = True
 
        remote_side = MagicMock(key='uuid')
 
        prop = MagicMock(__class__=RelationshipProperty, key='foreign_thing')
 
        prop.remote_side.__len__.return_value = 1
 
        prop.remote_side.__iter__.return_value = [remote_side]
 
        object_mapper.return_value.iterate_properties.__iter__.return_value = [prop]
 
        
 
        # uuid fetched from existing foreign key object
 
        get_uuid.reset_mock()
 
        instance = Mock(uuid=None, foreign_thing=Mock(uuid='secondary_uuid'))
 
        recorder.ensure_uuid(instance)
 
        self.assertFalse(get_uuid.called)
 
        self.assertEqual(instance.uuid, 'secondary_uuid')
 

	
 
        # foreign key object doesn't exist; uuid generated as fallback
 
        get_uuid.return_value = 'fallback_uuid'
 
        instance = Mock(uuid=None, foreign_thing=None)
 
        recorder.ensure_uuid(instance)
 
        get_uuid.assert_called_once_with()
 
        self.assertEqual(instance.uuid, 'fallback_uuid')
 

	
 
    #     # uuid fetched from foreign key object
 
    #     product = Product(uuid='latest_uuid')
 
    #     product_ext = ProductExtTest(product=product)
 
    #     self.assertIsNone(product_ext.uuid)
 
    #     get_uuid.reset_mock()
 
    #     changes.ensure_uuid(product_ext)
 
    #     self.assertFalse(get_uuid.called)
 
    #     self.assertEqual(product_ext.uuid, 'latest_uuid')
 

	
 
    # @patch('rattail.db.changes.get_uuid')
 
    # def test_ensure_uuid(self, get_uuid):
 

	
 
    #     # fallback scenario.  "extended product" is expected to be associated
 
    #     # with a product, but if it isn't, a UUID must still be generated
 
    #     product_ext = ProductExtTest()
 
    #     self.assertIsNone(product_ext.uuid)
 
    #     changes.ensure_uuid(product_ext)
 
    #     get_uuid.assert_called_once_with()
 
    #     self.assertEqual(product_ext.uuid, 'some_uuid')
 

	
 

	
 
# class ProductExtTest(Base):
 
#     __tablename__ = 'products_extended_test'
 
#     uuid = uuid_column(ForeignKey('products.uuid'), default=None)
 
#     some_attr = Column(String(length=10))
 
#     product = relationship(Product)
 

	
 

	
 
# class TestChanges(DataTestCase):
 

	
 
#     def setUp(self):
 
#         Base.metadata.tables['products_extended_test'].create(bind=engine)
 
#         super(TestChanges, self).setUp()
 

	
 
#     def tearDown(self):
 
#         super(TestChanges, self).tearDown()
 
#         Base.metadata.tables['products_extended_test'].drop(bind=engine)
 

	
 
#     @patch('rattail.db.changes.listen')
 
#     def test_record_changes(self, listen):
 
#         session = Mock()
 

	
 
#         changes.record_changes(session)
 

	
 
#         # # changes not being recorded yet
 
#         # self.session.add(Product())
 
#         # self.session.commit()
 
#         # self.assertEqual(self.session.query(Change).count(), 0)
 

	
 
#         # # okay, record them and confirm
 
#         # changes.record_changes(self.session)
 
#         # self.session.add(Product())
 
#         # self.session.commit()
 
#         # self.assertEqual(self.session.query(Change).count(), 1)
 

	
 
#     @patch('rattail.db.changes.get_uuid')
 
#     def test_ensure_uuid(self, get_uuid):
 

	
 
#         # uuid already present
 
#         product = Product(uuid='whatever')
 
#         changes.ensure_uuid(product)
 
#         self.assertEqual(product.uuid, 'whatever')
 
#         self.assertFalse(get_uuid.called)
 

	
 
#         # no uuid yet, auto-generate
 
#         get_uuid.return_value = 'some_uuid'
 
#         product = Product()
 
#         self.assertIsNone(product.uuid)
 
#         changes.ensure_uuid(product)
 
#         get_uuid.assert_called_once_with()
 
#         self.assertEqual(product.uuid, 'some_uuid')
 

	
 
#         # uuid fetched from foreign key object
 
#         product = Product(uuid='latest_uuid')
 
#         product_ext = ProductExtTest(product=product)
 
#         self.assertIsNone(product_ext.uuid)
 
#         get_uuid.reset_mock()
 
#         changes.ensure_uuid(product_ext)
 
#         self.assertFalse(get_uuid.called)
 
#         self.assertEqual(product_ext.uuid, 'latest_uuid')
 

	
 
#         # fallback scenario.  "extended product" is expected to be associated
 
#         # with a product, but if it isn't, a UUID must still be generated
 
#         product_ext = ProductExtTest()
 
#         self.assertIsNone(product_ext.uuid)
 
#         changes.ensure_uuid(product_ext)
 
#         get_uuid.assert_called_once_with()
 
#         self.assertEqual(product_ext.uuid, 'some_uuid')
tests/db/test_model.py
Show inline comments
 

	
 
from unittest import TestCase
 
from . import DataTestCase
 
from mock import patch, DEFAULT, Mock, MagicMock
 

	
 
from rattail.db.extension import model
 
from sqlalchemy import String, Boolean, Numeric
 
from rattail.db.types import GPCType
 
from sqlalchemy.exc import IntegrityError
 
from rattail.db import record_changes
 

	
 

	
 
class TestBatch(TestCase):
 

	
 
    @patch('rattail.db.extension.model.object_session')
 
    def test_rowclass(self, object_session):
 
        object_session.return_value = object_session
 

	
 
        # no row classes to start with
 
        self.assertEqual(model.Batch._rowclasses, {})
 

	
 
        # basic, empty row class
 
        batch = model.Batch(uuid='some_uuid')
 
        batch.get_sqlalchemy_type = Mock(return_value='some_type')
 
        batch.columns = MagicMock()
 
        rowclass = batch.rowclass
 
        self.assertTrue(issubclass(rowclass, model.BatchRow))
 
        self.assertEqual(model.Batch._rowclasses.keys(), ['some_uuid'])
 
        self.assertIs(model.Batch._rowclasses['some_uuid'], rowclass)
 
        self.assertFalse(object_session.flush.called)
 

	
 
        # make sure rowclass.batch works
 
        object_session.query.return_value.get.return_value = batch
 
        self.assertIs(rowclass().batch, batch)
 
        object_session.query.return_value.get.assert_called_once_with('some_uuid')
 

	
 
        # row class with generated uuid and some columns
 
        batch = model.Batch(uuid=None)
 
        batch.columns = [model.BatchColumn(name='F01'), model.BatchColumn(name='F02')]
 
        model.Batch.get_sqlalchemy_type = Mock(return_value=String(length=20))
 
        def set_uuid():
 
            batch.uuid = 'fresh_uuid'
 
        object_session.flush.side_effect = set_uuid
 
        rowclass = batch.rowclass
 
        object_session.flush.assert_called_once_with()
 
        self.assertItemsEqual(model.Batch._rowclasses.keys(), ['some_uuid', 'fresh_uuid'])
 
        self.assertIs(model.Batch._rowclasses['fresh_uuid'], rowclass)
 

	
 
    def test_get_sqlalchemy_type(self):
 

	
 
        # gpc
 
        self.assertIsInstance(model.Batch.get_sqlalchemy_type('GPC(14)'), GPCType)
 

	
 
        # boolean
 
        self.assertIsInstance(model.Batch.get_sqlalchemy_type('FLAG(1)'), Boolean)
 

	
 
        # string
 
        type_ = model.Batch.get_sqlalchemy_type('CHAR(20)')
 
@@ -128,48 +129,98 @@ class TestCustomerPerson(DataTestCase):
 
        customer._people.append(assoc)
 
        self.session.commit()
 
        self.assertEqual(assoc.ordinal, 1)
 
        assoc = model.CustomerPerson(person=model.Person())
 
        customer._people.append(assoc)
 
        self.session.commit()
 
        self.assertEqual(assoc.ordinal, 2)
 

	
 

	
 
class TestCustomerGroupAssignment(DataTestCase):
 

	
 
    def test_repr(self):
 
        assignment = model.CustomerGroupAssignment(uuid='whatever')
 
        self.assertEqual(repr(assignment), "CustomerGroupAssignment(uuid='whatever')")
 

	
 
    def test_customer_required(self):
 
        assignment = model.CustomerGroupAssignment(group=model.CustomerGroup())
 
        self.session.add(assignment)
 
        self.assertRaises(IntegrityError, self.session.commit)
 
        self.session.rollback()
 
        self.assertEqual(self.session.query(model.CustomerGroupAssignment).count(), 0)
 
        assignment.customer = model.Customer()
 
        self.session.add(assignment)
 
        self.session.commit()
 
        self.assertEqual(self.session.query(model.CustomerGroupAssignment).count(), 1)
 

	
 
    def test_group_required(self):
 
        assignment = model.CustomerGroupAssignment(customer=model.Customer())
 
        self.session.add(assignment)
 
        self.assertRaises(IntegrityError, self.session.commit)
 
        self.session.rollback()
 
        self.assertEqual(self.session.query(model.CustomerGroupAssignment).count(), 0)
 
        assignment.group = model.CustomerGroup()
 
        self.session.add(assignment)
 
        self.session.commit()
 
        self.assertEqual(self.session.query(model.CustomerGroupAssignment).count(), 1)
 

	
 
    def test_ordinal_autoincrement(self):
 
        customer = model.Customer()
 
        self.session.add(customer)
 
        assignment = model.CustomerGroupAssignment(group=model.CustomerGroup())
 
        customer._groups.append(assignment)
 
        self.session.commit()
 
        self.assertEqual(assignment.ordinal, 1)
 
        assignment = model.CustomerGroupAssignment(group=model.CustomerGroup())
 
        customer._groups.append(assignment)
 
        self.session.commit()
 
        self.assertEqual(assignment.ordinal, 2)
 

	
 

	
 
class TestCustomerEmailAddress(DataTestCase):
 

	
 
    def test_pop(self):
 
        customer = model.Customer()
 
        customer.add_email_address('fred.home@mailinator.com')
 
        customer.add_email_address('fred.work@mailinator.com')
 
        self.session.add(customer)
 
        self.session.commit()
 
        self.assertEqual(self.session.query(model.Customer).count(), 1)
 
        self.assertEqual(self.session.query(model.CustomerEmailAddress).count(), 2)
 

	
 
        while customer.emails:
 
            customer.emails.pop()
 
        self.session.commit()
 
        self.assertEqual(self.session.query(model.Customer).count(), 1)
 
        self.assertEqual(self.session.query(model.CustomerEmailAddress).count(), 0)
 

	
 
        # changes weren't being recorded
 
        self.assertEqual(self.session.query(model.Change).count(), 0)
 

	
 
    def test_pop_with_changes(self):
 
        record_changes(self.session)
 

	
 
        customer = model.Customer()
 
        customer.add_email_address('fred.home@mailinator.com')
 
        customer.add_email_address('fred.work@mailinator.com')
 
        self.session.add(customer)
 
        self.session.commit()
 
        self.assertEqual(self.session.query(model.Customer).count(), 1)
 
        self.assertEqual(self.session.query(model.CustomerEmailAddress).count(), 2)
 

	
 
        while customer.emails:
 
            customer.emails.pop()
 
        self.session.commit()
 
        self.assertEqual(self.session.query(model.Customer).count(), 1)
 
        self.assertEqual(self.session.query(model.CustomerEmailAddress).count(), 0)
 

	
 
        # changes should have been recorded
 
        changes = self.session.query(model.Change)
 
        self.assertEqual(changes.count(), 3)
 

	
 
        customer_change = changes.filter_by(class_name='Customer').one()
 
        self.assertEqual(customer_change.uuid, customer.uuid)
 
        self.assertFalse(customer_change.deleted)
 

	
 
        email_changes = changes.filter_by(class_name='CustomerEmailAddress')
 
        self.assertEqual(email_changes.count(), 2)
 
        self.assertEqual([x.deleted for x in email_changes], [False, False])
0 comments (0 inline, 0 general)