文档章节

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:\Users\Administrator\Desktop\卡驱动\myKeyApi.dll') handle=0#设备句柄 ret=2 ret = lib.XF_EnumDev(0,'0') ret = lib.XF_Ope......

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

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

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

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

机器之心
2018/07/08
0
0
db2 卸载和安装

Db2 卸载步骤(参考) 开始卸载 1. 删除所有数据库。 可以使用“控制中心”或drop database命令删除数据库。笔者卸载而未删除数据库,结果是重新安装后无法建立同名数据库。 若要显示所有已创...

银河zlm
2016/04/15
0
0
Perl 脚本实现MySQL 异机导入导出

单位的开发同事需要将一个mysql 中的库导入到本地mysql中 所以为了减少手工操作,就写了下面的一个用perl 实现的 mysql 异机导入导出脚本:如下 #!/usr/bin/perl -w Author:andylhz Date:201...

andylhz
2012/09/03
0
0

没有更多内容

加载失败,请刷新页面

加载更多

OSChina 周一乱弹 —— 白掌柜说了卖货不卖身

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @爱漫爱 :这是一场修行分享羽肿的单曲《Moony》 手机党少年们想听歌,请使劲儿戳(这里) @clouddyy :开不开心? 开心呀, 我又不爱睡懒觉…...

小小编辑
今天
9
0
大数据教程(11.7)hadoop2.9.1平台上仓库工具hive1.2.2搭建

上一篇文章介绍了hive2.3.4的搭建,然而这个版本已经不能稳定的支持mapreduce程序。本篇博主将分享hive1.2.2工具搭建全过程。先说明:本节就直接在上一节的hadoop环境中搭建了! 一、下载apa...

em_aaron
今天
3
0
开始看《JSP&Servlet学习笔记》

1:WEB应用简介。其中1.2.1对Web容器的工作流程写得不错 2:编写Servlet。搞清楚了Java的Web目录结构,以及Web.xml的一些配置作用。特别是讲了@WebServlet标签 3:请求与响应。更细致的讲了从...

max佩恩
今天
4
0
mysql分区功能详细介绍,以及实例

一,什么是数据库分区 前段时间写过一篇关于mysql分表的的文章,下面来说一下什么是数据库分区,以mysql为例。mysql数据库中的数据是以文件的形势存在磁盘上的,默认放在/mysql/data下面(可...

吴伟祥
今天
5
0
SQL语句查询

1.1 排序 通过order by语句,可以将查询出的结果进行排序。放置在select语句的最后。 格式: SELECT * FROM 表名 ORDER BY 排序字段ASC|DESC; ASC 升序 (默认) DESC 降序 1.查询所有商品信息,...

stars永恒
今天
5
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部