rattail-project/rattail Changeset

Changeset - a2b13de31c61

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Lance Edgar (lance) - 11 years ago 2014-01-26 01:19:18
lance@edbob.org

Ignore SQLAlchemy's warnings about SQLite decimal storage when running tests.

2 files changed with 99 insertions and 109 deletions:

tests/db/__init__.py

tests/db/sync/test_init.py

109

0 comments (0 inline, 0 general)

tests/db/__init__.py

➞

Show inline comments

 import unittest
 import warnings
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.exc import SAWarning
 from rattail.db.model import Base
@@ @@ -10,6 +12,15 @@ from rattail.db.model import Base @@
 __all__ = ['DataTestCase']
 warnings.filterwarnings(
     'ignore',
     r"^Dialect sqlite\+pysqlite does \*not\* support Decimal objects natively\, "
     "and SQLAlchemy must convert from floating point - rounding errors and other "
     "issues may occur\. Please consider storing Decimal numbers as strings or "
     "integers on this platform for lossless storage\.$",
     SAWarning, r'^sqlalchemy\.sql\.type_api$')
 class DataTestCase(unittest.TestCase):
     def setUp(self):

tests/db/sync/test_init.py

➞

Show inline comments

 import warnings
 from unittest import TestCase
 from mock import patch, call, Mock, DEFAULT
-from sqlalchemy.exc import OperationalError, SAWarning
 from sqlalchemy.exc import OperationalError
 from . import SyncTestCase
 from rattail.db import sync
@@ @@ -157,76 +156,66 @@ class SynchronizerTests(SyncTestCase): @@
     def test_merge_Product(self):
         synchronizer = sync.Synchronizer(self.local_engine, self.remote_engines)
         with warnings.catch_warnings():
             warnings.filterwarnings(
                 'ignore',
                 r'^Dialect sqlite\+pysqlite does \*not\* support Decimal '
                 'objects natively\, and SQLAlchemy must convert from floating '
                 'point - rounding errors and other issues may occur\. Please '
                 'consider storing Decimal numbers as strings or integers on '
                 'this platform for lossless storage\.$',
                 SAWarning, r'^sqlalchemy\.types$')
             # no prices
             local_session = self.Session(bind=self.local_engine)
             remote_session = self.Session(bind=self.remote_engines['one'])
             source_product = model.Product()
             local_session.add(source_product)
             local_session.flush()
             self.assertTrue(source_product.regular_price_uuid is None)
             self.assertTrue(source_product.regular_price is None)
             self.assertTrue(source_product.current_price_uuid is None)
             self.assertTrue(source_product.current_price is None)
             target_product = synchronizer.merge_Product(remote_session, source_product)
             self.assertFalse(target_product is None)
             self.assertFalse(source_product is target_product)
             self.assertEqual(source_product.uuid, target_product.uuid)
             self.assertTrue(target_product.regular_price_uuid is None)
             self.assertTrue(target_product.regular_price is None)
             self.assertTrue(target_product.current_price_uuid is None)
             self.assertTrue(target_product.current_price is None)
             local_session.rollback()
             local_session.close()
             remote_session.rollback()
             remote_session.close()
             # regular price
             local_session = self.Session(bind=self.local_engine)
             remote_session = self.Session(bind=self.remote_engines['one'])
             source_product = model.Product()
             regular_price = model.ProductPrice()
             source_product.prices.append(regular_price)
             source_product.regular_price = regular_price
             local_session.add(source_product)
             local_session.flush()
             self.assertFalse(source_product.regular_price_uuid is None)
             self.assertFalse(source_product.regular_price is None)
             target_product = synchronizer.merge_Product(remote_session, source_product)
             self.assertEqual(target_product.regular_price_uuid, source_product.regular_price_uuid)
             self.assertFalse(target_product.regular_price is None)
             local_session.rollback()
             local_session.close()
             remote_session.rollback()
             remote_session.close()
             # current price
             local_session = self.Session(bind=self.local_engine)
             remote_session = self.Session(bind=self.remote_engines['one'])
             source_product = model.Product()
             current_price = model.ProductPrice()
             source_product.prices.append(current_price)
             source_product.current_price = current_price
             local_session.add(source_product)
             local_session.flush()
             self.assertFalse(source_product.current_price_uuid is None)
             self.assertFalse(source_product.current_price is None)
             target_product = synchronizer.merge_Product(remote_session, source_product)
             self.assertEqual(target_product.current_price_uuid, source_product.current_price_uuid)
             self.assertFalse(target_product.current_price is None)
             local_session.rollback()
             local_session.close()
             remote_session.rollback()
             remote_session.close()
         # no prices
         local_session = self.Session(bind=self.local_engine)
         remote_session = self.Session(bind=self.remote_engines['one'])
         source_product = model.Product()
         local_session.add(source_product)
         local_session.flush()
         self.assertTrue(source_product.regular_price_uuid is None)
         self.assertTrue(source_product.regular_price is None)
         self.assertTrue(source_product.current_price_uuid is None)
         self.assertTrue(source_product.current_price is None)
         target_product = synchronizer.merge_Product(remote_session, source_product)
         self.assertFalse(target_product is None)
         self.assertFalse(source_product is target_product)
         self.assertEqual(source_product.uuid, target_product.uuid)
         self.assertTrue(target_product.regular_price_uuid is None)
         self.assertTrue(target_product.regular_price is None)
         self.assertTrue(target_product.current_price_uuid is None)
         self.assertTrue(target_product.current_price is None)
         local_session.rollback()
         local_session.close()
         remote_session.rollback()
         remote_session.close()
         # regular price
         local_session = self.Session(bind=self.local_engine)
         remote_session = self.Session(bind=self.remote_engines['one'])
         source_product = model.Product()
         regular_price = model.ProductPrice()
         source_product.prices.append(regular_price)
         source_product.regular_price = regular_price
         local_session.add(source_product)
         local_session.flush()
         self.assertFalse(source_product.regular_price_uuid is None)
         self.assertFalse(source_product.regular_price is None)
         target_product = synchronizer.merge_Product(remote_session, source_product)
         self.assertEqual(target_product.regular_price_uuid, source_product.regular_price_uuid)
         self.assertFalse(target_product.regular_price is None)
         local_session.rollback()
         local_session.close()
         remote_session.rollback()
         remote_session.close()
         # current price
         local_session = self.Session(bind=self.local_engine)
         remote_session = self.Session(bind=self.remote_engines['one'])
         source_product = model.Product()
         current_price = model.ProductPrice()
         source_product.prices.append(current_price)
         source_product.current_price = current_price
         local_session.add(source_product)
         local_session.flush()
         self.assertFalse(source_product.current_price_uuid is None)
         self.assertFalse(source_product.current_price is None)
         target_product = synchronizer.merge_Product(remote_session, source_product)
         self.assertEqual(target_product.current_price_uuid, source_product.current_price_uuid)
         self.assertFalse(target_product.current_price is None)
         local_session.rollback()
         local_session.close()
         remote_session.rollback()
         remote_session.close()
     def test_delete_instance(self):
