diff --git a/install_gmssl.sh b/install_gmssl.sh index e942f26..9190b48 100644 --- a/install_gmssl.sh +++ b/install_gmssl.sh @@ -3,7 +3,7 @@ mkdir lib mkdir include -cp gmssl/include include +cp -r gmssl/include include mkdir gmssl/build cd gmssl/build || exit diff --git a/src/node_test.py b/src/node_test.py index e69de29..2b5cdaf 100644 --- a/src/node_test.py +++ b/src/node_test.py @@ -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() \ No newline at end of file diff --git a/src/node_test1.py b/src/node_test1.py deleted file mode 100644 index 66a7a47..0000000 --- a/src/node_test1.py +++ /dev/null @@ -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() diff --git a/src/node_test2.py b/src/node_test2.py deleted file mode 100644 index 7e01de8..0000000 --- a/src/node_test2.py +++ /dev/null @@ -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)即可看到测试代码返回的节点 diff --git a/src/node_test3.py b/src/node_test3.py deleted file mode 100644 index a92d921..0000000 --- a/src/node_test3.py +++ /dev/null @@ -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() diff --git a/src/node_test5.py b/src/node_test5.py index 3432bfe..124f970 100644 --- a/src/node_test5.py +++ b/src/node_test5.py @@ -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"} \ No newline at end of file + assert response.json() == {"status": "success"} + +if __name__ == '__main__': + unittest.main() \ No newline at end of file