sangge/tpre-python
2
1
Fork 1
Code Issues 6 Pull Requests Packages Projects 1 Activity

test: Add unit tests

Browse Source
This commit is contained in:
muzhi 2024-09-30 12:06:21 +08:00 committed by sangge-redmi
parent acbc9ecb1a
commit 05de02f2a5
6 changed files with 164 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
install_gmssl.sh
View File

@ -3,7 +3,7 @@
mkdir lib
mkdir include
cp gmssl/include include
cp -r gmssl/include include
mkdir gmssl/build
cd gmssl/build || exit

76
src/node_test.py
View File

@ -0,0 +1,76 @@
# 测试 node.py 中的函数
import os
import unittest
from unittest.mock import patch, MagicMock, Mock
import node
class TestGetLocalIP(unittest.TestCase):
@patch.dict('os.environ', {'HOST_IP': '60.204.193.58'}) # 模拟设置 HOST_IP 环境变量
def test_get_ip_from_env(self):
# 调用被测函数
node.get_local_ip()
# 检查函数是否正确获取到 HOST_IP
self.assertEqual(node.ip, '60.204.193.58')
@patch('socket.socket') # Mock socket 连接行为
@patch.dict('os.environ', {}) # 模拟没有 HOST_IP 环境变量
def test_get_ip_from_socket(self, mock_socket):
# 模拟 socket 返回的 IP 地址
mock_socket_instance = MagicMock()
mock_socket.return_value = mock_socket_instance
mock_socket_instance.getsockname.return_value = ('110.41.155.96', 0)
# 调用被测函数
node.get_local_ip()
# 确认 socket 被调用过
mock_socket_instance.connect.assert_called_with(('8.8.8.8', 80))
mock_socket_instance.close.assert_called_once()
# 检查是否通过 socket 获取到正确的 IP 地址
self.assertEqual(node.ip, '110.41.155.96')
class TestSendIP(unittest.TestCase):
@patch.dict(os.environ, {'HOST_IP': '60.204.193.58'}) # 设置环境变量 HOST_IP
@patch('requests.get') # Mock requests.get 调用
def test_send_ip(self, mock_get):
# 设置模拟返回的 HTTP 响应
mock_response = Mock()
mock_response.text = "node123" # 模拟返回的节点ID
mock_response.status_code = 200
mock_get.return_value = mock_response # 设置 requests.get() 的返回值为 mock_response
# 保存原始的全局 id 值
original_id = node.id
# 调用待测函数
node.send_ip()
# 确保 requests.get 被正确调用
expected_url = f"{node.server_address}/get_node?ip={node.ip}"
mock_get.assert_called_once_with(expected_url, timeout=3)
# 检查 id 是否被正确更新
self.assertIs(node.id, mock_response) # 检查 id 是否被修改
self.assertEqual(node.id.text, "node123") # 检查更新后的 id 是否与 mock_response.text 匹配
class TestNode(unittest.TestCase):
@patch('node.send_ip')
@patch('node.get_local_ip')
@patch('node.asyncio.create_task')
def test_init(self, mock_create_task, mock_get_local_ip, mock_send_ip):
# 调用 init 函数
node.init()
# 验证 get_local_ip 和 send_ip 被调用
mock_get_local_ip.assert_called_once()
mock_send_ip.assert_called_once()
# 确保 create_task 被调用来启动心跳包
mock_create_task.assert_called_once()
if __name__ == '__main__':
unittest.main()

35
src/node_test1.py
View File

