您的位置：首页 > 其它

jmeter+jenkins+ant发送邮件

2017-11-09 16:06 465 查看

准备工作：

jenkins+jmeter+ant+mysql+python

流程图：

安装配置篇：

1、jmeter+jenkins+ant 安装略

2、python3.6版本安装略（配置环境变量；安装pyinstaller：pip install pyinstaller；安装jinjia2： pip install jinjia2 官网：http://docs.jinkan.org/docs/jinja2/）

3、jenkins ant构建完成之后，构建execute Windows batch command执行python命令C:\Users\kevin\dist\SelectData.exe

代码如下：

SendEmail.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import smtplib
from email.mime.text import MIMEText
from email.header import Header
import time
from jinja2 import Template

def test01(str):
mail_host="smtp.miao.cn"
mail_user="weiqqidong@miao.cn"
mail_pass="qw"
sslPort="465"
time1=time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))

sender = 'weixidong@miao.cn'
receivers = ['weiqqidong@miao.cn']

#  message = MIMEText(time1+'_'+str, 'plain', 'utf-8')
message = MIMEText(str, _subtype='html', _charset='utf-8')
message['From'] = Header("weiqqidong@miao.cn", 'utf-8')
message['To'] =  Header("weiqqidong@miao.cn", 'utf-8')

subject ='接口自动化测试报告'+'-'+time1
message['Subject'] = Header(subject, 'utf-8')

try:
# smtpObj = smtplib.SMTP()
# smtpObj.connect(mail_host, 25)
#  smtpObj.login(mail_user,mail_pass)

smtpObj = smtplib.SMTP_SSL(mail_host,sslPort)
smtpObj.ehlo()
smtpObj.login(mail_user,mail_pass)
smtpObj.sendmail(sender, receivers, message.as_string())
print ("邮件发送成功")

except Exception as n:
print ("Error: 无法发送邮件")
print(n)

ReturnData.py

#coding=utf-8
__author__ = 'kevin'
import pymysql
from Send.SendEmail import test01
import json
from jinja2 import Template
import time

def recode():
#  time2=time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
demo_list= list()
# 连接数据库
connect = pymysql.Connect(
host='172.31.85.110',
port=3306,
user='root',
passwd='a12345',
db='autotest',
charset='utf8'
)

t= '''
<html>
<body>
<table border=1 cellpadding=10>
<tr>
<h4>异常业务接口详情:</h4>
<table border="1">
<tr>
<td width="200"><b>name</b></td>
<td width="200"><b>status</b></td>
<td width="200"><b>msg</b></td>
</tr>
</tr>
{% for d in demo %}
<tr>
<td>
{{ d.name }}
</td>
<td>
{{ d.status }}
</td>
<td>
{{ d.msg }}
</td>
</tr>
{% endfor %}
</table>
</body>
</html>
'''
cursor = connect.cursor()
# 查询异常数据
sql="SELECT a.name ,b.status,b.msg FROM interfacename AS a,DATA AS b WHERE a.id=b.InterfaceId AND b.status<>200 AND createtime>(SELECT DATE_SUB(NOW(), INTERVAL 1 HOUR)) AND createtime<(SELECT DATE_ADD(NOW(), INTERVAL 1 HOUR))"
try:
cursor.execute(sql)
results = cursor.fetchall()
print(results)
for row in results:
name=row[0]
status=row[1]
msg=row[2]
demo = dict()
demo["name"] = name
demo["status"] = status
demo["msg"] = msg
# print(msg,type(msg))
demo_list.append(demo)
# return demo_list
# 打印结果
#print("name=%s,status=%s,msg=%s" % (row[0],row[1],row[2] ))
# print("name:%s status:%s msg:%s"%(row[0],row[1],row[2] ))
#map(None,row[0])
template = Template(t)
result = template.render(demo=demo_list)
# print(result)
# print(demo_list)

test01(result)
except Exception as e:
print(e)
print("Error: unable to fecth data")
cursor.close()
connect.close()

selectdata.py

#coding=utf-8
__author__ = 'kevin'

import pymysql
from Send.SendEmail import test01
from Send.ReturnData import recode
import time
# 连接数据库
connect = pymysql.Connect(
host='172.31.85.110',
port=3306,
user='root',
passwd='a12345',
db='autotest',
charset='utf8'
)
cursor = connect.cursor()
# 查询数据
sql="SELECT COUNT(1) FROM DATA WHERE STATUS<>200 and createtime>(SELECT DATE_SUB(NOW(), INTERVAL 1 HOUR)) AND createtime<(SELECT DATE_ADD(NOW(), INTERVAL 1 HOUR))"
cursor.execute(sql)
data = cursor.fetchone()
print ("error data count is : %s " % data)
print(data[0])

if data[0]!= 0:
recode()
else:
print("dddd")
cursor.close()
connect.close()

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航