Initial release: CPCHECK cloud port detection tool

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>

tests/test_detector.py

@@ -0,0 +1,119 @@
+"""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()