テストソケットインタフェース

テストソケットインタフェース
インタフェーステストでは、一般的なhttpインタフェースのほかに、socketインタフェースがよく見られます.今日はPythonが持参したsocketライブラリでsocketインタフェーステストを行う方法について説明します.
以前に構築したテストフレームワークで実現します.具体的には、ゼロから自動化されたテストフレームワークシリーズを構築することができます.
1.ソケットコネクタsocketは とも呼ばれ、アプリケーションのアドレスであり、ネットワーク通信を実現する鍵であると理解できる.IPでホストを見つけることができ、ホストのポートでホスト上のアプリケーションを見つけることができます.
これにより、2つのアプリケーション間の通信をsocketで行うことができます.具体的な実装は、一端で常に傍受されているserverを実装し、他端で要求を送信し、応答を取得することである.serverは、異なるリクエストを異なる処理して返します.これがsocketインタフェースです.
次に、socketのインタフェースを実装し、テストします.
2.socket serverインタフェースを実現する
testの下にmockフォルダを作成し、mock_を作成します.socket_server.pyファイル:

"""
socket server  mock。
    ，add sub
  ：
{
    "action": "add",
    "params": {"a": 1, "b": 2}
}
  ：
{
    "action": "add",
    "result": 3
}
"""
import socket
import json


def add(a, b):
    return a + b


ip_port = ('127.0.0.1', 8080)
server = socket.socket()
server.bind(ip_port)
server.listen(5)

while True:
    conn, addr = server.accept()
    data = conn.recv(1024)
    try:
        ddata = json.loads(data)
        action = ddata.get('action')
        params = ddata.get('params')
        if action == 'add':
            res = add(**params)
            conn.send(b'{"action": "add", "result": %d}' % res)
        else:
            conn.send(b'{"code":-2, "message":"Wrong Action"}')
    except (AttributeError, ValueError):
        conn.send(b'{"code":-1, "message":"Data Error"}')
    finally:
        conn.close()

インタフェースaddを持つ簡単なsocket serverを実現しました.インタフェースドキュメントを開発するには、次のようにします.
インタフェースタイプ:socketインタフェースアドレス:127.0.0.1ポート:8080インタフェース名:加算action name:addパラメータ:
名前
を選択します.
必要かどうか
a
int
はい
b
int
はい
参照:
名前
を選択します.
意味
result
int
結果
入力例:

 {
   "action": "add",
    "params": {"a": 1, "b": 2}
 }

例を示します.

{
    "action": "add",
    "result": 3
}

error code:
code
message
-1
Data Error
-2
Wrong Action
インタフェースドキュメントを入手したら、どうやってテストすればいいですか?
3.socketインタフェースのテスト
まず、毎回たくさん書かないように、socketインタフェーステストの共通の方法をカプセル化する共通のclientクラスが必要です.Client.pyにTCPClientを追加

class TCPClient(object):
    """    socket  """
    def __init__(self, domain, port, timeout=30, max_receive=102400):
        self.domain = domain
        self.port = port
        self.connected = 0  #      1
        self.max_receive = max_receive
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._sock.settimeout(timeout)

    def connect(self):
        """    IP、  """
        if not self.connected:
            try:
                self._sock.connect((self.domain, self.port))
            except socket.error as e:
                logger.exception(e)
            else:
                self.connected = 1
                logger.debug('TCPClient connect to {0}:{1} success.'.format(self.domain, self.port))

    def send(self, data, dtype='str', suffix=''):
        """       send_string，     ，   ，   None"""
        if dtype == 'json':
            send_string = json.dumps(data) + suffix
        else:
            send_string = data + suffix
        self.connect()
        if self.connected:
            try:
                self._sock.send(send_string.encode())
                logger.debug('TCPClient Send {0}'.format(send_string))
            except socket.error as e:
                logger.exception(e)

            try:
                rec = self._sock.recv(self.max_receive).decode()
                if suffix:
                    rec = rec[:-len(suffix)]
                logger.debug('TCPClient received {0}'.format(rec))
                return rec
            except socket.error as e:
                logger.exception(e)

    def close(self):
        """    """
        if self.connected:
            self._sock.close()
            logger.debug('TCPClient closed.')

次にconfig.ymlにsocketのインタフェースの基本構成を追加します.

ip: 127.0.0.1
port: 8080

その後インターネットでテストtestを作成しますsocket.py

import unittest
from utils.client import TCPClient
from utils.config import Config
from utils.extractor import JMESPathExtractor

je = JMESPathExtractor()


class TestAdd(unittest.TestCase):

    def setUp(self):
        c = Config()
        ip = c.get('ip')
        port = c.get('port')
        self.client = TCPClient(ip, port)

    def tearDown(self):
        self.client.close()

    def test_add(self):
        data = {
            'action': 'add',
            'params': {'a': 1, 'b': 2}
        }
        res = self.client.send(data, dtype='json')
        self.assertEqual(je.extract('result', res), 3)
        self.assertEqual(je.extract('action', res), 'add')

    def test_wrong_action(self):
        data = {
            'action': 'sub',
            'params': {'a': 1, 'b': 2}
        }
        res = self.client.send(data, dtype='json')
        self.assertEqual(je.extract('code', res), -2)
        self.assertEqual(je.extract('message', res), 'Wrong Action')

    def test_wrong_data(self):
        data = 'xxxxx'
        res = self.client.send(data)
        self.assertEqual(je.extract('code', res), -1)
        self.assertEqual(je.extract('message', res), 'Data Error')


if __name__ == '__main__':
    unittest.main(verbosity=2)

簡単なテスト用例が完成しましたので、実行してください(まずmock serverを走り出しますよ):

test_add (__main__.TestAdd) ... ok
test_wrong_action (__main__.TestAdd) ... ok
test_wrong_data (__main__.TestAdd) ... ok

----------------------------------------------------------------------
Ran 3 tests in 0.009s

OK

もちろん、インタフェースはこんなに簡単ではありません.例もこれだけではありません.ここでは栗を簡単に挙げるだけで、応用できます.どんなに複雑なsocketインタフェースでも同じです.
ここでは自分のmockのserverを使って、本当のインタフェースを開発して通じた後、config.ymlのipとportを変更して、直接テストを実行することができます.