@ -1,35 +0,0 @@
#测试 get_local_ip()函数
import unittest
from unittest.mock import patch, MagicMock
import node
class TestGetLocalIP(unittest.TestCase):
@patch.dict('os.environ', {'HOST_IP': '60.204.193.58'}) # 模拟设置 HOST_IP 环境变量
def test_get_ip_from_env(self):
# 调用被测函数
node.get_local_ip()
# 检查函数是否正确获取到 HOST_IP
self.assertEqual(node.ip, '60.204.193.58')
@patch('socket.socket') # Mock socket 连接行为
@patch.dict('os.environ', {}) # 模拟没有 HOST_IP 环境变量
def test_get_ip_from_socket(self, mock_socket):
# 模拟 socket 返回的 IP 地址
mock_socket_instance = MagicMock()
mock_socket.return_value = mock_socket_instance
mock_socket_instance.getsockname.return_value = ('110.41.155.96', 0)
# 调用被测函数
node.get_local_ip()
# 确认 socket 被调用过
mock_socket_instance.connect.assert_called_with(('8.8.8.8', 80))
mock_socket_instance.close.assert_called_once()
# 检查是否通过 socket 获取到正确的 IP 地址
self.assertEqual(node.ip, '110.41.155.96')
if __name__ == '__main__':
unittest.main()

34
src/node_test2.py
View File

@ -1,34 +0,0 @@
#测试send_ip()函数
import os
import unittest
from unittest.mock import patch, Mock
import node # 导入要测试的模块
class TestSendIP(unittest.TestCase):
@patch.dict(os.environ, {'HOST_IP': '60.204.193.58'}) # 设置环境变量 HOST_IP
@patch('requests.get') # Mock requests.get 调用
def test_send_ip(self, mock_get):
# 设置模拟返回的 HTTP 响应
mock_response = Mock()
mock_response.text = "node123" # 模拟返回的节点ID
mock_response.status_code = 200
mock_get.return_value = mock_response # 设置 requests.get() 的返回值为 mock_response
# 保存原始的全局 id 值
original_id = node.id
# 调用待测函数
node.send_ip()
# 确保 requests.get 被正确调用
expected_url = f"{node.server_address}/get_node?ip={node.ip}"
mock_get.assert_called_once_with(expected_url, timeout=3)
# 检查 id 是否被正确更新
self.assertIs(node.id, mock_response) # 检查 id 是否被修改
self.assertEqual(node.id.text, "node123") # 检查更新后的 id 是否与 mock_response.text 匹配
if __name__ == "__main__":
unittest.main()
#node.py中
#print("中心服务器返回节点ID为: ", id.text)即可看到测试代码返回的节点

23
src/node_test3.py
View File

@ -1,23 +0,0 @@
#测试init()函数
import unittest
from unittest.mock import patch, AsyncMock
import node
class TestNode(unittest.TestCase):
@patch('node.send_ip')
@patch('node.get_local_ip')
@patch('node.asyncio.create_task')
def test_init(self, mock_create_task, mock_get_local_ip, mock_send_ip):
# 调用 init 函数
node.init()
# 验证 get_local_ip 和 send_ip 被调用
mock_get_local_ip.assert_called_once()
mock_send_ip.assert_called_once()
# 确保 create_task 被调用来启动心跳包
mock_create_task.assert_called_once()
if __name__ == '__main__':
unittest.main()

99
src/node_test5.py
View File

