文档章节

db_help.py

xxjbs001
 xxjbs001
发布于 2014/06/12 19:34
字数 823
阅读 6
收藏 0
#!/usr/bin/env python
# coding=utf-8
from optparse import OptionParser
import sys, time, os, re, logging
from pprint import pprint, pformat
from traceback import format_exc
from copy import deepcopy
import MySQLdb
from datetime import datetime


class DBI():
    """
    """
    _host = ''
    _port = ''
    _user = ''
    _passwd = ''
    _dbname = ''
    _conn_timeout = 30
    _logger = None

    def __init__(self, host='127.0.0.1', port='3306', user='root', passwd='123qaz', dbname='automation_test', conn_timeout=30, logger=None):
        """
        """
        (self._host, self._port, self._user, self._passwd, self._dbname, self._conn_timeout) = (host, port, user, passwd, dbname, conn_timeout)
        self._logger = logger
        if not logger:
            logger = logging.getLogger("TEST")
            logger.setLevel(logging.DEBUG)
            stdhdlr = logging.StreamHandler(sys.stdout)
            # FORMAT = '[%(levelname)s] %(message)s'
            FORMAT = '%(message)s'
            stdhdlr.setFormatter(logging.Formatter(FORMAT))
            logger.addHandler(stdhdlr)
            self._logger = logger
            pass
        else:
            self._logger = logger

        pass

    def _openDB(self):
        """
        """
        conn = None
        try:
            conn = MySQLdb.connect(host=self._host, user=self._user, passwd=self._passwd, port=int(self._port), connect_timeout=self._conn_timeout)
            conn.select_db(self._dbname)
        except Exception, e:
            self._logger.error('open db failed', exc_info=1)
            pass

        return conn

    def exec_SQL(self, sql):
        """
        """
        conn = self._openDB()
        r = False
        result_set = None
        if conn:
            try:
                cursor = conn.cursor()
                r = cursor.execute(sql)
                conn.commit()
                result_set = cursor.fetchall()
                cursor.close()
                conn.close()
                pass
            except Exception, e:
                self._logger.error("exec_SQL except :", exc_info=1)
        else:
            pass

        return (r, result_set)

    def exec_SQLs(self, sqls):
        """
        """
        conn = self._openDB()
        r = False
        if conn:
            try:
                cursor = conn.cursor()
                r = []
                for sql in sqls:
                    _r = cursor.execute(sql)
                    r.append(_r)
                    pass

                conn.commit()
                cursor.close()
                conn.close()
                # return r
                pass
            except Exception, e:
                self._logger.error("exec_SQLs except :", exc_info=1)
                pass
        else:
            pass

        return r

    def batch_SQL(self, sql_str, values):
        """
        """
        conn = self._openDB()
        r = False
        if conn:
            try:
                cursor = conn.cursor()
                r = cursor.executemany(sql_str, values)
                conn.commit()
                cursor.close()
                conn.close()
            except Exception, e:
                self._logger.error("batch_SQL except :", exc_info=1)
                pass

            # return r
        else:
            pass

        return r

def insert_variable(testbed='', key='', value=''):
    """
    """
    pass
    host = os.getenv('U_CUSTOM_AT_DB_HOST', '192.168.20.108')
    port = os.getenv('U_CUSTOM_AT_DB_PORT', '3306')
    user = os.getenv('U_CUSTOM_AT_DB_USERNAME', 'root')
    passwd = os.getenv('U_CUSTOM_AT_DB_PASSWORD', '123qaz')
    dbname = os.getenv('U_CUSTOM_AT_DB_NAME', 'automation_test')
    dbi = DBI(host=host, port=port, user=user, passwd=passwd, dbname=dbname)
    sql = "SELECT * FROM t_am_mediaroom_variables where testbed ='" + str(testbed) + "' AND var_item='" + str(key) + "'"
    r = dbi.exec_SQL(sql)
    if int(r[0]):
        sql = "UPDATE t_am_mediaroom_variables SET var_value='" + value + "' WHERE testbed='" + str(testbed) + "' AND var_item='" + str(key) + "'"
    else:
        fmt = "INSERT INTO t_am_mediaroom_variables VALUES(NULL,'%s','%s','%s')"
        sql = fmt % (testbed, key, value)
    r = dbi.exec_SQL(sql)
    pass

