每日一题_Python.利用yield生成器实现协程下的tps透明传输CS测试

标签：每日一题   python   

具体需求:
1. 模拟Device首先发送注册包注册到TPS服务器,然后Client发送私有数据包到TPS,测试Device可以接收到私有数据包则返回成功或标志位(0, ‘‘)或(1, ‘errormsg‘)
2. 由于此插件是用于自己写的插件式监控系统,所以入口函数名必须和文件名保持一致,这里暂定为server_tps_status.py

实现思路:

具体代码:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.163.com/
# Purpose:
# Install:
#
"""
# 说明: 导入公共模块
import json
import socket


def device_send_recv(device_socket, tps_host, tps_port, private_data, extra_data):
    send_data = {
        "TransProxy": {
            "Body": {
                ‘AuthCode‘: ‘0000000000000000‘,
                ‘SerialNumber‘: ‘0000000000000000‘
            },
            ‘Header‘: {
                ‘CSeq‘: ‘1‘,
                ‘MessageType‘: ‘MSG_TRANSPROXY_REGISTER_REQ‘,
                ‘TerminalType‘: ‘Camera‘,
                ‘Version‘: ‘1.0‘
            }
        }
    }
    send_data = json.dumps(send_data)
    send_data = ("POST /TransProxy HTTP/1.1\r\n"
                 "Host:%s\r\n"
                 "Port:%s\r\n"
                 "Connection:keep-alive\r\n"
                 "Content-Length:%s\r\n\r\n"
                 "%s\r\n"
                 ) % (tps_host, tps_port, len(send_data), send_data)
    device_socket.send(send_data)
    while True:
        cur_buffer = device_socket.recv(1024)
        if ‘TransProxy‘ in cur_buffer:
            yield (0, ‘‘)
        elif private_data in cur_buffer:
            yield (0, ‘‘)
            break
        else:
            yield (1, cur_buffer)
            break


def client_send_recv(device_socket, tps_host, tps_port, private_data, extra_data):
    send_data = ("POST /PrivateData HTTP/1.1\r\n"
                 "Host:%s\r\n"
                 "Port:%s\r\n"
                 "AuthCode:%s\r\n"
                 "SrcUuid:%s\r\n"
                 "DestUuid:%s\r\n"
                 "Connection:keep-alive\r\n"
                 "Content-Length:%s\r\n\r\n"
                 "%s\r\n"
                 ) % (tps_host, tps_port, extra_data[0], extra_data[1],
                      extra_data[2], len(private_data), private_data)
    device_socket.send(send_data)


def service_tps_status():
    private_data = ‘xmdevops‘
    exec_ret_dicts = {‘status‘: 0, ‘target‘: 0, ‘errors‘: ‘‘}
    tps_host, tps_port = ‘127.0.0.1‘, 6604
    extra_data = (
        ‘02899574bd6e899060‘,
        ‘1111111111111111‘,
        ‘0000000000000000‘,
    )
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    device_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        client_socket.connect((tps_host, tps_port))
        device_socket.connect((tps_host, tps_port))
    except socket.error, e:
        exec_ret_dicts.update({
            ‘status‘: 1,
            ‘target‘: 1,
            ‘errors‘: e.strerror,
        })
        return exec_ret_dicts

    device = device_send_recv(device_socket, tps_host, tps_port, private_data, extra_data)
    device.next()
    client_send_recv(client_socket, tps_host, tps_port, private_data, extra_data)
    while True:
        try:
            result = device.next()
        except StopIteration, e:
            break
    exec_ret_dicts.update({
        ‘status‘: result[0],
        ‘target‘: result[0],
        ‘errors‘: result[1],
    })
    return exec_ret_dicts

if __name__ == ‘__main__‘:
    print(service_tps_status())

本文出自 “ζ自动化运维开发之路ζ” 博客，请务必保留此出处http://xmdevops.blog.51cto.com/11144840/1846887

每日一题_Python.利用yield生成器实现协程下的tps透明传输CS测试

标签：每日一题   python   

原文地址：http://xmdevops.blog.51cto.com/11144840/1846887