# Path: oslo_db/sqlalchemy/update_match.py
# def update_on_match(
#     query,
#     specimen,
#     surrogate_key,
#     values=None,
#     attempts=3,
#     include_only=None,
#     process_query=None,
#     handle_failure=None
# ):
# def manufacture_persistent_object(
#         session, specimen, values=None, primary_key=None):
# def manufacture_entity_criteria(entity, include_only=None, exclude=None):
# def manufacture_criteria(mapped, values):
# def _sql_crit(expression, value):
# def update_returning_pk(query, values, surrogate_key):
# def _assert_single_row(rows_updated):
# def _pk_strategy_refetch(query, mapper, values, surrogate_key):
# def _pk_strategy_returning(query, mapper, values, surrogate_key):
# def _pk_strategy_mysql_last_insert_id(query, mapper, values, surrogate_key):
# def _update_stmt_from_query(mapper, query, values):
# class CantUpdateException(Exception):
# class NoRowsMatched(CantUpdateException):
# class MultiRowsMatched(CantUpdateException):
# 
# Path: oslo_db/tests/base.py
# class BaseTestCase(base.BaseTestCase):
#     def setUp(self):
# 
# Path: oslo_db/tests/sqlalchemy/base.py
# class Context(object):
# class DbTestCase(db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase):
# class MySQLOpportunisticTestCase(DbTestCase):
# class PostgreSQLOpportunisticTestCase(DbTestCase):
# class _DbTestCase(
#     db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase,
# ):
# class _MySQLOpportunisticTestCase(_DbTestCase):
# class _PostgreSQLOpportunisticTestCase(_DbTestCase):
#     def setUp(self):
#     def setUp(self):
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture


from sqlalchemy import orm
from sqlalchemy import schema
from sqlalchemy import sql
from sqlalchemy import types as sqltypes
from oslo_db.sqlalchemy import update_match
from oslo_db.tests import base as test_base
from oslo_db.tests.sqlalchemy import base as db_test_base
    def test_custom_handle_failure_cancel_raise(self):
        uuid = '136254d5-3869-408f-9da7-190e0072641a'

        class MyException(Exception):
            pass

        def handle_failure(query):
            # ensure the query is usable
            result = query.count()
            self.assertEqual(0, result)

            return True

        specimen = MyModel(
            id=2, y='y1', z='z3', uuid=uuid
        )

        result = self.session.query(MyModel).update_on_match(
            specimen, 'uuid', values={'x': 9, 'z': 'z3'},
            handle_failure=handle_failure
        )
        self.assertEqual(uuid, result.uuid)
        self.assertEqual(2, result.id)
        self.assertEqual('z3', result.z)
        self.assertEqual(9, result.x)
        self.assertIn(result, self.session)

    def test_update_specimen_on_none_successful(self):
        uuid = 'bdf3893c-ee3c-40a0-bc79-960adb6cd1d4'


# Path: oslo_db/sqlalchemy/update_match.py
# def update_on_match(
#     query,
#     specimen,
#     surrogate_key,
#     values=None,
#     attempts=3,
#     include_only=None,
#     process_query=None,
#     handle_failure=None
# ):
# def manufacture_persistent_object(
#         session, specimen, values=None, primary_key=None):
# def manufacture_entity_criteria(entity, include_only=None, exclude=None):
# def manufacture_criteria(mapped, values):
# def _sql_crit(expression, value):
# def update_returning_pk(query, values, surrogate_key):
# def _assert_single_row(rows_updated):
# def _pk_strategy_refetch(query, mapper, values, surrogate_key):
# def _pk_strategy_returning(query, mapper, values, surrogate_key):
# def _pk_strategy_mysql_last_insert_id(query, mapper, values, surrogate_key):
# def _update_stmt_from_query(mapper, query, values):
# class CantUpdateException(Exception):
# class NoRowsMatched(CantUpdateException):
# class MultiRowsMatched(CantUpdateException):
# 
# Path: oslo_db/tests/base.py
# class BaseTestCase(base.BaseTestCase):
#     def setUp(self):
# 
# Path: oslo_db/tests/sqlalchemy/base.py
# class Context(object):
# class DbTestCase(db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase):
# class MySQLOpportunisticTestCase(DbTestCase):
# class PostgreSQLOpportunisticTestCase(DbTestCase):
# class _DbTestCase(
#     db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase,
# ):
# class _MySQLOpportunisticTestCase(_DbTestCase):
# class _PostgreSQLOpportunisticTestCase(_DbTestCase):
#     def setUp(self):
#     def setUp(self):
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture


# Path: oslo_db/tests/sqlalchemy/test_update_match.py
from sqlalchemy import orm
from sqlalchemy import schema
from sqlalchemy import sql
from sqlalchemy import types as sqltypes
from oslo_db.sqlalchemy import update_match
from oslo_db.tests import base as test_base
from oslo_db.tests.sqlalchemy import base as db_test_base

    def test_custom_handle_failure_cancel_raise(self):
        uuid = '136254d5-3869-408f-9da7-190e0072641a'

        class MyException(Exception):
            pass

        def handle_failure(query):
            # ensure the query is usable
            result = query.count()
            self.assertEqual(0, result)

            return True

        specimen = MyModel(
            id=2, y='y1', z='z3', uuid=uuid
        )

        result = self.session.query(MyModel).update_on_match(
            specimen, 'uuid', values={'x': 9, 'z': 'z3'},
            handle_failure=handle_failure
        )
        self.assertEqual(uuid, result.uuid)
        self.assertEqual(2, result.id)
        self.assertEqual('z3', result.z)
        self.assertEqual(9, result.x)
        self.assertIn(result, self.session)

    def test_update_specimen_on_none_successful(self):
        uuid = 'bdf3893c-ee3c-40a0-bc79-960adb6cd1d4'


        specimen = MyModel(