如何用seldom救火,记录一次救援行动

2022-04-20 18:00:00

背景

最近正产环境接到不定期巡检,于是需要验证网站是否挂掉,保证网站持续正常运行。

需求其实很简单,但脚本就是要得急,且马上就要写好去巡检生产环境上各个系统。

怎么破?

喝杯☕️冷静一下,其实要做的事情,无外乎就以下

  • 在各大应用市场调用查询接口,看app是否还在

  • 调用目标系统地址,获取网页信息,判断指定关键字是否存在

  • 收集结果并调用钉钉api把结果发到群里通知

其实很容易就想到用主力用requests进行接口请求,收集结果,最后再调用钉钉机器人的API就行了。

如果有必要,整个流程还可以套在pytest里面。

甚至allure还能提供优美的报告。

但时间好像并不允许我先去手搓一两个基础轮子再把各个库组在一起。那有没有什么快速有效的方法能救火呢?

大概是在两年前了解到seldom这么一个框架(或者说库?),现在已经更新到了v2.8.0,功能已经相对完善了。于是我想着是不是可以尝试一次,相信seldom的魔法✨,优雅地实施一次救援呢。

记流水账

  • Target 1:在各大应用市场调用查询接口,看app是否还在

首先来到某应用市场,假设需要查找<微信>,通过搜索,直接拿到接口相关信息

class TestRequest(seldom.TestCase):

    def test_case1(self):
        payload = {
            "method": "internal.getTabDetail",
            "serviceType": 20,
            "reqPageNum": 1,
            "uri": "searchApp|微信",
            "maxResults": 25,
            "version": "10.0.0",
            "zone": None,
            "locale": "zh",
        }
        self.get("https://xxxx.xxxxx.dbankcloud.cn/uowap/index", params=payload)

seldom 提供便捷的断言,这里直接使用assertPath(),便可以对json数据进行断言

self.assertPath("layoutData[0].dataList[0].name", "微信")

第一个应用市场搞定

后续再去其他应用市场查看接口信息,跟着第一个例子,做成完整脚本:

class TestRequest(seldom.TestCase):

    def test_case1(self):
        payload = {
            "method": "internal.getTabDetail",
            "serviceType": 20,
            "reqPageNum": 1,
            "uri": "searchApp|微信",
            "maxResults": 25,
            "version": "10.0.0",
            "zone": None,
            "locale": "zh",
        }
        self.get("https://xxxx.xxxxx.dbankcloud.cn/uowap/index", params=payload)
        self.assertStatusCode(200)
        self.assertPath("layoutData[0].dataList[0].name", "微信")

    def test_case2(self):
        params = {
            "cmdid": "2654"
        }
        payload = {
            "req": {
                "commReq": {
                    "reqId": 946,
                    "businessId": "",
                    "userId": "",
                    "timestamp": round(time.time()),
                    "nonce": "n6akd78ir9o",
                    "signature": ""
                },
                "keyword": "微信",
                "pageNumberStack": "",
                "pageSize": 15
            }
        }
        self.post("https://xxxxxx.3g.qq.com/forward", params=params, json=payload)
        self.assertStatusCode(200)
        self.assertPath("data.resp.searchAppResult.appDetails[0].appName", "微信")

    def test_case3(self):
        params = {
            "src": "globalnav"
        }
        self.get("https://xxxxxxx.xpple.com.cn/cn/search/xxxxxxxE5%AE%B6", params=params)
        self.assertStatusCode(200)
        self.assertPath("data.resp.searchAppResult.appDetails[0].appName", "微信")
  • Target 2:调用目标系统地址,获取网页信息,判断指定关键字是否存在

首先访问并登录目标系统,直接被拦在门外,验证码校验是滑块,且安全防护等级很高。在有自知之明的前提下,准备采用载入cookie的方式进行页面访问。

seldom继承了requests的session,这边可以拿来直接使用,并载入cookie

class TestRequest(seldom.TestCase):

    def start(self):
        self.s = self.Session()
        pass

    def test_case1(self):
        self.s.cookies["cookie"] = "_trs_uv=l26x17id_4338_f8lg; zg_did=this is a cookie" #设置cookie
        self.s.get("http://cx-xxxxxxx.xxxxx.cq.gov.cn/", params={})

最后再补一个断言

assert "xxxx" in self.response

补全后续,做成完整脚本

import seldom


class TestRequest(seldom.TestCase):

    def start(self):
        self.s = self.Session()
        pass

    def test_case1(self):
        self.s.cookies["cookie"] = "_trs_uv=l26x17id_4338_f8lg; zg_did=this is a cookie"
        self.s.get("http://cx-xxxxx.xxxxxxw.cn/", params={})
        self.assertStatusCode(200)
        assert "重大新产品申报" in self.response

    def test_case2(self):
        self.s.cookies["cookie"] = "_trs_uv=l26x17id_4338_f8lg; zg_did=this is a cookie"
        self.s.get("https://xxx.xxxxx.cq.xx/hce/index.html", params={})
        self.assertStatusCode(200)
        assert "./resource/img/enter_logo.png" in self.response

    def test_case3(self):
        self.s.cookies["cookie"] = "_trs_uv=l26x17id_4338_f8lg; zg_did=this is a cookie"
        self.s.get("https://xx.xxw.cq.xox.cn/page/index.html", params={})
        self.assertStatusCode(200)
        assert "xxxx系统" in self.response
  • Target 3:收集结果并调用钉钉api把结果发到群里通知

得益于XTestRunner,seldom 天生提供发送钉钉群的能力

根据钉钉文档,依次填入access_token,key(选填),app_secret(选填),at_mobiles,is_at_all即可

import seldom
from seldom import DingTalk

# ...

if __name__ == '__main__':
    seldom.main()
    ding = DingTalk(
        access_token="690900b5ce6d5d10bb1218b8e64a4e2b55f96a6d116aaf50",
        key="xxxx",
        app_secret="xxxxx",
        at_mobiles=[13700000000, 13800000000],
        is_at_all=False,
    )
    ding.sender()

最后,一个做成的脚本可能是这样的:

import seldom
from seldom import DingTalk


class TestRequest(seldom.TestCase):

    def start(self):
        self.s = self.Session()
        pass

    def test_case1(self):
        self.s.cookies["cookie"] = "_trs_uv=l26x17id_4338_f8lg; zg_did=this is a cookie"
        self.s.get("http://cx-xxxxx.xxxxxxw.cn/", params={})
        self.assertStatusCode(200)
        assert "重大新产品申报" in self.response

    def test_case2(self):
        self.s.cookies["cookie"] = "_trs_uv=l26x17id_4338_f8lg; zg_did=this is a cookie"
        self.s.get("https://xxx.xxxxx.cq.xx/hce/index.html", params={})
        self.assertStatusCode(200)
        assert "./resource/img/enter_logo.png" in self.response

    def test_case3(self):
        self.s.cookies["cookie"] = "_trs_uv=l26x17id_4338_f8lg; zg_did=this is a cookie"
        self.s.get("https://xx.xxw.cq.xox.cn/page/index.html", params={})
        self.assertStatusCode(200)
        assert "xxxx系统" in self.response


if __name__ == '__main__':
    seldom.main()
    ding = DingTalk(
        access_token="690900b5ce6d5d10bb1218b8e64a4e2b55f96a6d116aaf50",
        key="xxxx",
        app_secret="xxxxx",
        at_mobiles=[13700000000, 13800000000],
        is_at_all=False,
    )
    ding.sender()

接近40行代码,达到了目的,一次舒适愉快的救火行动🏃🏻‍♀️