文档章节

db_help.py

xxjbs001
 xxjbs001
发布于 2014/06/12 19:34
字数 823
阅读 6
收藏 0
点赞 0
评论 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
Linux下DB2数据库打补丁步骤经验总结

DB2补丁没有亲自实践过,记录别人安装作为参考。 首先停掉所有的实例 1、$db2ilist 结果有db2inst1 和db2inst2 (查看数据库实例$ db2level ) 2、两个都停掉(db2stop),各停止各的 su - db2...

学习环境
2015/12/04
43
0
db2 卸载和安装

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

银河zlm
2016/04/15
0
0
DB2常用命令

1. 启动实例(db2inst1): db2start 2. 停止实例(db2inst1): db2stop 3. 列出所有实例(db2inst1) db2ilist 3-1.列出当前实例: db2 get instance 3-2.列出所有数据库: db2 list database direc......

xujh1986
2016/02/25
0
0
CodeIgniter数据库连接时报错,已经在网上搜了一天了

在数据库连接是已经在database.php里修改$db['default']['hostname'] = 'localhost'; $db['default']['username'] = 'root'; $db['default']['password'] = '900130'; $db['default']['datab......

月下之风
2012/06/20
3K
3
DB2 HADR 配置示例

介绍   DB2的HA可以在OS级别使用专门的HA来实现,也可使用DB2内置的HADR来实现,且配置和管理很简单。   一个HADR环境需要两台数据库服务器:主数据库服务器(primary)和备用数据库服务...

adelphos
2015/02/05
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

Java8新特性之接口

在JDK8以前,我们定义接口类中,方法都是抽象的,并且不能存在静态方法。所有的方法命名规则基本上都是 public [返回类型] [方法名](参数params) throws [异常类型] {}。 JDK8为接口的定义带...

developlee的潇洒人生
29分钟前
0
0
aop + annotation 实现统一日志记录

aop + annotation 实现统一日志记录 在开发中,我们可能需要记录异常日志。由于异常比较分散,每个 service 方法都可能发生异常,如果我们都去做处理,会出现很多重复编码,也不好维护。这种...

长安一梦
40分钟前
1
0
将博客搬至CSDN

AHUSKY
51分钟前
1
0
Python web框架Django学习(1)

1.Django简介 (1)Python下有许多款不同的 Web 框架。Django是重量级选手中最有代表性的一位。许多成功的网站和APP都基于Django。Django是一个开放源代码的Web应用框架,由Python写成。 (2...

十年磨一剑3344
今天
0
0
Databook-数据之书

Databook-数据之书 用于数据分析的Jupyter Notebooks。 不需购买服务器,快速开始自己的数据分析过程。 源码:https://github.com/openthings/databook 作者:openthings,https://github.co...

openthings
今天
5
0
Python PIPEs

https://www.python-course.eu/pipes.php https://www.tutorialspoint.com/python/os_pipe.htm

zungyiu
今天
1
0
gRPC学习笔记

gRPC编程流程 1. proto文件定义 proto文件用于定义需要通过gRPC生成的接口,可以理解为接口定义文档 2. 通过构建工具生成服务基类代码-Maven或Gradle 3. 服务端开发 服务端实现类须实现通过构...

OSC_fly
今天
0
0
Docker Mac (三) Dockerfile 及命令

Dockerfile 最近学习docker的时候,遇到一件怪事,关于docker镜像可能会被破坏,还不知道它会有此措施 所以需要了解构建Dockerfile的正确方法 Dockerfile是由一系列命令和参数构成的脚本,这些命...

___大侠
今天
0
0
Android Studio+NDK+Cmake 移植FFmpeg-4.0.2命令行工具

一、编译 参考大神的帖子,亲测一次编译成功:https://blog.csdn.net/bobcat_kay/article/details/80889398 鉴于以前查文档的经验,这里附上编写例子的时间:2018年7月22日 我用的是ubantu,...

她叫我小渝
今天
0
0
mysql创建数据库

登录MYSQL mysql -u root -p 脚本创建数据库WeChat,并制定默认的字符集是utf8mb4。 CREATE DATABASE Wechat DEFAULT CHARSET utf8mb4 COLLATE utf8mb4_general_ci; 授权 grant all......

niithub
今天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部