# 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_instance_criteria_tuples(self):
        specimen = MyModel(
            y='y1', z=('z1', 'z2'),
        )
        self.assertRegex(
            str(update_match.manufacture_entity_criteria(specimen).compile()),
            r"my_table.y = :y_1 AND my_table.z IN \(.+?\)",
        )

    def test_instance_criteria_tuples_wnone(self):
        specimen = MyModel(
            y='y1', z=('z1', 'z2', None),
        )
        self.assertRegex(
            str(update_match.manufacture_entity_criteria(specimen).compile()),
            r"my_table.y = :y_1 AND \(my_table.z IS NULL OR "
            r"my_table.z IN \(.+?\)\)",
        )

    def test_instance_criteria_none_list(self):
        specimen = MyModel(
            y='y1', z=[None],
        )
        self.assertEqual(
            "my_table.y = :y_1 AND my_table.z IS NULL",
            str(update_match.manufacture_entity_criteria(specimen).compile())
        )



# 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_instance_criteria_tuples(self):
        specimen = MyModel(
            y='y1', z=('z1', 'z2'),
        )
        self.assertRegex(
            str(update_match.manufacture_entity_criteria(specimen).compile()),
            r"my_table.y = :y_1 AND my_table.z IN \(.+?\)",
        )

    def test_instance_criteria_tuples_wnone(self):
        specimen = MyModel(
            y='y1', z=('z1', 'z2', None),
        )
        self.assertRegex(
            str(update_match.manufacture_entity_criteria(specimen).compile()),
            r"my_table.y = :y_1 AND \(my_table.z IS NULL OR "
            r"my_table.z IN \(.+?\)\)",
        )

    def test_instance_criteria_none_list(self):
        specimen = MyModel(
            y='y1', z=[None],
        )
        self.assertEqual(
            "my_table.y = :y_1 AND my_table.z IS NULL",
            str(update_match.manufacture_entity_criteria(specimen).compile())
        )



class UpdateMatchTest(db_test_base._DbTestCase):