From 65f39050c3dee265d02f33e6d7f9093fdff068ec Mon Sep 17 00:00:00 2001 From: Miro Stauder Date: Thu, 12 Jan 2023 12:43:38 +0000 Subject: [PATCH] add test_sqlite3_from_unixtime-t --- .../tap/tests/test_sqlite3_from_unixtime-t.py | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 test/tap/tests/test_sqlite3_from_unixtime-t.py diff --git a/test/tap/tests/test_sqlite3_from_unixtime-t.py b/test/tap/tests/test_sqlite3_from_unixtime-t.py new file mode 100644 index 000000000..19efcf4d4 --- /dev/null +++ b/test/tap/tests/test_sqlite3_from_unixtime-t.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 + +import pymysql +import time + +def open_mysql_conn(host, port=3306, user=None, passwd=None, timeout=60): + conn = None + try: + if user is None and passwd is None: + conn = pymysql.connect( host=host, port=port, read_default_group='client', + connect_timeout=timeout, cursorclass=pymysql.cursors.DictCursor, + defer_connect=True) + else: + conn = pymysql.connect( host=host, port=port, user=user, passwd=passwd, + connect_timeout=timeout, cursorclass=pymysql.cursors.DictCursor, + defer_connect=True) + conn.client_flag |= pymysql.constants.CLIENT.MULTI_STATEMENTS + conn.connect() + conn.autocommit(True) + except Exception as e: + print(e) + raise + + return conn + +def padmin_command(command, display=False): + with padmin_conn.cursor() as cursor: + cursor.execute(command) + result = cursor.fetchall() + if display: + print(result) + return result + +def pmysql_command(command, display=False): + with pmysql_conn.cursor() as cursor: + cursor.execute(command) + result = cursor.fetchall() + if display: + print(result) + return result + +admin_conn_args = { 'host':'127.0.0.1', 'port':6032, 'user':'admin', 'passwd':'admin' } +mysql_conn_args = { 'host':'127.0.0.1', 'port':6033, 'user':'root', 'passwd':'root' } +puser_conn_args = { 'host':'127.0.0.1', 'port':6033, 'user':'user', 'passwd':'user' } + + +if __name__ == '__main__': + + padmin_conn = open_mysql_conn(**admin_conn_args) + + # test edge cases + assert padmin_command('SELECT from_unixtime();', display=True) == [{'from_unixtime()': None}] + assert padmin_command('SELECT from_unixtime(0);', display=True) == [{'from_unixtime(0)': '1970-01-01 00:00:00'}] + assert padmin_command('SELECT from_unixtime(-1);', display=True) == [{'from_unixtime(-1)': '1969-12-31 23:59:59'}] + assert padmin_command('SELECT from_unixtime(2147483647);', display=True) == [{'from_unixtime(2147483647)': '2038-01-19 03:14:07'}] + assert padmin_command('SELECT from_unixtime(2147483648);', display=True) == [{'from_unixtime(2147483648)': '2038-01-19 03:14:08'}] + # test values from monitor.mysql_server_ping_log + results = padmin_command('SELECT time_start_us/1000000, from_unixtime(time_start_us/1000000) FROM monitor.mysql_server_ping_log;', display=True) + for res in results: + assert time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(int(res['time_start_us/1000000']))) == res['from_unixtime(time_start_us/1000000)'] + + # PASS + exit(0)