def test_load_dialect_impl_other(self):
        """Assert with dialects other than PostgreSQL, a CHAR type is used."""
        guid = models.GUID()
        dialect = sqlite.dialect()

        result = guid.load_dialect_impl(dialect)

        self.assertTrue(isinstance(result, CHAR))

    def test_process_bind_param_none(self):
        """Assert UUIDs with other dialects are hex-encoded strings of length 32."""
        guid = models.GUID()
        dialect = sqlite.dialect()

        result = guid.process_bind_param(None, dialect)

        self.assertTrue(result is None)

    def log_query(self, q):
        """
        Log Query
        :param q: Query
        """
        query_str =str(q.statement.compile(dialect=sqlite.dialect()))

        log.debug(query_str)

def pprint(query_or_sql, engine):
    if isinstance(query_or_sql, Query):
        sql = query_or_sql.statement.compile(dialect=sqlite.dialect())
    else:
        sql = query_or_sql
    result_proxy = engine.execute(sql)
    p_table = from_db_cursor(result_proxy.cursor)
    print(p_table)

    def test_process_result_short_string(self):
        """Assert when the result value is a short string, a native UUID is returned."""
        guid = models.GUID()
        uuid = uuid4()

        result = guid.process_result_value(str(uuid).replace("-", ""), sqlite.dialect())

        self.assertTrue(isinstance(result, UUID))
        self.assertEqual(uuid, result)

    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_process_bind_param_str_other(self):
        """Assert UUIDs with other dialects are hex-encoded strings of length 32."""
        guid = models.GUID()
        uuid = uuid4()
        dialect = sqlite.dialect()

        result = guid.process_bind_param(str(uuid), dialect)

        self.assertEqual(32, len(result))
        self.assertEqual(str(uuid).replace("-", ""), result)

    def log_query(q):
        """Log Query
        :param q: Query
        """
        statement = q.statement.compile(dialect=sqlite.dialect())
        if statement.params:
            query_str = '{0}\n{1}'.format(statement, statement.params)
        else:
            query_str = str(statement)

        log.debug(query_str)

    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 test_nextval_unsupported(self):
        """test next_value() used on non-sequence platform
        raises NotImplementedError."""

        s = Sequence("my_seq")
        d = sqlite.dialect()
        assert_raises_message(
            NotImplementedError,
            "Dialect 'sqlite' does not support sequence increments.",
            s.next_value().compile,
            dialect=d
        )

def test_sqlite_in_with_multiple_columns():
    vals = (
        (1, '1'),
        (2, '2'),
    )
    cols = table.c.id, table.c.a
    select_in = sa.select(cols).where(sa.tuple_(*cols).in_(vals))
    select_and = sa.select(cols).where(sa.or_(
        sa.and_(col == value for col, value in zip(cols, vs))
        for vs in vals
    ))
    compiled_in = select_in.compile(dialect=sqlite.dialect())
    compiled_and = select_and.compile(dialect=sqlite.dialect())
    assert str(compiled_in) == str(compiled_and)
    update_in = table.update().where(sa.tuple_(*cols).in_(vals))
    update_and = table.update().where(sa.or_(
        sa.and_(col == value for col, value in zip(cols, vs))
        for vs in vals
    ))
    compiled_in = update_in.compile(dialect=sqlite.dialect())
    compiled_and = update_and.compile(dialect=sqlite.dialect())
    assert str(compiled_in) == str(compiled_and)

    def test_sql_expression_pk_autoinc_lastinserted(self):
        # test that postfetch isn't invoked for a SQL expression
        # in a primary key column.  the DB either needs to support a lastrowid
        # that can return it, or RETURNING.  [ticket:3133]
        metadata = MetaData()
        table = Table(
            "sometable",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("data", String),
        )

        stmt = table.insert().return_defaults().values(id=func.foobar())
        compiled = stmt.compile(dialect=sqlite.dialect(), column_keys=["data"])
        eq_(compiled.postfetch, [])
        eq_(compiled.returning, [])

        self.assert_compile(
            stmt,
            "INSERT INTO sometable (id, data) VALUES " "(foobar(), ?)",
            checkparams={"data": "foo"},
            params={"data": "foo"},
            dialect=sqlite.dialect(),
        )

    def __init__(self, uri):
        parsed = urlparse.urlparse(uri)

        args = []
        if parsed.scheme in ("sqlite", "sqlite3"):
            args.append("sqlite3")
            if not parsed.path:
                args.append(":memory:")
            else:
                args.append(parsed.path)

            from sqlalchemy.dialects.sqlite import dialect
            self.dialect = dialect()

        self.pool = adbapi.ConnectionPool(*args, cp_min=1, cp_max=1)

    def test_process_result_value_none(self):
        """Assert when the result value is None, None is returned."""
        guid = models.GUID()

        self.assertTrue(guid.process_result_value(None, sqlite.dialect()) is None)

 def test_mac_load_dialect_impl(self):
     dialect = self.mac.load_dialect_impl(sqlite.dialect())
     self.assertEqual(type(dialect), sqlite.CHAR)

 def test_process_result_value_with_value(self):
     bind = self.inet.process_result_value(1.0, sqlite.dialect())
     self.assertEqual(bind, 1.0)

 def test_process_bind_param_with_value(self):
     bind = self.inet.process_bind_param("foo", sqlite.dialect())
     self.assertEqual(bind, "foo")

 def test_inet_load_dialect_impl_sqlite(self):
     dialect = self.inet.load_dialect_impl(sqlite.dialect())
     self.assertEqual(type(dialect), sqlite.CHAR)

from sqlalchemy.dialects import sqlite, mysql

__all__ = ['sqlite_dialect', 'mysql_dialect', 'create_table_sql', 'sqlite3']

sqlite_dialect = sqlite.dialect()
mysql_dialect = mysql.dialect()

def create_table_sql(table, dialect):
    from sqlalchemy.schema import CreateTable
    return str(CreateTable(table).compile(dialect=dialect)).strip()

def sqlite3(db_file, command):
    import subprocess
    return subprocess.check_output(['sqlite3', db_file, command])

 def test_match_1(self):
     self.assert_compile(self.table1.c.myid.match('somstr'),
                     "mytable.myid MATCH ?",
                     dialect=sqlite.dialect())