Python微信订餐小程序课程视频
https://edu.csdn.net/course/detail/36074
Python实战量化交易理财系统
https://edu.csdn.net/course/detail/35475
Jenkins Pipeline 的核心概念：
- Pipeline 是一套运行于
Jenkins上的工作流框架，将原本独立运行于单个或者多个节点的任务连接起来，实现单个任务难以完成的复杂流程编排与可视化。 - Pipeline是 Jenkins2.X 的
最核心的特性，帮助Jenkins 实现从CI到CD与DevOps的转变。 - Pipeline是一组插件，让
Jenkins可以实现持续交付管道的落地和实施。
持续交付管道（CD Pipeline）是将软件从版本控制阶段到交付给用户或客户的完整过程的自动化表现。软件的每一次更改（提交到源代码管理系统）都要经过一个复杂的过程才能被发布。
Pipeline提供了一组可扩展的工具，通过Pipeline Domain Specific Language（DSL）syntax可以达到Pipeline as Code（Jenkinsfile存储在项目的源代码库）的目的。
Pipeline入门：
先决条件
要使用Jenkins Pipeline，您将需要：
- Jenkins 2.x或更高版本
- Pipeline插件（请自行在插件管理中安装。）
Pipeline 定义
脚本Pipeline是用Groovy写的 。Groovy相关语法请移步>。
Pipeline支持两种语法：
- Declarative 声明式
- Scripted pipeline 脚本式
如何创建基本Pipeline：
- 直接在Jenkins网页界面中输入脚本。
- 通过创建一个Jenkinsfile可以检入项目的源代码管理库。
用任一方法定义Pipeline的语法是一样的，但是Jenkins支持直接进入Web UI的Pipeline，通常认为最佳实践是在Jenkinsfile Jenkins中直接从源代码控制中加载Pipeline。
在Web UI中定义Pipeline
要在Jenkins Web UI中创建基本Pipeline：
- 单击Jenkins主页上的New Item。
- 输入Pipeline的名称，选择Pipeline，然后单击确定。
- 在脚本文本区域中，输入Pipeline，然后单击保存。
- 单击立即生成以运行Pipeline，查看控制台输出。
Pipeline几个核心概念：
-
Stages：阶段组/Stage：阶段
- 一个 Pipeline 有多个 Stage 组成，每个 Stage 包含一组 Step。
- 注意一个 Stage 可以跨多个 Node 执行，即 Stage 实际上是 Step 的逻辑分组。
- 一个Jenkinsfile 可以分为大的阶段，如打包、构建、 部署。测试
- 构建的流程，可以分为这几步，获取源代码，然后打包，构建，进行编译，替换配置文件，编译完打包，进行部署 这个阶段就是stage
- Node：节点，一个Node就是一个Jenkins节点，或者是Master，或者是Agent，是执行Step的具体运行环境。
- Steps：步骤，Step是最基本的操作单元，小到创建一个目录，大到构建一个Docker镜像，由各类Jenklins Plugin提供，例如：sh ‘make’
Pipeline几个核心关键字：
- 块（blocks{}）:
由大括号括起来的语句，如pipeline{},Section{},parameters{},script{} - 章节（Sections）:
通常包含一个或多个指令或步骤。如 agent 、post、stages、steps - 指令（Directives）:
environment、options、parameters、triggers（触发）、stage、tools、when - 步骤（Steps）:
执行脚本式pipeline：在该语句块内使用script{} - agent
必须存在，agent必须在pipeline块内的顶层定义，但stage内是否使用是可选的
参数：any/none/label/node/docker/dockerfile
常用选项 label/cuetomWorkspace/reuseNode
| 指令名 | 说明 | 作用域 |
|---|---|---|
| agent | 定义执行任务的代理 | stage 或pipeline |
| input | 暂停pipeline,提示输入内容 | stage |
| environment | 设置环境变量 | stage或pipeline |
| tools | 自动下载并安装指定的工具，并将其加入到PATH变量中 | stage或pipeline |
| options | 配置Jenkins pipeline本身，如options{retry(3}},指pipeline失败时再重试2次 | stage 或 pipeline |
| build | 触发其他的job | steps |
| when | 定义阶段执行的条件 | stage |
| triggers | 定义执行pipeline的触发器 | pipeline |
| parameters | 执行pipeline前传入一些参数 | pipeline |
| parallel | 并行执行多个step | stage |
示例：
- agent：
agent { label 'my-label' }
agent {
node {
label 'my-label'
customWorkspace '/some/other/path'
}
}
agent {
docker {
image 'application_name:verison'
label 'my-label'
args '-v /tmp:/tmp'
}
}
- stage间通过stash进行文件共享，即使stage不在同一个执行主机上：
pipeline{
agent none
stages{
stage('stash'){
agent { label "master" }
steps{
writeFile file: "a.txt", text: "$BUILD_NUMBER"
stash name: "abc", includes: "a.txt"
}
}
stage('unstash'){
agent { label "node" }
steps{
script{
unstash("abc")
def content = readFile("a.txt")
echo "${content}"
}
}
}
}
}
- steps中的一些操作：
| 命令名 | 说明 |
|---|---|
| error | 抛出异常，中断整个pipeline |
| timeout | timeout闭包内运行的步骤超时时间 |
| waitUntil | 一直循环运行闭包内容，直到return true，经常与timeout同时使用 |
| retry | 闭包内脚本重复执行次数 |
| sleep | 暂停pipeline一段时间，单位为秒 |
pipeline{
agent any
stages{
stage('stash'){
steps{
timeout(50){
waitUntil{
script{
def r = sh script: 'curl http://xxx', returnStatus: true
return (r == 0)
}
}
}
retry(10){
script{
sh script: 'curl http://xxx', returnStatus: true
}
}
sleep(20)
}
}
}
}
- triggers：定时构建
pipeline {
agent any
triggers {
cron('H 9 * * *')
}
}
-
paramparameters：参数化构建
- pipeline 脚本
pipeline { agent any parameters { choice(name: 'ENV', choices: 'dev\nsit\nuat', description: '环境') // 或者 choice(name: 'ENV', choices: ['dev','sit',uat'], description: '环境') string(name: 'PROJECT', defaultValue: 'example-demo', description: '项目') booleanParam(defaultValue: true, description: '', name: 'BOOLEAN') text(defaultValue: '''this is a multi-line string parameter example ''', name: 'MULTI-LINE-STRING') } stages { stage('Hello') { steps { echo 'Hello World' } } } }
2. web\_ui配置：  也可用于选择git分支/tag 进行发布 1. 添加参数类型为：  2. 编写groovy脚本： **替换掉自己的git项目地址！**def gettags = ("git ls-remote --heads --tags ssh://git@{ your ip }/xx/project_name.git").execute()
if (ENV.equals("pre")){
gettags = ("git ls-remote --heads --tags ssh://git@{ your ip }/xx/project_name.git").execute()
}
def repoNameList = gettags.text.readLines().collect {
it.split()[1].replaceAll('refs/heads/', '').replaceAll('refs/tags/', '').replaceAll("\^\{\}", '')
}
repoNameList.eachWithIndex { it, i ->
if (it.equals("master")){
repoNameList[i]= "master:selected"
}
}
return repoNameList - pipeline 脚本
效果：
* **post：后置操作**
Jenkinsfile (Declarative Pipeline)
pipeline {
agent any
stages {
stage('Hello') {
steps {
sh 'ls'
}
post {
always {
echo '步骤Hello体里的post操作'
}
}
}
}
// post部分可以同时包含多种条件块。
post {
always {
echo '永远都会执行'
}
success {
echo '本次构建成功时执行'
}
unstable {
echo '构建状态为不稳定时执行。'
}
failure {
echo '本次构建失败时执行'
}
changed {
echo '只要本次构建状态与上一次构建状态不同就执行。'
}
fixed {
echo '上一次构建状态为失败或不稳定，当前完成状态为成功时执行。'
}
regression {
echo '上一次构建状态为成功，当前构建状态为失败、不稳定或中止时执行。'
}
aborted {
echo '当前执行结果是中止状态时（一般为人为中止）执行。'
}
cleanup {
echo '清理条件块。不论当前完成状态是什么，在其他所有条件块执行完成后都执行'}
}
}
效果：
#### 案例：
* 执行自动化测试脚本并通过飞书卡片消息发送执行结果到飞书群：
先上效果图：
*图中用例执行结果需在代码中获取pytest执行结果统计*
需在最外层conftest中加以下代码：from _pytest import terminal
def pytest_terminal_summary(terminalreporter):
"""收集测试结果，注：跟xdist插件不兼容"""
print(terminalreporter.stats)
total = terminalreporter._numcollected - len(terminalreporter.stats.get('deselected', [])) # 收集总数-未选中用例数
passed = len([i for i in terminalreporter.stats.get('passed', []) if i.when == 'call'])
failed = len([i for i in terminalreporter.stats.get('failed', []) if i.when == 'call'])
error = len([i for i in terminalreporter.stats.get('error', []) if i.when != 'teardown'])
skipped = len([i for i in terminalreporter.stats.get('skipped', []) if i.when != 'teardown'])
pass_rate = round(passed / (total - skipped) * 100, 2)
# terminalreporter.\_sessionstarttime 会话开始时间
duration = time.time() - terminalreporter._sessionstarttime
# 将结果写入到一个文件中，后续再读取出来
result_path = os.path.join(PROJECT_DIR, "reports/result.json")
with open(result_path, "w")as fp:
fp.write(
str({"total": total, "passed": passed, "failed": failed, "error": error, "skipped": skipped,
"pass\_rate": pass_rate, "duration": round(duration, 3)}))
如何发送卡片消息到飞书群(企微/钉钉同理)：
1. 创建群机器人
**保存好这个webhook接收地址！**
2. 官网查找[开发者文档](https://blog.csdn.net/biggbang)，定制消息样式。
本文使用消息卡片，自己按需使用
3. 编写发送消息脚本
!/bin/env python3
-- coding: utf-8 --
desc：封装Jenkins构建结果发送飞书通知
import requests
import sys
class LarkBotReq(object):
机器人webhook接收地址
API_ENDPOINT = "https://open.feishu.cn/open-apis/bot/v2/hook/{}"
def __init__(self, bot_id=None, current_result=None, result_space=None):
获取用例执行结果
with open(result_space, 'r') as fp:
result = fp.read()
result = eval(result)
self.BUILD_MSG_TMPL = '''*构建结果：*部署环境: {env}
当前版本: {target}
构建结果: {current_result}
构建发起: {build_user}
持续时间: {duration}
构建日志: 点击查看详情
提交信息: 点击查看详情
'''
self.RUN_MSG_TMPL = f'''\n --------------\n本次执行结果：
总用例数: {result["total"]}
通过用例: {result['passed']}
失败用例: {result["failed"]}
跳过用例: {result['skipped']}
异常用例：{result['error']}
通过率（%）: {result['pass_rate']}
执行耗时：{result['duration']}s
'''
self.COMMIT_MSG_TMPL = '''\n --------------\n历史更新记录：
{git_commit_msg}
'''
self.url = self.API_ENDPOINT.format(bot_id)
self.body = None
self.msg_tmpl = self.BUILD_MSG_TMPL + self.COMMIT_MSG_TMPL + self.RUN_MSG_TMPLdef post_json(self):
Send http request with json body
headers = {
"Content-Type": "application/json; charset=utf-8"
}
res = requests.post(self.url, json=self.body, headers=headers)
return res.textdef formatter(self, kwargs):
self.body = {
"msg_type": "interactive",
"card": {
"config": {
"wide_screen_mode": True
},
"header": {
"title": {
"tag": "plain_text",
"content": "{project} 项目构建信息".format(kwargs)
},
控制卡片颜色
"template": "green" if result.get('pass\_rate') >= 100 else "red"
},
"elements": [
{
"tag": "markdown", # 使用markdown格式
"content": self.msg_tmpl.format(**kwargs),
}
]
}
}if name == '__main__':
bot = LarkBotReq(bot_id=sys.argv[1], current_result=sys.argv[2], result_space=sys.argv[12])
bot.formatter(
project=sys.argv[3],
env=sys.argv[4],
target=sys.argv[5],
current_result=sys.argv[2],
build_user=sys.argv[6],
duration=sys.argv[7],
build_url=sys.argv[8],
git_url=sys.argv[9],
env_commit=sys.argv[10],
git_commit_msg=sys.argv[11]
)
print(bot.post_json())
"""
调用方：
/usr/local/python3/bin/python3 send_msg_2_lark.py '你的机器人id' ' success' 'project_name' 'env' 'master' '{buildUser}' '{持续时间}' '{build_url}' '{git_url}' '{commit_id}' '{commit_msg}' 'reports/result.json'
"""
4. 把上面这个发送消息脚本放到jenkins服务器的某个目录里，如：/data/deploy/notify
5. 配置jenkins
* 配置上文提到的获取git分支/tag groovy脚本
* 编写pipeline脚本
下文中引用的credentialsId[配置credentialsId参阅往期>](https://blog.csdn.net/biggbang)
下文中的测试报告插件使用的是pytest-html，请自行去插件管理下载HTML Publisher；如果用的allure替换掉下文的pipeline中publishHTML配置。
下文中所引用的env.xx变量是Jenkins提供的全局变量，请自行查阅；你的job地址后拼接/pipeline-syntax/globals即可查阅。
或：
 
def project_url = 'ssh://git@{your host}/xx/project_name.git' // 用于拉代码
def GIT_URL = 'https://{your host}/xx/project_name' // 用于发送的消息卡片跳转
def credentialsId = 'your credentialsId' // git登陆凭证id
pipeline {
agent any
parameters {
choice(name: 'ENV', choices: ['dev','sit','prod'], description: '环境')
string(name: 'PROJECT', defaultValue: 'xx_project', description: '项目')
}
triggers {
cron('H 9 * * *')
}
stages {
stage('Checkout') {
steps {
script {
// 使用echo 输出变量需用双引号，单引号会当成字符串
echo "${target}"
checkout([$class: 'GitSCM', branches: [[name: '$target']], doGenerateSubmoduleConfigurations: false, extensions: [[$class: 'CleanBeforeCheckout']], submoduleCfg: [], userRemoteConfigs: [[credentialsId: '${credentialsId}', url: '$project_url']]])
echo 'Checkout'
// 此处路径地址需要和前文中conftest中配置的一致，发送消息的脚本需读取；workspace相当于你的项目根路径
result_json_path="${env.WORKSPACE}"+"/"+"reports/result.json"
echo "${result_json_path}"
}
}
}
stage('Set Build Name And Description'){
steps{
script{
currentBuild.displayName = "#${BUILD_NUMBER} *** ${PROJECT} ** ${ENV} * ${target}"
currentBuild.description = "本次构建信息:${BUILD_NUMBER} 项目名：${PROJECT} 环境：${ENV} 分支名：${target}"
}
}
}
stage('get_GIT_COMMIT_MSG') {
steps {
script {
env.GIT_COMMIT_MSG = sh (script: 'git log -1 --pretty=%B ${GIT_COMMIT}', returnStdout: true).trim()
env.COMMT= sh( returnStdout: true, script: 'git log --oneline -1 | awk \'{print \$1}\'')
echo "${env.GIT_COMMIT_MSG}"
}
}
}
stage('get_BUILD_USER') {
steps {
script{
wrap([$class: 'BuildUser']) {
BUILD_USER = "${env.BUILD_USER}"
}
if ("${BUILD_USER}"!="null"){
echo "${BUILD_USER}"
}else{
BUILD_USER = "定时器"
}
}
}
}
stage('执行全部项目检查脚本 ') {
steps {
// 执行测试脚本
sh '''
/usr/bin/python3 -m pytest tests --html=reports/report.html --self-contained-html
'''
}
}
}
post {
always {
//jenkins插件中的pytest-html报告插件，如果用allure的自行替换即可
publishHTML (target:[
allowMissing: false,
alwaysLinkToLastBuild: true,
keepAll: true,
reportDir: './reports/',
reportFiles: 'report.html',
reportName: 'HTML Report',
reportTitles: ''
]
)
script{
// 执行前面的发送消息脚本，机器人id请自行替换。
sh """
cd /data/deploy/notify
/usr/local/python3/bin/python3 send_msg_2_lark_for_test.py '机器人id' '${currentBuild.currentResult}' \
'${PROJECT}' '${ENV}' '${target}' '${BUILD_USER}' '${currentBuild.durationString}' '${env.BUILD_URL}' '${GIT_URL}' '${env.COMMT}' '${env.GIT_COMMIT_MSG}' '${result_json_path}'
"""
}
}
}
}
```- 执行job
转载请注明：xuhss » Jenkins系列之pipeline语法介绍与案例