重构测试套件并优化 CI 配置

.github/workflows/ci.yml

@@ -3,76 +3,75 @@ name: CI
 on: [push, pull_request]
 
 jobs:
-  C_test:
-    strategy:
-      matrix:
-        os: [ubuntu-latest]
+  # C_test:
+  #   strategy:
+  #     matrix:
+  #       os: [ubuntu-latest]
 
-    runs-on: ${{ matrix.os }}
-    timeout-minutes: 30
+  #   runs-on: ${{ matrix.os }}
+  #   timeout-minutes: 30
 
-    steps:
-    - name: Checkout code
-      uses: actions/checkout@v4
+  #   steps:
+  #   - name: Checkout code
+  #     uses: actions/checkout@v6.0.2
 
-    - name: Install C dependencies (Ubuntu)
-      if: matrix.os == 'ubuntu-latest'
-      run: |
-        sudo apt-get remove needrestart
-        sudo apt-get update
-        sudo apt-get install -y cmake
+  #   - name: Install C dependencies (Ubuntu)
+  #     if: matrix.os == 'ubuntu-latest'
+  #     run: |
+  #       sudo apt-get remove needrestart
+  #       sudo apt-get update
+  #       sudo apt-get install -y cmake
 
-    - name: Build C code
-      run: |
-        mkdir build
-        cd build
-        cmake ..
-        make
+  #   - name: Build C code
+  #     run: |
+  #       mkdir -p build
+  #       cd build
+  #       cmake ..
+  #       make
 
-    - name: Run C tests
-      run: |
-        cd build
-        make test
+  #   - name: Run C tests
+  #     run: |
+  #       cd build
+  #       ctest
 
-  Pylint:
-    strategy:
-      matrix:
-        os: [ubuntu-latest]
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+  Lint:
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v6.0.2
+    - name: Run Ruff linter on Python code
+      uses: astral-sh/ruff-action@v4.0.0
+      with:
+        args: "check"
+      continue-on-error: true
+    - name: Check code formatting
+      uses: astral-sh/ruff-action@v4.0.0
+      with:
+        args: "format --check"
+      continue-on-error: true
 
-    runs-on: ${{ matrix.os }}
-    timeout-minutes: 30
-
+  TypeCheck:
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
     steps:
-      - name: Checkout code
-        uses: actions/checkout@v4
+      - uses: actions/checkout@v6.0.2
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.1.0
 
-      - name: Set up Python
-        uses: actions/setup-python@v4
+      - name: Run ty
+        run: uvx ty check
+        continue-on-error: true
+  Test:
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v6.0.2
+      - uses: astral-sh/setup-uv@v8.1.0
         with:
           python-version: ${{ matrix.python-version }}
-
-      - name: Install linting tools
-        run: |
-          python -m pip install --upgrade pip
-          pip install pylint flake8 mypy black
-          python -c "import black, flake8, mypy, pylint; print('All imports OK')"
-
-      # - name: Check Python code formatting
-      #   run: |
-      #     black ./src/
-
-      - name: Run Python linter
-        run: |
-          flake8 ./src/
-        continue-on-error: true
-
-      - name: Run python lint score
-        run: |
-          pylint --max-line-length=150 ./src/
-        continue-on-error: true
-
-      - name: Run Python type checking
-        run: |
-          mypy ./src/
-        continue-on-error: true
+      - run: uv sync --group dev
+      - run: uv run pytest

.gitignore

@@ -12,10 +12,10 @@ _deps
 CMakeUserPresets.json
 .vscode/
 .vs/
-.test/build/
-.test/venv/
-.test/src/
-.src/__pycache__/
+test/build/
+test/venv/
+test/src/
+*/__pycache__/
 
 # CLion
 #  JetBrains specific template is maintained in a separate JetBrains.gitignore that can

pyproject.toml

@@ -0,0 +1,32 @@
+[project]
+name = "servspy"
+version = "0.1.0"
+description = "ServSpy is a lightweight tool used for monitoring server status and simply controlling servers, such as through the command lines or the desktops."
+requires-python = ">=3.10"
+dependencies = []
+
+[dependency-groups]
+dev = [
+    "pytest>=9.0.0",
+]
+
+[tool.ruff]
+target-version = "py310"
+line-length = 150
+exclude = ["build/", "dist/", ".git/", "__pycache__/"]
+
+[tool.ruff.lint]
+select = ["E", "F", "PL", "I"]
+
+[tool.ruff.format]
+quote-style = "double"
+indent-style = "space"
+
+[tool.ty.environment]
+python-version = "3.10"
+
+[tool.ty.src]
+exclude = ["test/"]
+
+[tool.pytest.ini_options]
+testpaths = ["test"]

src/command_control_extension_tcp.py

@@ -13,6 +13,42 @@
 client_instance=None
 command_counter={}
 command_counter_lock=threading.Lock()