def delete_variable(id=''):
    """
    """
    pass
    id = str(id)
    host = os.getenv('U_CUSTOM_AT_DB_HOST', '192.168.20.108')
    port = os.getenv('U_CUSTOM_AT_DB_PORT', '3306')
    user = os.getenv('U_CUSTOM_AT_DB_USERNAME', 'root')
    passwd = os.getenv('U_CUSTOM_AT_DB_PASSWORD', '123qaz')
    dbname = os.getenv('U_CUSTOM_AT_DB_NAME', 'automation_test')
    dbi = DBI(host=host, port=port, user=user, passwd=passwd, dbname=dbname)
 
    fmt = "DELETE FROM t_am_mediaroom_variables WHERE var_id=('%s')"
    sql = fmt % (id)
    # print('AT_INFO : sql(%s)' % (sql))
 
    r = dbi.exec_SQL(sql)
    pass



def insert_sample(task_start_time='', testbed='', product='', dut_br0_mac='', case='', sample_time='', sample_item='', sample_value='', logger=None):
    """
    """
    pass
    host = os.getenv('U_CUSTOM_AT_DB_HOST', '192.168.20.108')
    port = os.getenv('U_CUSTOM_AT_DB_PORT', '3306')
    user = os.getenv('U_CUSTOM_AT_DB_USERNAME', 'root')
    passwd = os.getenv('U_CUSTOM_AT_DB_PASSWORD', '123qaz')
    dbname = os.getenv('U_CUSTOM_AT_DB_NAME', 'automation_test')
    dbi = DBI(host=host, port=port, user=user, passwd=passwd, dbname=dbname, logger=logger)
    if not task_start_time :
        task_start_time = os.getenv('U_CUSTOM_TEST_TASK_START_TIME', 'UNKNOWN')
        pass
 
    if not testbed :
        testbed = os.getenv('G_TBNAME', 'UNKNOWN')
        pass
    if not product :
        product = os.getenv('U_DUT_TYPE', 'UNKNOWN')
        pass
    if not dut_br0_mac :
        dut_br0_mac = os.getenv('G_PROD_MAC_BR0_0_0', 'UNKNOWN')
        pass
    if not case :
        case = os.getenv('U_CUSTOM_CURRENT_CASE_ID', 'MediaRoomTest')
        pass
    if not sample_time :
        sample_time = datetime.now()
        pass
 
    fmt = "INSERT INTO t_am_mediaroom_sample VALUES(0,'%s','%s','%s','%s','%s','%s','%s','%s')"
    sql = fmt % (task_start_time, testbed, product, dut_br0_mac, case, sample_time, sample_item, sample_value)
    # print('AT_INFO : sql(%s)' % (sql))
 
    r = dbi.exec_SQL(sql)
    pass
def insert_roy(item_name='',item_value='',task_start_time=''):
    """
    """
def update_traffic(iperf_num='', iperf_band='', curl_num='', curl_band='', testbed='', logger=None):
    """
    """
    host = os.getenv('U_CUSTOM_AT_DB_HOST', '192.168.20.108')
    port = os.getenv('U_CUSTOM_AT_DB_PORT', '3306')
    user = os.getenv('U_CUSTOM_AT_DB_USERNAME', 'root')
    passwd = os.getenv('U_CUSTOM_AT_DB_PASSWORD', '123qaz')
    dbname = os.getenv('U_CUSTOM_AT_DB_NAME', 'automation_test')
    dbi = DBI(host=host, port=port, user=user, passwd=passwd, dbname=dbname, logger=logger)
    sql = "UPDATE t_am_mediaroom_iperf SET "
    flag = False
    if iperf_band:
        sql += "iperf_band=" + "'" + str(iperf_band) + "'"
        flag = True
        
    if iperf_num:
        if flag:
            sql += ",iperf_num=" + "'" + str(iperf_num) + "'"
        else:
            sql += "iperf_num=" + "'" + str(iperf_num) + "'"
            flag = True
            
    if curl_num:
        if flag:
            sql += ",curl_num=" + "'" + str(curl_num) + "'"
        else:
            sql += "curl_num=" + "'" + str(curl_num) + "'"
            flag = True
    
    if curl_band:
        if flag:
            sql += ",curl_band=" + "'" + str(curl_band) + "'"
        else:
            sql += "curl_band=" + "'" + str(curl_band) + "'"
            flag = True
            
    sql += " where testbed ='" + str(testbed) + "'"
    
    if flag:
        r = dbi.exec_SQL(sql)

def get_env(testbed='', logger=None):
    """
    """
    host = os.getenv('U_CUSTOM_AT_DB_HOST', '192.168.20.108')
    port = os.getenv('U_CUSTOM_AT_DB_PORT', '3306')
    user = os.getenv('U_CUSTOM_AT_DB_USERNAME', 'root')
    passwd = os.getenv('U_CUSTOM_AT_DB_PASSWORD', '123qaz')
    dbname = os.getenv('U_CUSTOM_AT_DB_NAME', 'automation_test')
    dbi = DBI(host=host, port=port, user=user, passwd=passwd, dbname=dbname, logger=logger)
    sql = "SELECT * FROM t_am_mediaroom_variables where testbed ='" + str(testbed) + "'"
    
    r = dbi.exec_SQL(sql)
    return r

