Today.py

from util.dnodes import *
from util.log import *
from util.sql import *
from util.cases import *
import datetime


class TDTestCase:

    def init(self, conn, logSql):
        tdLog.debug(f"start to excute {__file__}")
        tdSql.init(conn.cursor())
        self.today_date = datetime.datetime.strptime(
            datetime.datetime.now().strftime("%Y-%m-%d"), "%Y-%m-%d")
        self.time_unit = ['b','u','a','s','m','h','d','w']
        self.error_param = ['1.5','abc','!@#','"abc"','today()']
        self.arithmetic_operators = ['+','-','*','/']
        self.relational_operator = ['<','<=','=','>=','>']
        # prepare data
        self.ntbname = 'ntb'
        self.stbname = 'stb'
        self.column_dict = {
            'ts':'timestamp',
            'c1':'int',
            'c2':'float',
            'c3':'double',
            'c4':'timestamp'
        }
        self.tag_dict = {
            't0':'int'
        }
        self.tbnum = 2
        self.tag_values = [
            f'10',
            f'100'
        ]
        self.values_list = [f'now,1,1.55,100.555555,today()',
                        f'now+1d,10,11.11,99.999999,now()',
                        f'today(),3,3.333,333.333333,now()',
                        f'today()-1d,10,11.11,99.999999,now()',
                        f'today()+1d,1,1.55,100.555555,today()']
                       
    def set_create_normaltable_sql(self, ntbname, column_dict):
        column_sql = ''
        for k, v in column_dict.items():
            column_sql += f"{k} {v},"
        create_ntb_sql = f'create table {ntbname} ({column_sql[:-1]})'
        return create_ntb_sql
    def set_create_stable_sql(self,stbname,column_dict,tag_dict):
        column_sql = ''
        tag_sql = ''
        for k,v in column_dict.items():
            column_sql += f"{k} {v},"
        for k,v in tag_dict.items():
            tag_sql += f"{k} {v},"
        create_stb_sql = f'create table {stbname} ({column_sql[:-1]}) tags({tag_sql[:-1]})'
        return create_stb_sql
    def data_check(self,column_dict={},tbname = '',values_list = [],tb_num = 1,tb = 'tb'):
        for k,v in column_dict.items():
            num_up = 0
            num_down = 0
            num_same = 0
            if v.lower() == 'timestamp':
                tdSql.query(f'select {k} from {tbname}')
                for i in tdSql.queryResult:
                    if i[0] > self.today_date:
                        num_up += 1
                    elif i[0] == self.today_date:
                        num_same += 1
                    elif i[0] < self.today_date:
                        num_down += 1
                tdSql.query(f"select today() from {tbname}")
                tdSql.checkRows(len(values_list)*tb_num)
                tdSql.checkData(0, 0, str(self.today_date))
                tdSql.query(f"select * from {tbname} where {k}=today()")
                if tb == 'tb':
                    tdSql.checkRows(num_same*tb_num)
                elif tb == 'stb':
                    tdSql.checkRows(num_same)
                for i in [f'{tbname}',f'db.{tbname}']:
                    for unit in self.time_unit:
                        for symbol in ['+','-']:
                            tdSql.query(f"select today() {symbol}1{unit} from {i}")
                            tdSql.checkRows(len(values_list)*tb_num)
                    for unit in self.error_param:
                        for symbol in self.arithmetic_operators:
                            tdSql.error(f'select today() {symbol}{unit} from {i}')
                    for symbol in self.arithmetic_operators:
                        tdSql.query(f'select now(){symbol}null from {i}')
                        tdSql.checkData(0,0,None)
                    for symbol in self.relational_operator:
                        tdSql.query(f'select * from {i} where {k} {symbol} today()')
                        if symbol == '<' :
                            if tb == 'tb':
                                tdSql.checkRows(num_down*tb_num)
                            elif tb == 'stb':
                                tdSql.checkRows(num_down)
                        elif symbol == '<=':
                            if tb == 'tb':
                                tdSql.checkRows((num_same+num_down)*tb_num)
                            elif tb == 'stb':
                                tdSql.checkRows(num_same+num_down)
                        elif symbol == '=':
                            if tb == 'tb':
                                tdSql.checkRows(num_same*tb_num)
                            elif tb == 'stb':
                                tdSql.checkRows(num_same)
                        elif symbol == '>=':
                            if tb == 'tb':
                                tdSql.checkRows((num_up + num_same)*tb_num)
                            elif tb == 'stb':
                                tdSql.checkRows(num_up + num_same)
                        elif symbol == '>':
                            if tb == 'tb':
                                tdSql.checkRows(num_up*tb_num)
                            elif tb == 'stb':
                                tdSql.checkRows(num_up)
                tdSql.query(f"select today()/0 from {tbname}")
                tdSql.checkRows(len(values_list)*tb_num)
                tdSql.checkData(0,0,None)
                tdSql.query(f"select {k} from {tbname} where {k}=today()")
                if tb == 'tb':
                    tdSql.checkRows(num_same*tb_num)
                    for i in range(num_same*tb_num):
                        tdSql.checkData(i, 0, str(self.today_date))
                elif tb == 'stb':
                    tdSql.checkRows(num_same)
                    for i in range(num_same):
                        tdSql.checkData(i, 0, str(self.today_date))
    def today_check_ntb(self):
        tdSql.prepare()
        tdSql.execute(self.set_create_normaltable_sql(self.ntbname,self.column_dict))
        for i in self.values_list:
            tdSql.execute(
                f'insert into {self.ntbname} values({i})')
        self.data_check(self.column_dict,self.ntbname,self.values_list)
        tdSql.execute('drop database db')
    def today_check_stb_tb(self):
        tdSql.prepare()
        tdSql.execute(self.set_create_stable_sql(self.stbname,self.column_dict,self.tag_dict))
        for i in range(self.tbnum):
            tdSql.execute(f'create table if not exists {self.stbname}_{i} using {self.stbname} tags({self.tag_values[i]})')
            for j in self.values_list:
                tdSql.execute(f'insert into {self.stbname}_{i} values ({j})')
        # check child table
        for i in range(self.tbnum):
            self.data_check(self.column_dict,f'{self.stbname}_{i}',self.values_list)
        # check stable
        self.data_check(self.column_dict,self.stbname,self.values_list,self.tbnum,'stb')
        tdSql.execute('drop database db')

    def run(self):  # sourcery skip: extract-duplicate-method
        
        tdLog.printNoPrefix("==========check today() for normal table ==========")
        self.today_check_ntb()
        tdLog.printNoPrefix("==========check today() for stable and child table==========")
        self.today_check_stb_tb()

    def stop(self):
        tdSql.close()
        tdLog.success(f"{__file__} successfully executed")

tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())