def test_startswith_literal_mysql(self):
     self.assert_compile(
         column('x').startswith(literal_column('y')),
         "x LIKE concat(y, '%%')",
         checkparams={},
         dialect=mysql.dialect()
     )

 def test_mariadb_normalized_version(self):
     for expected, raw_version, version, is_mariadb in [
         ((10, 2, 7), "10.2.7-MariaDB", (10, 2, 7, "MariaDB"), True),
         (
             (10, 2, 7),
             "5.6.15.10.2.7-MariaDB",
             (5, 6, 15, 10, 2, 7, "MariaDB"),
             True,
         ),
         ((10, 2, 10), "10.2.10-MariaDB", (10, 2, 10, "MariaDB"), True),
         ((5, 7, 20), "5.7.20", (5, 7, 20), False),
         ((5, 6, 15), "5.6.15", (5, 6, 15), False),
         (
             (10, 2, 6),
             "10.2.6.MariaDB.10.2.6+maria~stretch-log",
             (10, 2, 6, "MariaDB", 10, 2, "6+maria~stretch", "log"),
             True,
         ),
         (
             (10, 1, 9),
             "10.1.9-MariaDBV1.0R050D002-20170809-1522",
             (10, 1, 9, "MariaDB", "V1", "0R050D002", 20170809, 1522),
             True,
         ),
     ]:
         dialect = mysql.dialect()
         eq_(dialect._parse_server_version(raw_version), version)
         dialect.server_version_info = version
         eq_(dialect._mariadb_normalized_version_info, expected)
         assert dialect._is_mariadb is is_mariadb

 def test_contains_text_concat(self):
     self.assert_compile(
         column('x').contains(text('y')),
         "x LIKE concat(concat('%%', y), '%%')",
         checkparams={},
         dialect=mysql.dialect()
     )

    def test_update_bound_ordering(self):
        """test that bound parameters between the UPDATE and FROM clauses
        order correctly in different SQL compilation scenarios.

        """
        table1 = self.tables.mytable
        table2 = self.tables.myothertable
        sel = select([table2]).where(table2.c.otherid == 5).alias()
        upd = table1.update().where(table1.c.name == sel.c.othername).values(name="foo")

        dialect = default.DefaultDialect()
        dialect.positional = True
        self.assert_compile(
            upd,
            "UPDATE mytable SET name=:name FROM (SELECT "
            "myothertable.otherid AS otherid, "
            "myothertable.othername AS othername "
            "FROM myothertable "
            "WHERE myothertable.otherid = :otherid_1) AS anon_1 "
            "WHERE mytable.name = anon_1.othername",
            checkpositional=("foo", 5),
            dialect=dialect,
        )

        self.assert_compile(
            upd,
            "UPDATE mytable, (SELECT myothertable.otherid AS otherid, "
            "myothertable.othername AS othername "
            "FROM myothertable "
            "WHERE myothertable.otherid = %s) AS anon_1 SET mytable.name=%s "
            "WHERE mytable.name = anon_1.othername",
            checkpositional=(5, "foo"),
            dialect=mysql.dialect(),
        )

 def left_choice(self, evt):
     """Sets the left column and updates the available columns in the right choice"""
     choice = self.panel.choiceLeft.GetCurrentSelection()
     if choice == 0:
         self.workingJoin['tableValue'] = None
         self.workingJoin['joiningValue'] = None
         self.panel.choiceRight.Clear()
         self.rightSelections = ['Choose a compatible column...']
         self.panel.choiceRight.Append(self.rightSelections[0])
         for i in self.rightSelectionsTypes:
             self.rightSelections.append((i[0], i[2]))
             self.panel.choiceRight.Append(i[0])
         self.panel.rightChoice.SetSelection(0)
         self.panel.btnOk.Enable(False)
     else:
         #set values
         self.workingJoin['tableValue'] = self.leftColumnNames[choice]
         self.workingJoin['joiningValue'] = None
         #load up compatible columns into right hand side
         choiceType = self.leftSelectionsTypes[choice]
         self.rightSelections = ['Choose a compatible column...']
         for k in self.rightSelectionsTypes[1:]:
             #run through columns and check if types are the same, some columns are MySQL types
             #and non-standard SQL types, thus the dialect type checking.
             if str(k[1].compile(dialect=mysql.dialect())) == str(choiceType.compile(dialect=mysql.dialect())):
                 self.rightSelections.append((k[0], k[2]))
         self.panel.choiceRight.Clear()
         self.panel.choiceRight.Append(self.rightSelections[0])
         for i in self.rightSelections[1:]:
             self.panel.choiceRight.Append(i[0])
         self.panel.choiceRight.SetSelection(0)
         self.panel.btnOk.Enable(False)

 def test_not_endswith_mysql(self):
     self.assert_compile(
         ~column('x').endswith('y'),
         "x NOT LIKE concat('%%', %s)",
         checkparams={'x_1': 'y'},
         dialect=mysql.dialect()
     )

 def test_not_startswith_concat(self):
     self.assert_compile(
         ~column('x').startswith('y'),
         "x NOT LIKE concat(%s, '%%')",
         checkparams={'x_1': 'y'},
         dialect=mysql.dialect()
     )

 def test_endswith_text_mysql(self):
     self.assert_compile(
         column('x').endswith(text('y')),
         "x LIKE concat('%%', y)",
         checkparams={},
         dialect=mysql.dialect()
     )

 def test_not_contains_concat(self):
     self.assert_compile(
         ~column('x').contains('y'),
         "x NOT LIKE concat(concat('%%', %s), '%%')",
         checkparams={'x_1': 'y'},
         dialect=mysql.dialect()
     )

    def test_repr_plain_sqla_type(self):
        type_ = Integer()
        autogen_context = {
            "opts": {"sqlalchemy_module_prefix": "sa.", "alembic_module_prefix": "op."},
            "dialect": mysql.dialect(),
        }

        eq_ignore_whitespace(autogenerate.render._repr_type(type_, autogen_context), "sa.Integer()")

 def setup_class(cls):
     cls.autogen_context = {
         'opts':{
             'sqlalchemy_module_prefix' : 'sa.',
             'alembic_module_prefix' : 'op.',
         },
         'dialect':mysql.dialect()
     }

 def setup_class(cls):
     cls.autogen_context = {
         "opts": {"sqlalchemy_module_prefix": "sa.", "alembic_module_prefix": "op."},
         "dialect": mysql.dialect(),
     }
     cls.pg_autogen_context = {
         "opts": {"sqlalchemy_module_prefix": "sa.", "alembic_module_prefix": "op."},
         "dialect": postgresql.dialect(),
     }

    def test_generic_now(self):
        assert isinstance(func.now().type, sqltypes.DateTime)

        for ret, dialect in [
            ('CURRENT_TIMESTAMP', sqlite.dialect()),
            ('now()', postgresql.dialect()),
            ('now()', mysql.dialect()),
            ('CURRENT_TIMESTAMP', oracle.dialect())
        ]:
            self.assert_compile(func.now(), ret, dialect=dialect)

    def test_generic_random(self):
        assert func.random().type == sqltypes.NULLTYPE
        assert isinstance(func.random(type_=Integer).type, Integer)

        for ret, dialect in [
            ('random()', sqlite.dialect()),
            ('random()', postgresql.dialect()),
            ('rand()', mysql.dialect()),
            ('random()', oracle.dialect())
        ]:
            self.assert_compile(func.random(), ret, dialect=dialect)