def testDBI():
    """
    """
    dbi = DBI()
    print('--' * 16)

    # sql = 'INSERT INTO t_am_keywordlib VALUES(%s,%s,%s,%s,%s,%s,%s)'
    # values = []
    # values.append(('','raytest','','','','CMD : ls\nCMD : ifconfig','ray Test'))
    # r = dbi.batch_SQL(sql,values)
    # print(r)
    task_start_time = datetime.now()
    testbed = 'at_sh1'
    product = 'CTLC2KA'
    dut_br0_mac = 'aa:bb:cc:dd:ee:ff'
    case = '99900001'
    sample_time = datetime.now()
    sample_item = 'unit test'
    sample_value = 'Pass'
    comment = ''

    fmt = "INSERT INTO t_am_stability_sampling VALUES(0,'%s','%s','%s','%s','%s','%s','%s','%s')"
    sql = fmt % (task_start_time, testbed, product, dut_br0_mac, case, sample_time, sample_item, sample_value)

    r = dbi.exec_SQL(sql)

    pass


if __name__ == '__main__':
    """
    """
    testDBI()
    print('== DONE ==')



© 著作权归作者所有

共有 人打赏支持
xxjbs001
粉丝 55
博文 749
码字总数 215621
作品 0
浦东
QA/测试工程师
私信 提问
python嵌套C语言、调用外部C的dll,python实参如何传递并修改??

# -- coding: cp936 -- import ctypes lib=ctypes.CDLL(r'C:UsersAdministratorDesktop卡驱动myKeyApi.dll') handle=0#设备句柄 ret=2 ret = lib.XFEnumDev(0,'0') ret = lib.XFOpenDev(0,h......

iPanH
2013/10/08
1K
1
入门Python神经机器翻译,这是一篇非常精简的实战指南

选自Medium,作者:Susan Li,机器之心编译。 机器翻译(MT)是一项极具挑战性的任务,其研究如何使用计算机将文本或是语音从一种语言翻译成另一种语言。本文借助 Keras 从最基本的文本加载与...

机器之心
07/09
0
0
教程 | 入门Python神经机器翻译,这是一篇非常精简的实战指南

  选自Medium   作者:Susan Li   机器之心编译   参与:Huiyuan Zhuo、思源      机器翻译(MT)是一项极具挑战性的任务,其研究如何使用计算机将文本或是语音从一种语言翻译成...

机器之心
07/08
0
0
RedHat Enterprise Linux 5.4上图形或者手动安装DB2 V9.7

Linux(RedHat Enterprise Linux 5.4)下的DB2安装使用图形界面 可以使用图形界面,也可以使用手工安装的方式。我们先介绍最简单的安装方法。 1.从IBM网站下载DB2 9.7 express C版本:文件名d...

鉴客
2010/07/05
2.1K
0
db2 v10.5 HADR快速部署手册

一、目录准备 mkdir -p /home/db2inst1/db2backup mkdir -p /home/db2inst1/db2archive mkdir -p /home/db2inst1/db2log chmod -R 775 /home/db2inst1/db2backup chmod -R 775 /home/db2inst......

18620626259
2017/12/20
0
0

没有更多内容

加载失败,请刷新页面

加载更多

小程序异步操作 跨js执行 在微信小程序里面实现跨页面通信

我们知道,在小程序里面一个页面的变化,是通过调用 setData 函数来实现的。所以想做到在二级页面里让一级页面产生变化,最 Quick And Dirty 的做法就是把一级页面的 this 传入到二级页面去,...

xiaogg
13分钟前
0
0
授于管理员登录其它用户

1.沙盒中,授予管理员登录 安全性控制==>登录访问权限政策

在山的那边
16分钟前
1
0
线程安全的CopyOnWriteArrayList介绍

证明CopyOnWriteArrayList是线程安全的 先写一段代码证明CopyOnWriteArrayList确实是线程安全的。 ReadThread.java import java.util.List; public class ReadThread implements Runnable {......

绝地逢生
18分钟前
0
0
Java重写的7个规则

几年前你可能会遇到这样一个面试题:“重写和重载的区别”、而现在随着科技的更迭、面试的问题越来越高级、面试官的问题也越来越深入、此文是上述面试题的一个延伸、让你从简单的重写规则中更...

architect刘源源
18分钟前
1
0
JavaScript异步编程:Generator与Async

从Promise开始,JavaScript就在引入新功能,来帮助更简单的方法来处理异步编程,帮助我们远离回调地狱。 Promise是下边要讲的Generator/yield与async/await的基础,希望你已经提前了解了它。...

前端攻城老湿
19分钟前
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部