dekun
32c8f4b156
Web-based TCP/UDP port checker with firewall/GFW diagnosis, PM2 deployment config, and Ubuntu one-click install script. Co-authored-by: Cursor <cursoragent@cursor.com>
120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
"""CPCHECK 单元测试"""
|
|
|
|
import socket
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.detector import (
|
|
CheckResult,
|
|
DiagnosisType,
|
|
PortStatus,
|
|
check_tcp,
|
|
diagnose,
|
|
resolve_host,
|
|
run_check,
|
|
)
|
|
|
|
|
|
class TestResolveHost(unittest.TestCase):
|
|
def test_resolve_localhost(self):
|
|
ip, err = resolve_host("127.0.0.1")
|
|
self.assertIsNone(err)
|
|
self.assertEqual(ip, "127.0.0.1")
|
|
|
|
def test_resolve_invalid(self):
|
|
ip, err = resolve_host("this-domain-does-not-exist-xyz123.invalid")
|
|
self.assertIsNone(ip)
|
|
self.assertIsNotNone(err)
|
|
|
|
|
|
class TestCheckTcp(unittest.TestCase):
|
|
def test_tcp_open(self):
|
|
with patch("socket.socket") as mock_sock_cls:
|
|
mock_sock = MagicMock()
|
|
mock_sock_cls.return_value = mock_sock
|
|
status, latency, detail = check_tcp("127.0.0.1", 80, timeout=1)
|
|
self.assertEqual(status, PortStatus.OPEN)
|
|
self.assertIsNotNone(latency)
|
|
|
|
def test_tcp_closed(self):
|
|
with patch("socket.socket") as mock_sock_cls:
|
|
mock_sock = MagicMock()
|
|
mock_sock.connect.side_effect = ConnectionRefusedError
|
|
mock_sock_cls.return_value = mock_sock
|
|
status, latency, detail = check_tcp("127.0.0.1", 59999, timeout=1)
|
|
self.assertEqual(status, PortStatus.CLOSED)
|
|
self.assertIsNone(latency)
|
|
|
|
def test_tcp_timeout(self):
|
|
with patch("socket.socket") as mock_sock_cls:
|
|
mock_sock = MagicMock()
|
|
mock_sock.connect.side_effect = socket.timeout
|
|
mock_sock_cls.return_value = mock_sock
|
|
status, latency, detail = check_tcp("127.0.0.1", 59998, timeout=1)
|
|
self.assertEqual(status, PortStatus.FILTERED)
|
|
|
|
|
|
class TestDiagnose(unittest.TestCase):
|
|
def test_port_open(self):
|
|
d, msg = diagnose(PortStatus.OPEN, "tcp", True, True)
|
|
self.assertEqual(d, DiagnosisType.PORT_OPEN)
|
|
|
|
def test_port_closed(self):
|
|
d, msg = diagnose(PortStatus.CLOSED, "tcp", True, True)
|
|
self.assertEqual(d, DiagnosisType.PORT_NOT_OPEN)
|
|
|
|
def test_filtered_reachable(self):
|
|
d, msg = diagnose(PortStatus.FILTERED, "tcp", True, True)
|
|
self.assertEqual(d, DiagnosisType.LIKELY_FIREWALL)
|
|
|
|
def test_filtered_unreachable(self):
|
|
d, msg = diagnose(PortStatus.FILTERED, "tcp", False, True)
|
|
self.assertEqual(d, DiagnosisType.LIKELY_BLOCKED)
|
|
|
|
def test_dns_failed(self):
|
|
d, msg = diagnose(PortStatus.UNKNOWN, "tcp", False, False)
|
|
self.assertEqual(d, DiagnosisType.DNS_FAILED)
|
|
|
|
|
|
class TestRunCheck(unittest.TestCase):
|
|
@patch("app.detector.ping_host", return_value=True)
|
|
@patch("app.detector.check_tcp")
|
|
@patch("app.detector.resolve_host", return_value=("93.184.216.34", None))
|
|
def test_run_check_success(self, mock_resolve, mock_tcp, mock_ping):
|
|
mock_tcp.return_value = (PortStatus.OPEN, 50.0, "TCP 连接成功")
|
|
result = run_check("example.com", 80, "tcp")
|
|
self.assertIsInstance(result, CheckResult)
|
|
self.assertTrue(result.dns_ok)
|
|
self.assertEqual(result.diagnosis, DiagnosisType.PORT_OPEN)
|
|
|
|
@patch("app.detector.resolve_host", return_value=(None, "Name resolution failed"))
|
|
def test_run_check_dns_fail(self, mock_resolve):
|
|
result = run_check("invalid.invalid", 80, "tcp")
|
|
self.assertFalse(result.dns_ok)
|
|
self.assertEqual(result.diagnosis, DiagnosisType.DNS_FAILED)
|
|
|
|
|
|
class TestAPIValidation(unittest.TestCase):
|
|
def test_check_request_validation(self):
|
|
try:
|
|
from pydantic import ValidationError
|
|
from app.main import CheckRequest
|
|
except ImportError:
|
|
self.skipTest("FastAPI/Pydantic not installed")
|
|
|
|
req = CheckRequest(host="example.com", port=443, protocol="tcp")
|
|
self.assertEqual(req.protocol, "tcp")
|
|
|
|
with self.assertRaises(ValidationError):
|
|
CheckRequest(host="example.com", port=443, protocol="icmp")
|
|
|
|
with self.assertRaises(ValidationError):
|
|
CheckRequest(host="example.com", port=0, protocol="tcp")
|
|
|
|
with self.assertRaises(ValidationError):
|
|
CheckRequest(host="", port=80, protocol="tcp")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|