fix: fix test_get_ip_from_env and send_ip
Some checks failed
Test CI / test speed (push) Failing after 1m8s
Some checks failed
Test CI / test speed (push) Failing after 1m8s
This commit is contained in:
parent
04d44afac0
commit
2308b01cee
@ -51,8 +51,7 @@ def send_ip():
|
|||||||
|
|
||||||
|
|
||||||
# 用环境变量获取本机ip
|
# 用环境变量获取本机ip
|
||||||
def get_local_ip():
|
def get_local_ip() -> str | None:
|
||||||
global ip
|
|
||||||
ip = os.environ.get("HOST_IP")
|
ip = os.environ.get("HOST_IP")
|
||||||
if not ip: # 如果环境变量中没有IP
|
if not ip: # 如果环境变量中没有IP
|
||||||
try:
|
try:
|
||||||
@ -61,12 +60,16 @@ def get_local_ip():
|
|||||||
s.connect(("8.8.8.8", 80)) # 通过连接Google DNS获取IP
|
s.connect(("8.8.8.8", 80)) # 通过连接Google DNS获取IP
|
||||||
ip = str(s.getsockname()[0])
|
ip = str(s.getsockname()[0])
|
||||||
s.close()
|
s.close()
|
||||||
|
return ip
|
||||||
except IndexError:
|
except IndexError:
|
||||||
raise ValueError("Unable to get IP")
|
raise ValueError("Unable to get IP")
|
||||||
|
else:
|
||||||
|
return ip
|
||||||
|
|
||||||
|
|
||||||
def init():
|
def init():
|
||||||
get_local_ip()
|
global ip
|
||||||
|
ip = get_local_ip()
|
||||||
send_ip()
|
send_ip()
|
||||||
asyncio.create_task(send_heartbeat_internal())
|
asyncio.create_task(send_heartbeat_internal())
|
||||||
print("Finish init")
|
print("Finish init")
|
||||||
|
@ -21,7 +21,6 @@ from node import (
|
|||||||
init,
|
init,
|
||||||
clear,
|
clear,
|
||||||
send_user_des_message,
|
send_user_des_message,
|
||||||
ip,
|
|
||||||
id,
|
id,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -34,26 +33,23 @@ server_address = "http://60.204.236.38:8000/server"
|
|||||||
|
|
||||||
class TestGetLocalIP(unittest.TestCase):
|
class TestGetLocalIP(unittest.TestCase):
|
||||||
|
|
||||||
os.environ["HOST_IP"] = "60.204.193.58" # 模拟设置 HOST_IP 环境变量
|
|
||||||
|
|
||||||
def test_get_ip_from_env(self):
|
def test_get_ip_from_env(self):
|
||||||
global ip
|
os.environ["HOST_IP"] = "60.204.193.58" # 模拟设置 HOST_IP 环境变量
|
||||||
# 调用被测函数
|
ip = get_local_ip()
|
||||||
get_local_ip()
|
|
||||||
# 检查函数是否正确获取到 HOST_IP
|
# 检查函数是否正确获取到 HOST_IP
|
||||||
self.assertEqual(ip, "60.204.193.58")
|
self.assertEqual(ip, "60.204.193.58")
|
||||||
|
|
||||||
@patch("socket.socket") # Mock socket 连接行为
|
@patch("socket.socket") # Mock socket 连接行为
|
||||||
@patch.dict("os.environ", {}) # 模拟没有 HOST_IP 环境变量
|
|
||||||
def test_get_ip_from_socket(self, mock_socket):
|
def test_get_ip_from_socket(self, mock_socket):
|
||||||
global ip
|
os.environ.pop("HOST_IP", None)
|
||||||
|
|
||||||
# 模拟 socket 返回的 IP 地址
|
# 模拟 socket 返回的 IP 地址
|
||||||
mock_socket_instance = MagicMock()
|
mock_socket_instance = MagicMock()
|
||||||
mock_socket.return_value = mock_socket_instance
|
mock_socket.return_value = mock_socket_instance
|
||||||
mock_socket_instance.getsockname.return_value = ("110.41.155.96", 0)
|
mock_socket_instance.getsockname.return_value = ("110.41.155.96", 0)
|
||||||
|
|
||||||
# 调用被测函数
|
# 调用被测函数
|
||||||
get_local_ip()
|
ip = get_local_ip()
|
||||||
|
|
||||||
# 确认 socket 被调用过
|
# 确认 socket 被调用过
|
||||||
mock_socket_instance.connect.assert_called_with(("8.8.8.8", 80))
|
mock_socket_instance.connect.assert_called_with(("8.8.8.8", 80))
|
||||||
|
Loading…
x
Reference in New Issue
Block a user