+
+def _load_json_file(path):
+    try:
+        with open(path, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+            return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def _merge_log_dicts(base_data, extra_data):
+    merged = {}
+    for command, entries in base_data.items():
+        merged[command] = list(entries) if isinstance(entries, list) else [entries]
+    for command, entries in extra_data.items():
+        if command not in merged:
+            merged[command] = []
+        if isinstance(entries, list):
+            merged[command].extend(entries)
+        else:
+            merged[command].append(entries)
+    return merged
+
+
+def _merge_all_logs(log_dir, merged_filename='merged_logs.json'):
+    merged_data = {}
+    for filename in os.listdir(log_dir):
+        if not filename.endswith('.json') or filename == merged_filename:
+            continue
+        file_path = os.path.join(log_dir, filename)
+        if os.path.isfile(file_path):
+            merged_data = _merge_log_dicts(merged_data, _load_json_file(file_path))
+    with open(os.path.join(log_dir, merged_filename), 'w', encoding='utf-8') as f:
+        json.dump(merged_data, f, ensure_ascii=False, indent=2)
+
+
 def _setup_command():
     print("Setting up server command...")
     server_instance.register_command(
@@ -149,18 +185,26 @@ def _command_done_dealing_server(sock, addr, cmd):
     cmd_parts = shlex.split(cmd)
     log_filename = cmd_parts[1]
     log_path = cmd_parts[2]
-    log_dir=os.path.join(os.path.dirname(__file__), 'logs')
-    if os.path.exists(log_dir):
-        pass
-    else:
-        os.mkdir(log_dir)
-    received_log_file=os.path.join(
-        server_instance.file_transfer_dir, log_path)
+    log_dir = os.path.join(os.path.dirname(__file__), 'logs')
+    os.makedirs(log_dir, exist_ok=True)
+
+    received_log_file = os.path.join(
+        server_instance.file_transfer_dir,
+        os.path.basename(log_path))
+    destination_log_file = os.path.join(log_dir, log_filename)
     try:
-        shutil.move(
-            received_log_file, os.path.join(log_dir, log_filename))
+        if os.path.exists(destination_log_file):
+            existing_logs = _load_json_file(destination_log_file)
+            incoming_logs = _load_json_file(received_log_file)
+            merged_logs = _merge_log_dicts(existing_logs, incoming_logs)
+            with open(destination_log_file, 'w', encoding='utf-8') as f:
+                json.dump(merged_logs, f, ensure_ascii=False, indent=2)
+            os.remove(received_log_file)
+        else:
+            shutil.move(received_log_file, destination_log_file)
+        _merge_all_logs(log_dir)
         print("Command Done!")
-    except:
+    except Exception:
         traceback.print_exc()
         print("ErrorWhileMovingTheLogFile: moving log file failed.")
 def client_setup():

test/conftest.py

@@ -0,0 +1,64 @@
+import importlib.util
+import os
+import sys
+
+import pytest
+
+# Only add project root to sys.path if src package is not importable
+if importlib.util.find_spec("src") is None:
+    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import src.command_control_extension_tcp as ctl
+from src.connect_tcp import TCP_Client_Base, TCP_Server_Base
+
+SERVER_PORT = 65001
+CLIENT_PORT = 65000
+SERVER2_PORT = 65002
+CLIENT2_PORT = 65003
+
+
+@pytest.fixture
+def server(monkeypatch):
+    """Create a TCP_Server_Base and inject into command_control_extension_tcp.server_instance.
+
+    Use this fixture when a test only needs a server instance.
+    """
+    s = TCP_Server_Base(host="127.0.0.1", port=SERVER_PORT, is_extend_command=True)
+    monkeypatch.setattr(ctl, "server_instance", s)
+    try:
+        yield s
+    finally:
+        s.stop()
+
+
+@pytest.fixture
+def client(monkeypatch):
+    """Create a TCP_Client_Base and inject into command_control_extension_tcp.client_instance.
+
+    Use this fixture when a test only needs a client instance.
+    """
+    c = TCP_Client_Base(host="127.0.0.1", port=CLIENT_PORT, client_host="127.0.0.1", is_extend_command=True)
+    monkeypatch.setattr(ctl, "client_instance", c)
+    try:
+        yield c
+    finally:
+        c.close()
+
+
+@pytest.fixture
+def server_client(monkeypatch):
+    """Create both server and client and inject both module globals.
+
+    Use this when a test requires both sides present.
+    """
+    s = TCP_Server_Base(host="127.0.0.1", port=SERVER2_PORT, is_extend_command=True)
+    c = TCP_Client_Base(host="127.0.0.1", port=CLIENT2_PORT, client_host="127.0.0.1", is_extend_command=True)
+    monkeypatch.setattr(ctl, "server_instance", s)
+    monkeypatch.setattr(ctl, "client_instance", c)
+    try:
+        yield s, c
+    finally:
+        try:
+            c.close()
+        finally:
+            s.stop()

test/test_TCP_client.py

@@ -1,9 +1,8 @@
-import os
-import sys
-package_dictionary=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-if package_dictionary not in os.sys.path:
-    sys.path.insert(0, package_dictionary)
-from src import connect_tcp
-def test_TCP_client():
-    connect_tcp.TCP_Client_Base(host="127.0.0.1")
-test_TCP_client()
+from src.connect_tcp import TCP_Client_Base
+
+
+def test_tcp_client_init():
+    client = TCP_Client_Base(host="127.0.0.1", port=65003, client_host="127.0.0.1", is_extend_command=True)
+    assert client.host == "127.0.0.1"
+    assert client.port == 65003  # noqa: PLR2004
+    assert callable(client.connect)