1088 lines
30 KiB
Markdown
1088 lines
30 KiB
Markdown
# jianda-proxy CLI Tool Packaging — Implementation Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** Transform the single-file `lfs-proxy.py` script into a distributable CLI tool with `uvx`/`uv tool install` support, INI-based config, and cross-platform background daemon management.
|
|
|
|
**Architecture:** Four focused modules — `config.py` (INI config CRUD), `proxy.py` (HTTP proxy server, refactored from `lfs-proxy.py`), `daemon.py` (PID file + cross-platform process lifecycle), `cli.py` (argparse entry point). Zero external dependencies. `hatchling` build backend.
|
|
|
|
**Tech Stack:** Python 3.8+, `configparser`, `argparse`, `logging`, `http.server`, `ctypes` (Windows process checks)
|
|
|
|
---
|
|
|
|
## File Map
|
|
|
|
| File | Responsibility |
|
|
|------|---------------|
|
|
| `pyproject.toml` | Package metadata, entry point, build config |
|
|
| `src/jianda_proxy/__init__.py` | Package marker, version |
|
|
| `src/jianda_proxy/config.py` | Config directory resolution, INI read/write, defaults |
|
|
| `src/jianda_proxy/proxy.py` | HTTP proxy server (from `lfs-proxy.py`), accepts config dict |
|
|
| `src/jianda_proxy/daemon.py` | PID file CRUD, cross-platform process check/kill, daemonize |
|
|
| `src/jianda_proxy/cli.py` | argparse subcommands, entry point `main()` |
|
|
| `tests/test_config.py` | Tests for config module |
|
|
| `tests/test_daemon.py` | Tests for daemon module |
|
|
| `tests/test_proxy.py` | Tests for proxy module |
|
|
| `tests/test_cli.py` | Integration tests for CLI |
|
|
|
|
---
|
|
|
|
### Task 1: Project Scaffolding
|
|
|
|
**Files:**
|
|
- Create: `pyproject.toml`
|
|
- Create: `src/jianda_proxy/__init__.py`
|
|
|
|
- [ ] **Step 1: Create `pyproject.toml`**
|
|
|
|
```toml
|
|
[build-system]
|
|
requires = ["hatchling"]
|
|
build-backend = "hatchling.build"
|
|
|
|
[project]
|
|
name = "jianda-proxy"
|
|
version = "0.1.0"
|
|
description = "HTTP proxy for Gitea LFS — rewrites Host header to bypass ICP domain block"
|
|
requires-python = ">=3.8"
|
|
dependencies = []
|
|
|
|
[project.scripts]
|
|
jianda-proxy = "jianda_proxy.cli:main"
|
|
|
|
[tool.hatch.build.targets.wheel]
|
|
packages = ["src/jianda_proxy"]
|
|
```
|
|
|
|
- [ ] **Step 2: Create `src/jianda_proxy/__init__.py`**
|
|
|
|
```python
|
|
__version__ = "0.1.0"
|
|
```
|
|
|
|
- [ ] **Step 3: Verify `uv` can see the package**
|
|
|
|
Run: `uv pip install -e .` (from project root, in a venv)
|
|
Expected: Package installs successfully
|
|
|
|
- [ ] **Step 4: Verify entry point works (will fail, that's OK)**
|
|
|
|
Run: `jianda-proxy` or `python -m jianda_proxy`
|
|
Expected: `ModuleNotFoundError: No module named 'jianda_proxy.cli'` (confirms entry point wiring)
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add pyproject.toml src/jianda_proxy/__init__.py
|
|
git commit -m "feat: add project scaffolding with pyproject.toml and src layout"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 2: Config Module
|
|
|
|
**Files:**
|
|
- Create: `src/jianda_proxy/config.py`
|
|
- Create: `tests/test_config.py`
|
|
|
|
- [ ] **Step 1: Write tests for config module**
|
|
|
|
```python
|
|
import os
|
|
import tempfile
|
|
import pytest
|
|
from configparser import ConfigParser
|
|
from jianda_proxy.config import (
|
|
get_config_dir,
|
|
get_config_path,
|
|
get_pid_path,
|
|
get_log_path,
|
|
load_config,
|
|
set_config_value,
|
|
get_config_value,
|
|
list_config,
|
|
DEFAULTS,
|
|
)
|
|
|
|
FIXTURES = {
|
|
"remote_domain": "git.zz.com",
|
|
"upstream_host": "62.234.191.215",
|
|
"upstream_port": "3000",
|
|
"listen_port": "13000",
|
|
"listen_host": "127.0.0.1",
|
|
}
|
|
|
|
|
|
class TestConfigDir:
|
|
def test_returns_path_with_app_name(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HOME", str(tmp_path))
|
|
result = get_config_dir()
|
|
assert "jianda-proxy" in result
|
|
|
|
def test_windows_uses_appdata(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("APPDATA", str(tmp_path), raising=False)
|
|
monkeypatch.delenv("HOME", raising=False)
|
|
monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
|
|
result = get_config_dir()
|
|
assert "jianda-proxy" in result
|
|
|
|
|
|
class TestConfigCRUD:
|
|
def test_load_creates_default_config(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
config = load_config()
|
|
for key, val in FIXTURES.items():
|
|
assert config.get("proxy", key) == val
|
|
|
|
def test_load_creates_config_file(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
load_config()
|
|
assert (tmp_path / "config.ini").exists()
|
|
|
|
def test_set_and_get_value(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
set_config_value("listen_port", "8080")
|
|
assert get_config_value("listen_port") == "8080"
|
|
|
|
def test_set_unknown_key_raises(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
with pytest.raises(ValueError, match="Unknown config key"):
|
|
set_config_value("nonexistent_key", "value")
|
|
|
|
def test_list_config(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
config = load_config()
|
|
result = list_config()
|
|
assert isinstance(result, dict)
|
|
for key, val in FIXTURES.items():
|
|
assert result[key] == val
|
|
|
|
|
|
class TestPaths:
|
|
def test_config_path(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
assert get_config_path().endswith("config.ini")
|
|
|
|
def test_pid_path(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
assert get_pid_path().endswith("daemon.pid")
|
|
|
|
def test_log_path(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("jianda_proxy.config.get_config_dir", lambda: str(tmp_path))
|
|
assert get_log_path().endswith("proxy.log")
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify they fail**
|
|
|
|
Run: `pytest tests/test_config.py -v`
|
|
Expected: FAIL with `ModuleNotFoundError`
|
|
|
|
- [ ] **Step 3: Write config module implementation**
|
|
|
|
```python
|
|
import os
|
|
import configparser
|
|
from pathlib import Path
|
|
|
|
SECTION = "proxy"
|
|
DEFAULTS = {
|
|
"remote_domain": "git.zz.com",
|
|
"upstream_host": "62.234.191.215",
|
|
"upstream_port": "3000",
|
|
"listen_port": "13000",
|
|
"listen_host": "127.0.0.1",
|
|
}
|
|
|
|
|
|
def get_config_dir():
|
|
if os.name == "nt":
|
|
base = os.environ.get("APPDATA", os.path.expanduser("~"))
|
|
else:
|
|
base = os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config"))
|
|
return os.path.join(base, "jianda-proxy")
|
|
|
|
|
|
def _ensure_dir(path):
|
|
os.makedirs(path, exist_ok=True)
|
|
|
|
|
|
def get_config_path():
|
|
return os.path.join(get_config_dir(), "config.ini")
|
|
|
|
|
|
def get_pid_path():
|
|
return os.path.join(get_config_dir(), "daemon.pid")
|
|
|
|
|
|
def get_log_path():
|
|
return os.path.join(get_config_dir(), "proxy.log")
|
|
|
|
|
|
def load_config():
|
|
path = get_config_path()
|
|
_ensure_dir(get_config_dir())
|
|
parser = configparser.ConfigParser()
|
|
if not os.path.exists(path):
|
|
parser[SECTION] = dict(DEFAULTS)
|
|
with open(path, "w") as f:
|
|
parser.write(f)
|
|
else:
|
|
parser.read(path)
|
|
if SECTION not in parser:
|
|
parser[SECTION] = dict(DEFAULTS)
|
|
return parser
|
|
|
|
|
|
def set_config_value(key, value):
|
|
if key not in DEFAULTS:
|
|
raise ValueError(f"Unknown config key: {key}. Valid keys: {', '.join(DEFAULTS)}")
|
|
parser = load_config()
|
|
parser[SECTION][key] = value
|
|
with open(get_config_path(), "w") as f:
|
|
parser.write(f)
|
|
|
|
|
|
def get_config_value(key):
|
|
parser = load_config()
|
|
return parser.get(SECTION, key)
|
|
|
|
|
|
def list_config():
|
|
parser = load_config()
|
|
return dict(parser[SECTION])
|
|
```
|
|
|
|
- [ ] **Step 4: Run tests to verify they pass**
|
|
|
|
Run: `pytest tests/test_config.py -v`
|
|
Expected: All PASS
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add src/jianda_proxy/config.py tests/test_config.py
|
|
git commit -m "feat: add config module with INI-based configuration management"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 3: Proxy Module
|
|
|
|
**Files:**
|
|
- Create: `src/jianda_proxy/proxy.py`
|
|
- Create: `tests/test_proxy.py`
|
|
|
|
- [ ] **Step 1: Write tests for proxy module**
|
|
|
|
```python
|
|
import threading
|
|
import http.client
|
|
import time
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from jianda_proxy.proxy import create_server
|
|
|
|
|
|
class TestCreateServer:
|
|
def test_creates_server_with_config(self):
|
|
config = {
|
|
"remote_domain": "git.example.com",
|
|
"upstream_host": "1.2.3.4",
|
|
"upstream_port": "9000",
|
|
"listen_host": "127.0.0.1",
|
|
"listen_port": "19999",
|
|
}
|
|
server = create_server(config)
|
|
assert server is not None
|
|
server.server_close()
|
|
|
|
def test_binds_to_correct_port(self):
|
|
config = {
|
|
"remote_domain": "git.example.com",
|
|
"upstream_host": "1.2.3.4",
|
|
"upstream_port": "9000",
|
|
"listen_host": "127.0.0.1",
|
|
"listen_port": "19998",
|
|
}
|
|
server = create_server(config)
|
|
assert server.server_address[1] == 19998
|
|
server.server_close()
|
|
|
|
|
|
class TestProxyHandler:
|
|
def test_forward_proxy_mode(self):
|
|
"""When request URL has a hostname matching remote_domain, rewrite to upstream IP."""
|
|
from jianda_proxy.proxy import ProxyHandler
|
|
|
|
config = {
|
|
"remote_domain": "git.example.com",
|
|
"upstream_host": "1.2.3.4",
|
|
"upstream_port": "9000",
|
|
"listen_host": "127.0.0.1",
|
|
"listen_port": "19997",
|
|
}
|
|
server = create_server(config)
|
|
handler = ProxyHandler
|
|
|
|
# The handler should have the correct config set as class attributes
|
|
assert hasattr(handler, "REMOTE_DOMAIN")
|
|
server.server_close()
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify they fail**
|
|
|
|
Run: `pytest tests/test_proxy.py -v`
|
|
Expected: FAIL with `ModuleNotFoundError`
|
|
|
|
- [ ] **Step 3: Write proxy module — extract and refactor from `lfs-proxy.py`**
|
|
|
|
Read `lfs-proxy.py` and refactor into `src/jianda_proxy/proxy.py`:
|
|
|
|
```python
|
|
import http.client
|
|
import http.server
|
|
import logging
|
|
from urllib.parse import urlparse
|
|
|
|
logger = logging.getLogger("jianda_proxy")
|
|
|
|
|
|
class ProxyHandler(http.server.BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
# Set by create_server() before serving
|
|
REMOTE_DOMAIN = ""
|
|
UPSTREAM_IP = ""
|
|
UPSTREAM_PORT = 0
|
|
|
|
def _proxy_request(self, target_host, target_port, target_path, body=None):
|
|
body_len = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(body_len) if body_len else None
|
|
|
|
conn = http.client.HTTPConnection(target_host, target_port, timeout=120)
|
|
try:
|
|
headers = {}
|
|
skip = {"host", "transfer-encoding", "connection"}
|
|
for k in self.headers:
|
|
if k.lower() not in skip:
|
|
headers[k] = self.headers[k]
|
|
headers["Host"] = f"{ProxyHandler.UPSTREAM_IP}:{ProxyHandler.UPSTREAM_PORT}"
|
|
if body_len:
|
|
headers["Content-Length"] = str(body_len)
|
|
|
|
conn.request(self.command, target_path, body=body, headers=headers)
|
|
resp = conn.getresponse()
|
|
resp_body = resp.read()
|
|
|
|
self.send_response(resp.status)
|
|
sent_cl = False
|
|
for k, v in resp.getheaders():
|
|
kl = k.lower()
|
|
if kl in ("transfer-encoding", "connection"):
|
|
continue
|
|
if kl == "content-length":
|
|
self.send_header(k, len(resp_body))
|
|
sent_cl = True
|
|
continue
|
|
self.send_header(k, v)
|
|
if not sent_cl:
|
|
self.send_header("Content-Length", len(resp_body))
|
|
self.end_headers()
|
|
self.wfile.write(resp_body)
|
|
finally:
|
|
conn.close()
|
|
|
|
def _do(self):
|
|
raw_path = self.path
|
|
parsed = urlparse(raw_path)
|
|
|
|
if (
|
|
parsed.hostname
|
|
and parsed.hostname != "127.0.0.1"
|
|
and parsed.hostname != "localhost"
|
|
):
|
|
if parsed.hostname == ProxyHandler.REMOTE_DOMAIN:
|
|
target_host = ProxyHandler.UPSTREAM_IP
|
|
else:
|
|
target_host = parsed.hostname
|
|
target_port = parsed.port or 80
|
|
target_path = parsed.path + ("?" + parsed.query if parsed.query else "")
|
|
self._proxy_request(target_host, target_port, target_path)
|
|
else:
|
|
self._proxy_request(
|
|
ProxyHandler.UPSTREAM_IP, ProxyHandler.UPSTREAM_PORT, raw_path
|
|
)
|
|
|
|
do_GET = do_POST = do_PUT = do_DELETE = do_HEAD = do_PATCH = _do
|
|
|
|
def log_message(self, fmt, *args):
|
|
logger.info("%s - %s", self.address_string(), fmt % args)
|
|
|
|
|
|
def create_server(config):
|
|
"""Create (but don't start) the proxy server.
|
|
|
|
Args:
|
|
config: dict with keys remote_domain, upstream_host, upstream_port,
|
|
listen_host, listen_port (all strings)
|
|
|
|
Returns:
|
|
ThreadingHTTPServer instance
|
|
"""
|
|
ProxyHandler.REMOTE_DOMAIN = config["remote_domain"]
|
|
ProxyHandler.UPSTREAM_IP = config["upstream_host"]
|
|
ProxyHandler.UPSTREAM_PORT = int(config["upstream_port"])
|
|
|
|
listen_host = config.get("listen_host", "127.0.0.1")
|
|
listen_port = int(config["listen_port"])
|
|
|
|
server = http.server.ThreadingHTTPServer((listen_host, listen_port), ProxyHandler)
|
|
logger.info(
|
|
"Proxy: %s:%s -> %s:%s",
|
|
listen_host,
|
|
listen_port,
|
|
ProxyHandler.UPSTREAM_IP,
|
|
ProxyHandler.UPSTREAM_PORT,
|
|
)
|
|
return server
|
|
```
|
|
|
|
- [ ] **Step 4: Run tests to verify they pass**
|
|
|
|
Run: `pytest tests/test_proxy.py -v`
|
|
Expected: All PASS
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add src/jianda_proxy/proxy.py tests/test_proxy.py
|
|
git commit -m "feat: extract proxy server from lfs-proxy.py into module"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 4: Daemon Module
|
|
|
|
**Files:**
|
|
- Create: `src/jianda_proxy/daemon.py`
|
|
- Create: `tests/test_daemon.py`
|
|
|
|
- [ ] **Step 1: Write tests for daemon module**
|
|
|
|
```python
|
|
import os
|
|
import tempfile
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from jianda_proxy.daemon import (
|
|
read_pid,
|
|
write_pid,
|
|
remove_pid,
|
|
is_process_alive,
|
|
is_running,
|
|
stop_daemon,
|
|
get_status,
|
|
)
|
|
|
|
|
|
class TestPidFile:
|
|
def test_write_and_read_pid(self, tmp_path, monkeypatch):
|
|
pid_file = str(tmp_path / "test.pid")
|
|
monkeypatch.setattr("jianda_proxy.daemon.get_pid_path", lambda: pid_file)
|
|
write_pid(12345)
|
|
assert read_pid() == 12345
|
|
|
|
def test_read_missing_pid_returns_none(self, tmp_path, monkeypatch):
|
|
pid_file = str(tmp_path / "nonexistent.pid")
|
|
monkeypatch.setattr("jianda_proxy.daemon.get_pid_path", lambda: pid_file)
|
|
assert read_pid() is None
|
|
|
|
def test_remove_pid(self, tmp_path, monkeypatch):
|
|
pid_file = str(tmp_path / "test.pid")
|
|
monkeypatch.setattr("jianda_proxy.daemon.get_pid_path", lambda: pid_file)
|
|
write_pid(12345)
|
|
remove_pid()
|
|
assert read_pid() is None
|
|
|
|
|
|
class TestProcessCheck:
|
|
@patch("os.kill")
|
|
def test_alive_on_unix(self, mock_kill):
|
|
mock_kill.return_value = None
|
|
assert is_process_alive(12345) is True
|
|
|
|
@patch("os.kill", side_effect=ProcessLookupError)
|
|
def test_dead_on_unix(self, mock_kill):
|
|
assert is_process_alive(12345) is False
|
|
|
|
@patch("os.kill", side_effect=PermissionError)
|
|
def test_permission_means_alive(self, mock_kill):
|
|
# Can't send signal but process exists
|
|
assert is_process_alive(12345) is True
|
|
|
|
|
|
class TestIsRunning:
|
|
@patch("jianda_proxy.daemon.is_process_alive", return_value=True)
|
|
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
|
|
def test_running_when_pid_and_alive(self, mock_read, mock_alive):
|
|
assert is_running() is True
|
|
|
|
@patch("jianda_proxy.daemon.is_process_alive", return_value=False)
|
|
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
|
|
def test_not_running_when_pid_dead(self, mock_read, mock_alive):
|
|
assert is_running() is False
|
|
|
|
@patch("jianda_proxy.daemon.read_pid", return_value=None)
|
|
def test_not_running_when_no_pid(self, mock_read):
|
|
assert is_running() is False
|
|
|
|
|
|
class TestStopDaemon:
|
|
@patch("jianda_proxy.daemon.is_process_alive", return_value=True)
|
|
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
|
|
@patch("time.sleep")
|
|
@patch("os.kill")
|
|
def test_stop_sends_sigterm(self, mock_kill, mock_sleep, mock_read, mock_alive):
|
|
# First call: alive, second call: dead
|
|
mock_alive.side_effect = [True, False]
|
|
stop_daemon()
|
|
mock_kill.assert_called_once_with(12345, 15) # SIGTERM=15
|
|
|
|
@patch("jianda_proxy.daemon.remove_pid")
|
|
@patch("jianda_proxy.daemon.is_process_alive", return_value=False)
|
|
@patch("jianda_proxy.daemon.read_pid", return_value=None)
|
|
def test_stop_noop_when_not_running(self, mock_read, mock_alive, mock_remove):
|
|
result = stop_daemon()
|
|
assert result == "not_running"
|
|
|
|
|
|
class TestGetStatus:
|
|
@patch("jianda_proxy.daemon.is_process_alive", return_value=True)
|
|
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
|
|
def test_status_running(self, mock_read, mock_alive):
|
|
status = get_status()
|
|
assert status["running"] is True
|
|
assert status["pid"] == 12345
|
|
|
|
@patch("jianda_proxy.daemon.read_pid", return_value=None)
|
|
def test_status_not_running(self, mock_read):
|
|
status = get_status()
|
|
assert status["running"] is False
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify they fail**
|
|
|
|
Run: `pytest tests/test_daemon.py -v`
|
|
Expected: FAIL with `ModuleNotFoundError`
|
|
|
|
- [ ] **Step 3: Write daemon module implementation**
|
|
|
|
```python
|
|
import os
|
|
import sys
|
|
import time
|
|
import signal
|
|
from jianda_proxy.config import get_pid_path
|
|
|
|
|
|
def read_pid():
|
|
path = get_pid_path()
|
|
if not os.path.exists(path):
|
|
return None
|
|
try:
|
|
with open(path) as f:
|
|
return int(f.read().strip())
|
|
except (ValueError, OSError):
|
|
return None
|
|
|
|
|
|
def write_pid(pid):
|
|
path = get_pid_path()
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w") as f:
|
|
f.write(str(pid))
|
|
|
|
|
|
def remove_pid():
|
|
path = get_pid_path()
|
|
if os.path.exists(path):
|
|
os.remove(path)
|
|
|
|
|
|
def is_process_alive(pid):
|
|
if os.name == "nt":
|
|
import ctypes
|
|
import ctypes.wintypes
|
|
|
|
kernel32 = ctypes.windll.kernel32
|
|
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
|
|
STILL_ACTIVE = 259
|
|
handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
|
|
if not handle:
|
|
return False
|
|
try:
|
|
exit_code = ctypes.wintypes.DWORD()
|
|
kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code))
|
|
return exit_code.value == STILL_ACTIVE
|
|
finally:
|
|
kernel32.CloseHandle(handle)
|
|
else:
|
|
try:
|
|
os.kill(pid, 0)
|
|
return True
|
|
except ProcessLookupError:
|
|
return False
|
|
except PermissionError:
|
|
return True
|
|
|
|
|
|
def is_running():
|
|
pid = read_pid()
|
|
if pid is None:
|
|
return False
|
|
return is_process_alive(pid)
|
|
|
|
|
|
def stop_daemon():
|
|
pid = read_pid()
|
|
if pid is None:
|
|
return "not_running"
|
|
|
|
if not is_process_alive(pid):
|
|
remove_pid()
|
|
return "not_running"
|
|
|
|
if os.name == "nt":
|
|
os.system(f"taskkill /PID {pid} /T")
|
|
else:
|
|
try:
|
|
os.kill(pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
remove_pid()
|
|
return "stopped"
|
|
|
|
# Wait up to 3 seconds for graceful shutdown
|
|
for _ in range(6):
|
|
time.sleep(0.5)
|
|
if not is_process_alive(pid):
|
|
remove_pid()
|
|
return "stopped"
|
|
|
|
# Force kill
|
|
if os.name == "nt":
|
|
os.system(f"taskkill /F /PID {pid} /T")
|
|
else:
|
|
try:
|
|
os.kill(pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
remove_pid()
|
|
return "stopped"
|
|
|
|
|
|
def get_status():
|
|
pid = read_pid()
|
|
if pid is None or not is_process_alive(pid):
|
|
return {"running": False, "pid": None}
|
|
return {"running": True, "pid": pid}
|
|
|
|
|
|
def daemonize():
|
|
"""Fork/detach process for background operation."""
|
|
if os.name == "nt":
|
|
# Windows: relaunch self with detached flag
|
|
import subprocess
|
|
|
|
creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS
|
|
subprocess.Popen(
|
|
[sys.executable] + sys.argv + ["--_daemon_child"],
|
|
creationflags=creation_flags,
|
|
close_fds=True,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
else:
|
|
# Unix: double fork
|
|
pid = os.fork()
|
|
if pid > 0:
|
|
# Parent exits
|
|
os._exit(0)
|
|
os.setsid()
|
|
pid = os.fork()
|
|
if pid > 0:
|
|
os._exit(0)
|
|
```
|
|
|
|
- [ ] **Step 4: Run tests to verify they pass**
|
|
|
|
Run: `pytest tests/test_daemon.py -v`
|
|
Expected: All PASS
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add src/jianda_proxy/daemon.py tests/test_daemon.py
|
|
git commit -m "feat: add daemon module with cross-platform process lifecycle"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 5: CLI Entry Point
|
|
|
|
**Files:**
|
|
- Create: `src/jianda_proxy/cli.py`
|
|
- Create: `tests/test_cli.py`
|
|
|
|
- [ ] **Step 1: Write tests for CLI subcommands**
|
|
|
|
```python
|
|
import os
|
|
import tempfile
|
|
import subprocess
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from jianda_proxy.cli import build_parser
|
|
|
|
|
|
class TestParser:
|
|
def test_start_subcommand_exists(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["start"])
|
|
assert args.command == "start"
|
|
assert args.foreground is False
|
|
|
|
def test_start_foreground_flag(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["start", "--foreground"])
|
|
assert args.foreground is True
|
|
|
|
def test_stop_subcommand(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["stop"])
|
|
assert args.command == "stop"
|
|
|
|
def test_status_subcommand(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["status"])
|
|
assert args.command == "status"
|
|
|
|
def test_config_set(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["config", "set", "listen_port", "8080"])
|
|
assert args.config_command == "set"
|
|
assert args.key == "listen_port"
|
|
assert args.value == "8080"
|
|
|
|
def test_config_get(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["config", "get", "listen_port"])
|
|
assert args.config_command == "get"
|
|
assert args.key == "listen_port"
|
|
|
|
def test_config_list(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["config", "list"])
|
|
assert args.config_command == "list"
|
|
|
|
def test_config_path(self):
|
|
parser = build_parser()
|
|
args = parser.parse_args(["config", "path"])
|
|
assert args.config_command == "path"
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify they fail**
|
|
|
|
Run: `pytest tests/test_cli.py -v`
|
|
Expected: FAIL with `ModuleNotFoundError`
|
|
|
|
- [ ] **Step 3: Write CLI entry point**
|
|
|
|
```python
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
|
|
from jianda_proxy.config import (
|
|
load_config,
|
|
set_config_value,
|
|
get_config_value,
|
|
list_config,
|
|
get_config_path,
|
|
get_log_path,
|
|
)
|
|
from jianda_proxy.proxy import create_server
|
|
from jianda_proxy.daemon import (
|
|
write_pid,
|
|
is_running,
|
|
stop_daemon,
|
|
get_status,
|
|
daemonize,
|
|
)
|
|
|
|
|
|
def build_parser():
|
|
parser = argparse.ArgumentParser(prog="jianda-proxy", description="HTTP proxy for Gitea LFS")
|
|
sub = parser.add_subparsers(dest="command")
|
|
|
|
# start
|
|
p_start = sub.add_parser("start", help="Start the proxy daemon")
|
|
p_start.add_argument("--foreground", action="store_true", help="Run in foreground (no daemon)")
|
|
p_start.add_argument("--port", type=str, default=None, help="Override listen port")
|
|
|
|
# stop
|
|
sub.add_parser("stop", help="Stop the proxy daemon")
|
|
|
|
# status
|
|
sub.add_parser("status", help="Show proxy daemon status")
|
|
|
|
# config
|
|
p_config = sub.add_parser("config", help="Manage configuration")
|
|
config_sub = p_config.add_subparsers(dest="config_command")
|
|
|
|
p_set = config_sub.add_parser("set", help="Set a config value")
|
|
p_set.add_argument("key", help="Config key")
|
|
p_set.add_argument("value", help="Config value")
|
|
|
|
p_get = config_sub.add_parser("get", help="Get a config value")
|
|
p_get.add_argument("key", help="Config key")
|
|
|
|
config_sub.add_parser("list", help="List all config values")
|
|
config_sub.add_parser("path", help="Show config file path")
|
|
|
|
return parser
|
|
|
|
|
|
def cmd_start(args):
|
|
if is_running():
|
|
print("Error: jianda-proxy is already running.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
config = load_config()
|
|
config_dict = dict(config["proxy"])
|
|
|
|
if args.port:
|
|
config_dict["listen_port"] = args.port
|
|
|
|
if args.foreground:
|
|
_run_foreground(config_dict)
|
|
else:
|
|
_run_daemonized(config_dict)
|
|
|
|
|
|
def _run_foreground(config_dict):
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(name)s] %(levelname)s - %(message)s",
|
|
)
|
|
server = create_server(config_dict)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
server.server_close()
|
|
|
|
|
|
def _run_daemonized(config_dict):
|
|
from jianda_proxy.daemon import daemonize
|
|
|
|
# Set up file logging before daemonizing
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
log_path = get_log_path()
|
|
handler = RotatingFileHandler(log_path, maxBytes=5 * 1024 * 1024, backupCount=3)
|
|
handler.setFormatter(
|
|
logging.Formatter("%(asctime)s [%(name)s] %(levelname)s - %(message)s")
|
|
)
|
|
root = logging.getLogger()
|
|
root.setLevel(logging.INFO)
|
|
root.addHandler(handler)
|
|
|
|
daemonize()
|
|
|
|
# This runs in the child process
|
|
pid = os.getpid()
|
|
write_pid(pid)
|
|
|
|
server = create_server(config_dict)
|
|
try:
|
|
server.serve_forever()
|
|
finally:
|
|
server.server_close()
|
|
from jianda_proxy.daemon import remove_pid
|
|
remove_pid()
|
|
|
|
|
|
def cmd_stop(_args):
|
|
result = stop_daemon()
|
|
if result == "not_running":
|
|
print("jianda-proxy is not running.", file=sys.stderr)
|
|
sys.exit(1)
|
|
else:
|
|
print("jianda-proxy stopped.")
|
|
|
|
|
|
def cmd_status(_args):
|
|
status = get_status()
|
|
if status["running"]:
|
|
print(f"jianda-proxy is running (PID: {status['pid']})")
|
|
else:
|
|
print("jianda-proxy is not running.")
|
|
|
|
|
|
def cmd_config(args):
|
|
if args.config_command == "set":
|
|
set_config_value(args.key, args.value)
|
|
print(f"{args.key} = {args.value}")
|
|
elif args.config_command == "get":
|
|
print(get_config_value(args.key))
|
|
elif args.config_command == "list":
|
|
for k, v in list_config().items():
|
|
print(f"{k} = {v}")
|
|
elif args.config_command == "path":
|
|
print(get_config_path())
|
|
else:
|
|
print("Unknown config command. Use set/get/list/path.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def main():
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
commands = {
|
|
"start": cmd_start,
|
|
"stop": cmd_stop,
|
|
"status": cmd_status,
|
|
"config": cmd_config,
|
|
}
|
|
fn = commands.get(args.command)
|
|
if fn:
|
|
fn(args)
|
|
else:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
```
|
|
|
|
Note: `cli.py` needs `import os` at the top. Make sure it's included.
|
|
|
|
- [ ] **Step 4: Run CLI parser tests**
|
|
|
|
Run: `pytest tests/test_cli.py -v`
|
|
Expected: All PASS
|
|
|
|
- [ ] **Step 5: Manual smoke test — foreground mode**
|
|
|
|
Run: `jianda-proxy start --foreground` in one terminal, verify it starts and logs to terminal. Ctrl+C to stop.
|
|
|
|
- [ ] **Step 6: Manual smoke test — config commands**
|
|
|
|
Run:
|
|
```bash
|
|
jianda-proxy config list
|
|
jianda-proxy config path
|
|
jianda-proxy config set listen_port 14000
|
|
jianda-proxy config get listen_port
|
|
jianda-proxy config set listen_port 13000 # restore
|
|
```
|
|
|
|
- [ ] **Step 7: Manual smoke test — daemon lifecycle**
|
|
|
|
Run:
|
|
```bash
|
|
jianda-proxy start
|
|
jianda-proxy status
|
|
jianda-proxy stop
|
|
jianda-proxy status # should say not running
|
|
```
|
|
|
|
- [ ] **Step 8: Commit**
|
|
|
|
```bash
|
|
git add src/jianda_proxy/cli.py tests/test_cli.py
|
|
git commit -m "feat: add CLI entry point with start/stop/status/config subcommands"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 6: Integration Verification & Cleanup
|
|
|
|
**Files:**
|
|
- Modify: `README.md`
|
|
|
|
- [ ] **Step 1: Verify `uv tool install .` works**
|
|
|
|
Run: `uv tool install . --force`
|
|
Expected: `jianda-proxy` command available globally
|
|
|
|
- [ ] **Step 2: Verify `python -m jianda_proxy` works**
|
|
|
|
Run: `python -m jianda_proxy --help`
|
|
Expected: Help text displayed
|
|
|
|
- [ ] **Step 3: Run full test suite**
|
|
|
|
Run: `pytest tests/ -v`
|
|
Expected: All PASS
|
|
|
|
- [ ] **Step 4: Update README.md**
|
|
|
|
Replace README content with updated usage instructions reflecting the new CLI interface (`start`, `stop`, `status`, `config` subcommands, installation instructions).
|
|
|
|
- [ ] **Step 5: Remove old `lfs-proxy.py`**
|
|
|
|
```bash
|
|
git rm lfs-proxy.py
|
|
```
|
|
|
|
- [ ] **Step 6: Final commit**
|
|
|
|
```bash
|
|
git add README.md
|
|
git commit -m "docs: update README for CLI tool usage, remove old script"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 7: Verify and Finish
|
|
|
|
- [ ] **Step 1: Run all tests one final time**
|
|
|
|
Run: `pytest tests/ -v`
|
|
Expected: All PASS
|
|
|
|
- [ ] **Step 2: Full lifecycle manual test**
|
|
|
|
Run the complete workflow:
|
|
```bash
|
|
jianda-proxy config list
|
|
jianda-proxy start
|
|
jianda-proxy status
|
|
# Verify proxy works: curl through it
|
|
jianda-proxy stop
|
|
jianda-proxy status
|
|
```
|
|
|
|
- [ ] **Step 3: Verify `uv tool` uninstall and reinstall**
|
|
|
|
```bash
|
|
uv tool uninstall jianda-proxy
|
|
uv tool install .
|
|
jianda-proxy --help
|
|
uv tool uninstall jianda-proxy
|
|
```
|
|
|
|
- [ ] **Step 4: Run verification-before-completion skill**
|
|
|
|
Invoke the `superpowers:verification-before-completion` skill to confirm all requirements are met.
|