test_basic.py 文件源码

def test_json(self):
        args = self.databases[0].copy()
        args["charset"] = "utf8mb4"
        conn = pymysql.connect(**args)
        if not self.mysql_server_is(conn, (5, 7, 0)):
            raise SkipTest("JSON type is not supported on MySQL <= 5.6")

        self.safe_create_table(conn, "test_json", """\
create table test_json (
    id int not null,
    json JSON not null,
    primary key (id)
);""")
        cur = conn.cursor()

        json_str = u'{"hello": "?????"}'
        cur.execute("INSERT INTO test_json (id, `json`) values (42, %s)", (json_str,))
        cur.execute("SELECT `json` from `test_json` WHERE `id`=42")
        res = cur.fetchone()[0]
        self.assertEqual(json.loads(res), json.loads(json_str))

        cur.execute("SELECT CAST(%s AS JSON) AS x", (json_str,))
        res = cur.fetchone()[0]
        self.assertEqual(json.loads(res), json.loads(json_str))