# 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
            {
                'uuid': '136254d5-3869-408f-9da7-190e0072641a',
                'x': 9, 'y': 'y1', 'z': 'z3'
            }
        )

    def test_update_returning_wrong_uuid(self):
        exc = self.assertRaises(
            update_match.NoRowsMatched,
            self.session.query(MyModel).
            filter_by(y='y1', z='z2').update_returning_pk,
            {'x': 9, 'z': 'z3'},
            ('uuid', '23cb9224-9f8e-40fe-bd3c-e7577b7af37d')
        )

        self.assertEqual("No rows matched the UPDATE", exc.args[0])

    def test_update_returning_no_rows(self):
        exc = self.assertRaises(
            update_match.NoRowsMatched,
            self.session.query(MyModel).
            filter_by(y='y1', z='z3').update_returning_pk,
            {'x': 9, 'z': 'z3'},
            ('uuid', '136254d5-3869-408f-9da7-190e0072641a')
        )

        self.assertEqual("No rows matched the UPDATE", exc.args[0])

    def test_update_multiple_rows(self):
        exc = self.assertRaises(

# 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

            {
                'uuid': '136254d5-3869-408f-9da7-190e0072641a',
                'x': 9, 'y': 'y1', 'z': 'z3'
            }
        )

    def test_update_returning_wrong_uuid(self):
        exc = self.assertRaises(
            update_match.NoRowsMatched,
            self.session.query(MyModel).
            filter_by(y='y1', z='z2').update_returning_pk,
            {'x': 9, 'z': 'z3'},
            ('uuid', '23cb9224-9f8e-40fe-bd3c-e7577b7af37d')
        )

        self.assertEqual("No rows matched the UPDATE", exc.args[0])

    def test_update_returning_no_rows(self):
        exc = self.assertRaises(
            update_match.NoRowsMatched,
            self.session.query(MyModel).
            filter_by(y='y1', z='z3').update_returning_pk,
            {'x': 9, 'z': 'z3'},
            ('uuid', '136254d5-3869-408f-9da7-190e0072641a')
        )

        self.assertEqual("No rows matched the UPDATE", exc.args[0])

    def test_update_multiple_rows(self):
        exc = self.assertRaises(

            update_match.MultiRowsMatched,