def clean_allocations(instance_uuid, hypervisor, connection=None,
          debug=False, dry_run=True):
    """clean up multiple allocations in nova_api db for instance

    :param str instance_uuid: instance uuid
    :param str hypervisor: hypervisor running the instance
    :param str connection: specify db connection string
    :param bool debug (False): show sql calls
    :param bool dry_run (True): commit changes to the db
    """
    connection = get_connection(connection)
    engine = connect(connection, debug)
    db_conn = engine.connect()

    alloc_tb = Table('allocations', metadata,
                     autoload=True, autoload_with=engine)
    res_tb = Table('resource_providers', metadata,
                   autoload=True, autoload_with=engine)

    query = select(
        [alloc_tb.c.id,
         alloc_tb.c.consumer_id,
         alloc_tb.c.resource_provider_id,
          res_tb.c.name]).select_from(
              alloc_tb.join(
                  res_tb,
                  alloc_tb.c.resource_provider_id == res_tb.c.id)).where(
                      and_(res_tb.c.name != hypervisor,
                           alloc_tb.c.consumer_id == instance_uuid))

    results = db_conn.execute(query).fetchall()

    if not results:
        print("multiple allocation records not found for %s\n" % instance_uuid)
        return

    ids = [r[0] for r in results]
    if dry_run:
        print("extra allocations found:")
        print_table(results)
        stmt = alloc_tb.delete().where(alloc_tb.c.id.in_(ids))
        print('would execute statement:')
        print(str(stmt.compile(
              dialect=mysql.dialect(),
              compile_kwargs={"literal_binds": True})))
        print("run with --no-dry-run to commit changes to the db")
    else:
        print("deleting extra allocations:")
        print_table(results)
        stmt = alloc_tb.delete().where(alloc_tb.c.id.in_(ids))
        db_conn.execute(stmt)
        print("remaining allocations:")
        show_allocations(instance_uuid)

 def test_render_table_mysql(self):
     users, addresses = self.tables.users, self.tables.addresses
     self.assert_compile(
         users.update().\
             values(name='newname').\
             where(users.c.id==addresses.c.user_id).\
             where(addresses.c.email_address=='e1'),
         "UPDATE users, addresses SET users.name=%s "
         "WHERE users.id = addresses.user_id AND "
         "addresses.email_address = %s",
         checkparams={u'email_address_1': 'e1', 'name': 'newname'},
         dialect=mysql.dialect()
     )

    def test_prefix_with(self):
        table1 = self.tables.mytable

        stmt = table1.delete().\
            prefix_with('A', 'B', dialect='mysql').\
            prefix_with('C', 'D')

        self.assert_compile(stmt,
                            'DELETE C D FROM mytable')

        self.assert_compile(stmt,
                            'DELETE A B C D FROM mytable',
                            dialect=mysql.dialect())

    def test_repr_plain_sqla_type(self):
        type_ = Integer()
        autogen_context = {
            'opts': {
                'sqlalchemy_module_prefix': 'sa.',
                'alembic_module_prefix': 'op.',
            },
            'dialect': mysql.dialect()
        }

        eq_ignore_whitespace(
            autogenerate.render._repr_type(type_, autogen_context),
            "sa.Integer()"
        )

    def test_repr_dialect_type(self):
        from sqlalchemy.dialects.mysql import VARCHAR

        type_ = VARCHAR(20, charset="utf8", national=True)
        autogen_context = {
            "opts": {"sqlalchemy_module_prefix": "sa.", "alembic_module_prefix": "op.", "user_module_prefix": None},
            "imports": set(),
            "dialect": mysql.dialect(),
        }
        eq_ignore_whitespace(
            autogenerate.render._repr_type(type_, autogen_context),
            "mysql.VARCHAR(charset='utf8', national=True, length=20)",
        )
        eq_(autogen_context["imports"], set(["from sqlalchemy.dialects import mysql"]))

    def test_repr_user_type_user_prefix_present(self):
        from sqlalchemy.types import UserDefinedType

        class MyType(UserDefinedType):
            def get_col_spec(self):
                return "MYTYPE"

        type_ = MyType()
        autogen_context = {
            "opts": {"sqlalchemy_module_prefix": "sa.", "alembic_module_prefix": "op.", "user_module_prefix": "user."},
            "dialect": mysql.dialect(),
        }

        eq_ignore_whitespace(autogenerate.render._repr_type(type_, autogen_context), "user.MyType()")