@@ @@ -248,43 +237,33 @@ class SynchronizerTests(SyncTestCase): @@
     def test_delete_Department(self):
         synchronizer = sync.Synchronizer(self.local_engine, self.remote_engines)
         with warnings.catch_warnings():
             warnings.filterwarnings(
                 'ignore',
                 r'^Dialect sqlite\+pysqlite does \*not\* support Decimal '
                 'objects natively\, and SQLAlchemy must convert from floating '
                 'point - rounding errors and other issues may occur\. Please '
                 'consider storing Decimal numbers as strings or integers on '
                 'this platform for lossless storage\.$',
                 SAWarning, r'^sqlalchemy\.types$')
             session = self.Session(bind=self.local_engine)
             department = model.Department()
             department.subdepartments.append(model.Subdepartment())
             session.add(department)
             session.flush()
             self.assertEqual(session.query(model.Subdepartment).count(), 1)
             subdepartment = session.query(model.Subdepartment).one()
             self.assertEqual(subdepartment.department_uuid, department.uuid)
             synchronizer.delete_Department(session, department)
             self.assertEqual(session.query(model.Subdepartment).count(), 1)
             subdepartment = session.query(model.Subdepartment).one()
             self.assertTrue(subdepartment.department_uuid is None)
             session.rollback()
             session.close()
             session = self.Session(bind=self.local_engine)
             department = model.Department()
             product = model.Product(department=department)
             session.add(product)
             session.flush()
             product = session.query(model.Product).one()
             self.assertEqual(product.department_uuid, department.uuid)
             synchronizer.delete_Department(session, department)
             product = session.query(model.Product).one()
             self.assertTrue(product.department_uuid is None)
             session.rollback()
             session.close()
         session = self.Session(bind=self.local_engine)
         department = model.Department()
         department.subdepartments.append(model.Subdepartment())
         session.add(department)
         session.flush()
         self.assertEqual(session.query(model.Subdepartment).count(), 1)
         subdepartment = session.query(model.Subdepartment).one()
         self.assertEqual(subdepartment.department_uuid, department.uuid)
         synchronizer.delete_Department(session, department)
         self.assertEqual(session.query(model.Subdepartment).count(), 1)
         subdepartment = session.query(model.Subdepartment).one()
         self.assertTrue(subdepartment.department_uuid is None)
         session.rollback()
         session.close()
         session = self.Session(bind=self.local_engine)
         department = model.Department()
         product = model.Product(department=department)
         session.add(product)
         session.flush()
         product = session.query(model.Product).one()
         self.assertEqual(product.department_uuid, department.uuid)
         synchronizer.delete_Department(session, department)
         product = session.query(model.Product).one()
         self.assertTrue(product.department_uuid is None)
         session.rollback()
         session.close()
     def test_delete_Subdepartment(self):
         synchronizer = sync.Synchronizer(self.local_engine, self.remote_engines)

0 comments (0 inline, 0 general)