rattail-project/rattail Changeset

Changeset - d135e1fd1167

Parent rev.

Child rev.

[Not reviewed]

0 3 0

Lance Edgar (lance) - 10 years ago 2014-06-05 21:08:35
lance@edbob.org

Add support for older SQLAlchemy (0.6.3 specifically).

It's possible this still misses some things, but at least tests pass...

3 files changed with 50 insertions and 30 deletions:

rattail/db/changes.py

tests/db/test_changes.py

tests/db/test_model.py

0 comments (0 inline, 0 general)

rattail/db/changes.py

➞

Show inline comments

 #!/usr/bin/env python
 # -*- coding: utf-8  -*-
 # -*- coding: utf-8 -*-
 ################################################################################
+#
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2012 Lance Edgar
+#  Copyright © 2010-2014 Lance Edgar
+#
 #  This file is part of Rattail.
+#
 #  Rattail is free software: you can redistribute it and/or modify it under the
 #  terms of the GNU Affero General Public License as published by the Free
 #  Software Foundation, either version 3 of the License, or (at your option)
@@ @@ -23,14 +22,15 @@ @@
 ################################################################################
 """
 Data Changes
 """
 from sqlalchemy.event import listen
 from sqlalchemy.orm import object_mapper, RelationshipProperty
 from sqlalchemy.orm.interfaces import SessionExtension
 from sqlalchemy.orm.session import Session
 from . import model
 from ..core import get_uuid
 __all__ = ['record_changes']
@@ @@ -41,22 +41,31 @@ log = logging.getLogger(__name__) @@
 def record_changes(session, ignore_role_changes=True):
     """
     Record all changes which occur within a session.
     :param session: A ``sqlalchemy.orm.sessionmaker`` class, or an instance
        thereof.
     :param session: A :class:`sqlalchemy:sqlalchemy.orm.session.Session` class,
        or an instance thereof.
     :param ignore_role_changes: Whether changes involving roles and role
        membership should be ignored.  This defaults to ``True``, which means
        each database will be responsible for maintaining its own role (and by
        extension, permissions) data.
     """
     listen(session, 'before_flush', ChangeRecorder(ignore_role_changes))
     recorder = ChangeRecorder(ignore_role_changes)
     try:
         from sqlalchemy.event import listen
     except ImportError: # pragma: no cover
         extension = ChangeRecorderExtension(recorder)
         if isinstance(session, Session):
             session.extensions.append(extension)
         else:
             session.configure(extension=extension)
     else:
         listen(session, u'before_flush', recorder)
 class ChangeRecorder(object):
     """
     Listener for session ``before_flush`` events.
@@ @@ -193,6 +202,22 @@ class ChangeRecorder(object): @@
                     self.ensure_uuid(foreign_instance)
                     instance.uuid = foreign_instance.uuid
                     return
         instance.uuid = get_uuid()
         log.error("ChangeRecorder.ensure_uuid: unexpected scenario; generated new UUID for instance: {0}".format(repr(instance)))
 class ChangeRecorderExtension(SessionExtension): # pragma: no cover
     """
     Session extension for recording changes.
     .. note::
        This is only used when the installed SQLAlchemy version is old enough
        not to support the new event interfaces.
     """
     def __init__(self, recorder):
         self.recorder = recorder
     def before_flush(self, session, flush_context, instances):
         self.recorder(session, flush_context, instances)

tests/db/test_changes.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from unittest import TestCase
 from mock import patch, DEFAULT, Mock, MagicMock, call
 from . import DataTestCase
 from rattail.db import changes
 from rattail.db import model
 from sqlalchemy.orm import RelationshipProperty
 class TestChanges(TestCase):
     @patch.multiple('rattail.db.changes', listen=DEFAULT, ChangeRecorder=DEFAULT)
     def test_record_changes(self, listen, ChangeRecorder):
         session = Mock()
         ChangeRecorder.return_value = 'whatever'
         changes.record_changes(session)
         ChangeRecorder.assert_called_once_with(True)
         listen.assert_called_once_with(session, 'before_flush', 'whatever')
         ChangeRecorder.reset_mock()
         listen.reset_mock()
         changes.record_changes(session, ignore_role_changes=False)
         ChangeRecorder.assert_called_once_with(False)
         listen.assert_called_once_with(session, 'before_flush', 'whatever')
 class TestChangeRecorder(TestCase):
     def test_init(self):
         recorder = changes.ChangeRecorder()
         self.assertTrue(recorder.ignore_role_changes)
         recorder = changes.ChangeRecorder(False)

tests/db/test_model.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from unittest import TestCase
 from . import DataTestCase
 from mock import patch, DEFAULT, Mock, MagicMock
 from sqlalchemy import String, Boolean, Numeric
@@ @@ -8,12 +9,23 @@ from sqlalchemy.exc import IntegrityError @@
 from rattail.db import model
 from rattail.db.types import GPCType
 from rattail.db.changes import record_changes
 class SAErrorHelper(object):
     def integrity_or_flush_error(self):
         try:
             from sqlalchemy.exc import FlushError
         except ImportError:
             return IntegrityError
         else:
             return (IntegrityError, FlushError)
 class TestPerson(DataTestCase):
     def test_default_display_name_is_generated_from_first_and_last_name_if_both_provided(self):
         person = model.Person(first_name='Fred', last_name='Flintstone')
         self.assertTrue(person.display_name is None)
         self.session.add(person)
@@ @@ -116,22 +128,22 @@ class TestCustomer(DataTestCase): @@
         self.assertEqual(self.session.query(model.CustomerGroupAssignment).count(), 1)
         self.session.delete(customer)
         self.session.commit()
         self.assertEqual(self.session.query(model.CustomerGroupAssignment).count(), 0)
 class TestCustomerPerson(DataTestCase):
+class TestCustomerPerson(DataTestCase, SAErrorHelper):
     def test_repr(self):
         assoc = model.CustomerPerson(uuid='whatever')
         self.assertEqual(repr(assoc), "CustomerPerson(uuid='whatever')")
     def test_customer_required(self):
         assoc = model.CustomerPerson(person=model.Person())
         self.session.add(assoc)
-        self.assertRaises(IntegrityError, self.session.commit)
+        self.assertRaises(self.integrity_or_flush_error(), self.session.commit)
         self.session.rollback()
         self.assertEqual(self.session.query(model.CustomerPerson).count(), 0)
         assoc.customer = model.Customer()
         self.session.add(assoc)
         self.session.commit()
         self.assertEqual(self.session.query(model.CustomerPerson).count(), 1)
@@ @@ -157,22 +169,22 @@ class TestCustomerPerson(DataTestCase): @@
         assoc = model.CustomerPerson(person=model.Person())
         customer._people.append(assoc)
         self.session.commit()
         self.assertEqual(assoc.ordinal, 2)
 class TestCustomerGroupAssignment(DataTestCase):
+class TestCustomerGroupAssignment(DataTestCase, SAErrorHelper):
     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.assertRaises(self.integrity_or_flush_error(), 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)

0 comments (0 inline, 0 general)