123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322 |
- import pytest
- import pymysql
- def test_insert_with_single_quotes():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "insert into opensource(uid, name) values(1, 'hello')"
- cursor.execute(sql)
- db.commit()
- rowsaffected = cursor.rowcount
- print("affected rows: %s" % (rowsaffected))
- cursor.close()
- db.close()
- assert rowsaffected == 1
- def test_select():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "select uid, name from opensource where uid = 1"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 1
- for row in results:
- uid = row[0]
- name = row[1]
- print("uid=%s, name=%s" % (uid, name))
- assert uid == 1
- assert name == "hello"
- db.close()
- def test_update():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "select uid, name from opensource where uid = 1"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 1
- for row in results:
- uid = row[0]
- name = row[1]
- print("uid=%s, name=%s" % (uid, name))
- assert uid == 1
- assert name == "hello"
- cursor.close()
- cursor = db.cursor()
- sql = "update opensource set name = 'Lee' where uid = 1"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "update opensource set name = \"Lee3\" where uid = 1"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "update opensource set `name` = 'Lee2' where uid = 1"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "update opensource set \"name\" = 'Lee' where uid = 1"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
- sql = "update opensource set name = \"Lee4\" where uid = '1'"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
- sql = "update opensource set name = \"Lee\" where uid = \"1\""
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
-
- sql = "update opensource set name = \"Lee\" where uid = 1"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- db.commit()
- cursor.close()
- cursor = db.cursor()
- sql = "select uid, name from opensource where uid = 1"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 1
- for row in results:
- uid = row[0]
- name = row[1]
- print("uid=%s, name=%s" % (uid, name))
- assert uid == 1
- assert name == "Lee"
- cursor.close()
- db.close()
- def test_delete():
- print("----delete----")
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "select uid, name from opensource where uid = 1"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 1
- cursor.close()
- cursor = db.cursor()
- sql = "delete from opensource where uid = 1"
- cursor.execute(sql)
- db.commit()
- rowsaffected = cursor.rowcount
- print("affected rows: %s" % (rowsaffected))
- assert rowsaffected == 1
- cursor.close()
- cursor = db.cursor()
- sql = "select uid, name from opensource where uid = 1"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 0
- cursor.close()
- db.close()
- '''
- def test_check_tablename():
- '''
- def test_insert_remove_where_cluster():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "insert into opensource(uid, name) values(1, 'hello')"
- cursor.execute(sql)
- db.commit()
- rowsaffected = cursor.rowcount
- print("affected rows: %s" % (rowsaffected))
- cursor.close()
- db.close()
- assert rowsaffected == 1
- '''
- def test_insert_remove_where_cluster_without_specify_key():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "insert into opensource values(1, 'Jack', 'Shanghai', 1, 18)"
- cursor.execute(sql)
- db.commit()
- rowsaffected = cursor.rowcount
- print("affected rows: %s" % (rowsaffected))
- cursor.close()
- db.close()
- assert rowsaffected == 1
- '''
- def test_select_limit():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "insert into opensource(uid,name,city,sex,age) values(1, 'Jack', 'Shanghai', 1, 18)"
- cursor.execute(sql)
- db.commit()
- sql = "select uid, name from opensource where uid = 1 limit 2"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 2
-
- cursor = db.cursor()
- sql = "select uid, name from opensource where uid = 1 limit 1"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 1
- cursor = db.cursor()
- sql = "insert into opensource(uid,name,city,sex,age) values(1, 'Jack', 'Shanghai', 1, 19)"
- cursor.execute(sql)
- db.commit()
- cursor = db.cursor()
- sql = "select uid, name,city,sex,age from opensource where uid = 1 limit 1,3"
- cursor.execute(sql)
- results = cursor.fetchall()
- assert len(results) == 2
- assert results[0][4] == 18
- assert results[1][4] == 19
- db.close()
- def test_insert_automated_conversion():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- #name num to string
- sql = "insert into opensource(uid,name,city,sex,age) values(1, 123, 'Shanghai', 1, 18)"
- insert = cursor.execute(sql)
- assert insert == 1
- #name num to string
- sql = "insert into opensource(uid,name,city,sex,age) values(1, 123a, 'Shanghai', 1, 18)"
- insert = cursor.execute(sql)
- assert insert == 0
- #name num(float) to string
- sql = "insert into opensource(uid,name,city,sex,age) values(1, 123.3, 'Shanghai', 1, 18)"
- insert = cursor.execute(sql)
- assert insert == 1
- #name string to num
- sql = "insert into opensource(uid,name,city,sex,age) values(1, 'jack', 'Shanghai', 1, '18')"
- insert = cursor.execute(sql)
- assert insert == 1
- #name string to num
- sql = "insert into opensource(uid,name,city,sex,age) values(2, \"jack\", 'Shanghai', 1, '18')"
- insert = cursor.execute(sql)
- assert insert == 1
- #name string to num, error
- sql = "insert into opensource(uid,name,city,sex,age) values(1, 'jack', 'Shanghai', 1, '18a')"
- insert = cursor.execute(sql)
- assert insert == 0
- db.commit()
- db.close()
- def test_insert_with_double_quotes():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "insert into opensource(uid, name) values(33, \"hello\")"
- cursor.execute(sql)
- db.commit()
- rowsaffected = cursor.rowcount
- cursor.close()
- db.close()
- assert rowsaffected == 1
- def test_insert_with_grave():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
- sql = "insert into `opensource`(uid, name) values(33, 'hello')"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource(`uid`, name) values(33, 'hello')"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource(`uid`, `name`) values(33, 'hello')"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into `opensource`(`uid`, `name`) values(33, \"hello\")"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource(uid, name) values(33, `hello`)"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
- sql = "insert into opensource(uid, name) values(33, `123`)"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
- sql = "insert into \"opensource\"(uid, name) values(33, 'hello')"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
- sql = "insert into 'opensource'(uid, name) values(33, 'hello')"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
- db.commit()
- cursor.close()
- db.close()
- def test_insert_with_set_keyword():
- db = pymysql.connect(host='127.0.0.1', port=20015, user='test', password='test', database='test')
- cursor = db.cursor()
-
- sql = "insert into opensource set uid = 33, name = 'hello'"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set `uid` = 33, `name` = 'hello'"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set `uid` = 33, `name` = '12312'"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set `uid` = 33, `name` = \"12312\""
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set uid = 33, name = 'hello'"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set uid = 33, name = '12312'"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set uid = 33, name = \"12312waefoioiwa\""
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set uid = 33, name = \"waefoioiwawaefwa\""
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- sql = "insert into opensource set uid = 33, name = 123waefoioiwawaefwa"
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 0
- sql = "insert into opensource set uid = 33, name = \"12312\""
- rowsaffected = cursor.execute(sql)
- assert rowsaffected == 1
- db.commit()
- cursor.close()
- db.close()
-
|