@ -1,22 +1,92 @@
#node_test剩下部分(有问题)
import os
import unittest
import pytest
from unittest.mock import patch, MagicMock, Mock, AsyncMock
import requests
import asyncio
import httpx
import respx
import asyncio
from unittest.mock import patch, AsyncMock
from fastapi.testclient import TestClient
from node import app, send_heartbeat_internal, Req
from node import app, send_heartbeat_internal, Req, send_ip, get_local_ip, init, clear, send_user_des_message
client = TestClient(app)
server_address = "http://60.204.236.38:8000/server"
ip = "127.0.0.1"
ip = None # 初始化全局变量 ip
id = None # 初始化全局变量 id
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
class TestGetLocalIP(unittest.TestCase):
@patch.dict('os.environ', {'HOST_IP': '60.204.193.58'}) # 模拟设置 HOST_IP 环境变量
def test_get_ip_from_env(self):
global ip
# 调用被测函数
get_local_ip()
# 检查函数是否正确获取到 HOST_IP
self.assertEqual(ip, '60.204.193.58')
@patch('socket.socket') # Mock socket 连接行为
@patch.dict('os.environ', {}) # 模拟没有 HOST_IP 环境变量
def test_get_ip_from_socket(self, mock_socket):
global ip
# 模拟 socket 返回的 IP 地址
mock_socket_instance = MagicMock()
mock_socket.return_value = mock_socket_instance
mock_socket_instance.getsockname.return_value = ('110.41.155.96', 0)
# 调用被测函数
get_local_ip()
# 确认 socket 被调用过
mock_socket_instance.connect.assert_called_with(('8.8.8.8', 80))
mock_socket_instance.close.assert_called_once()
# 检查是否通过 socket 获取到正确的 IP 地址
self.assertEqual(ip, '110.41.155.96')
class TestSendIP(unittest.TestCase):
@patch.dict(os.environ, {'HOST_IP': '60.204.193.58'}) # 设置环境变量 HOST_IP
@respx.mock
def test_send_ip(self):
global ip, id
ip = '60.204.193.58'
mock_url = f"{server_address}/get_node?ip={ip}"
respx.get(mock_url).mock(return_value=httpx.Response(200, text="node123"))
# 调用待测函数
send_ip()
# 确保 requests.get 被正确调用
self.assertEqual(id, "node123") # 检查更新后的 id 是否与 mock_response.text 匹配
class TestNode(unittest.TestCase):
@patch('node.send_ip')
@patch('node.get_local_ip')
@patch('node.asyncio.create_task')
def test_init(self, mock_create_task, mock_get_local_ip, mock_send_ip):
# 调用 init 函数
init()
# 验证 get_local_ip 和 send_ip 被调用
mock_get_local_ip.assert_called_once()
mock_send_ip.assert_called_once()
# 确保 create_task 被调用来启动心跳包
mock_create_task.assert_called_once()
def test_clear(self):
# 调用 clear 函数
clear()
# 检查输出
self.assertTrue(True) # 这里只是为了确保函数被调用，没有实际逻辑需要测试
@pytest.mark.asyncio
@respx.mock
async def test_send_heartbeat_internal_success():
global ip
ip = '60.204.193.58'
# 模拟心跳请求
heartbeat_route = respx.get(f"{server_address}/heartbeat?ip={ip}").mock(
return_value=httpx.Response(200)
@ -40,6 +110,8 @@ async def test_send_heartbeat_internal_success():
@pytest.mark.asyncio
@respx.mock
async def test_send_heartbeat_internal_failure():
global ip
ip = '60.204.193.58'
# 模拟心跳请求以引发异常
heartbeat_route = respx.get(f"{server_address}/heartbeat?ip={ip}").mock(
side_effect=httpx.RequestError("Central server error")
@ -66,8 +138,8 @@ def test_user_src():
# 模拟 send_user_des_message 函数
with patch("node.send_user_des_message", new_callable=AsyncMock) as mock_send_user_des_message:
message = {
"source_ip": "127.0.0.1",
"dest_ip": "127.0.0.2",
"source_ip": "60.204.193.58",
"dest_ip": "60.204.193.59",
"capsule": (("x1", "y1"), ("x2", "y2"), 123),
"ct": 456,
"rk": ["rk1", "rk2"]
@ -79,11 +151,14 @@ def test_user_src():
def test_send_user_des_message():
with respx.mock:
dest_ip = "127.0.0.2"
dest_ip = "60.204.193.59"
re_message = (("a", "b", "c", "d"), 123)
respx.post(f"http://{dest_ip}:8002/receive_messages").mock(
return_value=httpx.Response(200, json={"status": "success"})
)
response = client.post(f"http://{dest_ip}:8002/receive_messages", json={"Tuple": re_message, "ip": "127.0.0.1"})
response = requests.post(f"http://{dest_ip}:8002/receive_messages", json={"Tuple": re_message, "ip": "60.204.193.58"})
assert response.status_code == 200
assert response.json() == {"status": "success"}
assert response.json() == {"status": "success"}
if __name__ == '__main__':
unittest.main()