mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-01 18:20:16 +08:00
Compare commits
1 Commits
v4.26.3
...
codex/fix-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eea74cf909 |
@@ -1,3 +1,3 @@
|
||||
from .core.log import LogManager
|
||||
import logging
|
||||
|
||||
logger = LogManager.GetLogger(log_name="astrbot")
|
||||
logger = logging.getLogger("astrbot")
|
||||
|
||||
@@ -1,3 +1,32 @@
|
||||
from astrbot.core.config.default import VERSION
|
||||
import re
|
||||
from importlib.metadata import PackageNotFoundError
|
||||
from importlib.metadata import version as package_version
|
||||
from pathlib import Path
|
||||
|
||||
__version__ = VERSION
|
||||
try:
|
||||
import tomllib
|
||||
except ModuleNotFoundError:
|
||||
tomllib = None
|
||||
|
||||
try:
|
||||
__version__ = package_version("astrbot")
|
||||
except PackageNotFoundError:
|
||||
pyproject_path = Path(__file__).resolve().parents[2] / "pyproject.toml"
|
||||
try:
|
||||
if tomllib is None:
|
||||
match = re.search(
|
||||
r"(?m)^version\s*=\s*[\"']([^\"']+)[\"']",
|
||||
pyproject_path.read_text(encoding="utf-8"),
|
||||
)
|
||||
__version__ = match.group(1) if match else "0.0.0"
|
||||
else:
|
||||
with pyproject_path.open("rb") as f:
|
||||
__version__ = tomllib.load(f)["project"]["version"]
|
||||
except (FileNotFoundError, IndexError, KeyError, TypeError, ValueError):
|
||||
__version__ = "0.0.0"
|
||||
|
||||
match = re.match(r"^(\d+(?:\.\d+)*)(a|b|rc)(\d+)$", __version__)
|
||||
if match:
|
||||
release, prerelease, number = match.groups()
|
||||
prerelease = {"a": "alpha", "b": "beta", "rc": "rc"}[prerelease]
|
||||
__version__ = f"{release}-{prerelease}.{number}"
|
||||
|
||||
@@ -1,16 +1,11 @@
|
||||
import json
|
||||
import os
|
||||
import zoneinfo
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import click
|
||||
|
||||
from astrbot.core.utils.auth_password import (
|
||||
hash_dashboard_password,
|
||||
hash_md5_dashboard_password,
|
||||
validate_dashboard_password,
|
||||
)
|
||||
|
||||
from ..utils import check_astrbot_root, get_astrbot_root
|
||||
|
||||
|
||||
@@ -44,6 +39,8 @@ def _validate_dashboard_username(value: str) -> str:
|
||||
|
||||
def _validate_dashboard_password(value: str) -> str:
|
||||
"""Validate Dashboard password"""
|
||||
from astrbot.core.utils.auth_password import validate_dashboard_password
|
||||
|
||||
try:
|
||||
validate_dashboard_password(value)
|
||||
except ValueError as e:
|
||||
@@ -89,6 +86,7 @@ def _load_config() -> dict[str, Any]:
|
||||
raise click.ClickException(
|
||||
f"{root} is not a valid AstrBot root directory. Use 'astrbot init' to initialize",
|
||||
)
|
||||
os.environ["ASTRBOT_ROOT"] = str(root)
|
||||
|
||||
config_path = root / "data" / "cmd_config.json"
|
||||
if not config_path.exists():
|
||||
@@ -107,7 +105,8 @@ def _load_config() -> dict[str, Any]:
|
||||
|
||||
def _save_config(config: dict[str, Any]) -> None:
|
||||
"""Save config file"""
|
||||
config_path = get_astrbot_root() / "data" / "cmd_config.json"
|
||||
root = get_astrbot_root()
|
||||
config_path = root / "data" / "cmd_config.json"
|
||||
|
||||
config_path.write_text(
|
||||
json.dumps(config, ensure_ascii=False, indent=2),
|
||||
@@ -139,6 +138,11 @@ def _get_nested_item(obj: dict[str, Any], path: str) -> Any:
|
||||
|
||||
def _set_dashboard_password(config: dict[str, Any], raw_password: str) -> None:
|
||||
"""Set dashboard password hashes and clear password migration flags."""
|
||||
from astrbot.core.utils.auth_password import (
|
||||
hash_dashboard_password,
|
||||
hash_md5_dashboard_password,
|
||||
)
|
||||
|
||||
_set_nested_item(
|
||||
config,
|
||||
"dashboard.pbkdf2_password",
|
||||
|
||||
@@ -21,17 +21,16 @@ def _initialize_config_from_env(astrbot_root: Path) -> None:
|
||||
|
||||
|
||||
async def initialize_astrbot(astrbot_root: Path) -> None:
|
||||
"""Execute AstrBot initialization logic"""
|
||||
"""Execute AstrBot initialization logic.
|
||||
|
||||
Args:
|
||||
astrbot_root: Runtime root directory to initialize.
|
||||
"""
|
||||
dot_astrbot = astrbot_root / ".astrbot"
|
||||
|
||||
if not dot_astrbot.exists():
|
||||
if click.confirm(
|
||||
f"Install AstrBot to this directory? {astrbot_root}",
|
||||
default=True,
|
||||
abort=True,
|
||||
):
|
||||
dot_astrbot.touch()
|
||||
click.echo(f"Created {dot_astrbot}")
|
||||
dot_astrbot.touch()
|
||||
click.echo(f"Created {dot_astrbot}")
|
||||
|
||||
paths = {
|
||||
"data": astrbot_root / "data",
|
||||
@@ -41,8 +40,9 @@ async def initialize_astrbot(astrbot_root: Path) -> None:
|
||||
}
|
||||
|
||||
for name, path in paths.items():
|
||||
path_exists = path.exists()
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
click.echo(f"{'Created' if not path.exists() else 'Directory exists'}: {path}")
|
||||
click.echo(f"{'Directory exists' if path_exists else 'Created'}: {path}")
|
||||
|
||||
_initialize_config_from_env(astrbot_root)
|
||||
|
||||
@@ -53,7 +53,25 @@ async def initialize_astrbot(astrbot_root: Path) -> None:
|
||||
def init() -> None:
|
||||
"""Initialize AstrBot"""
|
||||
click.echo("Initializing AstrBot...")
|
||||
astrbot_root = get_astrbot_root()
|
||||
if os.environ.get("ASTRBOT_ROOT"):
|
||||
astrbot_root = get_astrbot_root()
|
||||
click.echo(f"Using ASTRBOT_ROOT: {astrbot_root}")
|
||||
else:
|
||||
user_root = (Path.home() / ".astrbot").resolve()
|
||||
current_root = Path.cwd().resolve()
|
||||
click.echo("Choose AstrBot runtime directory:")
|
||||
click.echo(f"1. {user_root} (recommended)")
|
||||
click.echo(f"2. Current directory: {current_root}")
|
||||
choice = click.prompt(
|
||||
"Select",
|
||||
type=click.Choice(["1", "2"]),
|
||||
default="1",
|
||||
show_choices=False,
|
||||
)
|
||||
astrbot_root = user_root if choice == "1" else current_root
|
||||
|
||||
astrbot_root.mkdir(parents=True, exist_ok=True)
|
||||
os.environ["ASTRBOT_ROOT"] = str(astrbot_root)
|
||||
lock_file = astrbot_root / "astrbot.lock"
|
||||
lock = FileLock(lock_file, timeout=5)
|
||||
|
||||
@@ -65,6 +83,8 @@ def init() -> None:
|
||||
raise click.ClickException(
|
||||
"Cannot acquire lock file. Please check if another instance is running"
|
||||
)
|
||||
except click.Abort:
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
raise click.ClickException(f"Initialization failed: {e!s}")
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
@@ -7,7 +8,14 @@ _BUNDLED_DIST = Path(__file__).parent.parent.parent / "dashboard" / "dist"
|
||||
|
||||
|
||||
def check_astrbot_root(path: str | Path) -> bool:
|
||||
"""Check if the path is an AstrBot root directory"""
|
||||
"""Check whether a path is an AstrBot root directory.
|
||||
|
||||
Args:
|
||||
path: Directory path to inspect.
|
||||
|
||||
Returns:
|
||||
Whether the directory contains the AstrBot root marker.
|
||||
"""
|
||||
if not isinstance(path, Path):
|
||||
path = Path(path)
|
||||
if not path.exists() or not path.is_dir():
|
||||
@@ -18,8 +26,24 @@ def check_astrbot_root(path: str | Path) -> bool:
|
||||
|
||||
|
||||
def get_astrbot_root() -> Path:
|
||||
"""Get the AstrBot root directory path"""
|
||||
return Path.cwd()
|
||||
"""Get the AstrBot root directory path.
|
||||
|
||||
Returns:
|
||||
The explicit root, current local root, default user root, or current
|
||||
directory when no initialized root exists.
|
||||
"""
|
||||
if root := os.environ.get("ASTRBOT_ROOT"):
|
||||
return Path(root).expanduser().resolve()
|
||||
|
||||
current_root = Path.cwd().resolve()
|
||||
if check_astrbot_root(current_root):
|
||||
return current_root
|
||||
|
||||
user_root = (Path.home() / ".astrbot").resolve()
|
||||
if check_astrbot_root(user_root):
|
||||
return user_root
|
||||
|
||||
return current_root
|
||||
|
||||
|
||||
async def check_dashboard(astrbot_root: Path) -> None:
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from astrbot.cli.commands import cmd_init
|
||||
from astrbot.core.utils.auth_password import verify_dashboard_password
|
||||
@@ -14,6 +19,7 @@ async def test_init_without_initial_password_env_does_not_create_config(
|
||||
async def fake_check_dashboard(_data_path):
|
||||
return None
|
||||
|
||||
monkeypatch.delenv("ASTRBOT_ROOT", raising=False)
|
||||
monkeypatch.delenv(cmd_init.DASHBOARD_INITIAL_PASSWORD_ENV, raising=False)
|
||||
monkeypatch.setattr(cmd_init, "check_dashboard", fake_check_dashboard)
|
||||
(tmp_path / ".astrbot").touch()
|
||||
@@ -32,6 +38,7 @@ async def test_init_uses_initial_password_env_to_create_config(
|
||||
return None
|
||||
|
||||
initial_password = "AstrBotInitialPassword123"
|
||||
monkeypatch.setenv("ASTRBOT_ROOT", str(tmp_path))
|
||||
monkeypatch.setenv(cmd_init.DASHBOARD_INITIAL_PASSWORD_ENV, initial_password)
|
||||
monkeypatch.setattr(cmd_init, "check_dashboard", fake_check_dashboard)
|
||||
(tmp_path / ".astrbot").touch()
|
||||
@@ -52,3 +59,71 @@ async def test_init_uses_initial_password_env_to_create_config(
|
||||
)
|
||||
assert dashboard_config["password_change_required"] is True
|
||||
assert dashboard_config["password_storage_upgraded"] is True
|
||||
|
||||
|
||||
def test_cli_main_import_does_not_create_cwd_data(tmp_path):
|
||||
repo_root = Path(__file__).resolve().parents[1]
|
||||
env = os.environ.copy()
|
||||
env.pop("ASTRBOT_ROOT", None)
|
||||
env["HOME"] = str(tmp_path / "home")
|
||||
env["PYTHONPATH"] = (
|
||||
str(repo_root)
|
||||
if not env.get("PYTHONPATH")
|
||||
else f"{repo_root}{os.pathsep}{env['PYTHONPATH']}"
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", "import astrbot.cli.__main__"],
|
||||
cwd=tmp_path,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert not (tmp_path / "data").exists()
|
||||
|
||||
|
||||
def test_init_defaults_to_user_runtime(monkeypatch, tmp_path):
|
||||
async def fake_check_dashboard(_data_path):
|
||||
return None
|
||||
|
||||
home = tmp_path / "home"
|
||||
workdir = tmp_path / "workdir"
|
||||
home.mkdir()
|
||||
workdir.mkdir()
|
||||
|
||||
monkeypatch.setenv("HOME", str(home))
|
||||
monkeypatch.delenv("ASTRBOT_ROOT", raising=False)
|
||||
monkeypatch.chdir(workdir)
|
||||
monkeypatch.setattr(cmd_init, "check_dashboard", fake_check_dashboard)
|
||||
|
||||
result = CliRunner().invoke(cmd_init.init, input="\n", env={"ASTRBOT_ROOT": ""})
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert (home / ".astrbot" / ".astrbot").exists()
|
||||
assert (home / ".astrbot" / "data" / "config").is_dir()
|
||||
assert not (workdir / "data").exists()
|
||||
|
||||
|
||||
def test_init_can_install_to_current_directory(monkeypatch, tmp_path):
|
||||
async def fake_check_dashboard(_data_path):
|
||||
return None
|
||||
|
||||
home = tmp_path / "home"
|
||||
workdir = tmp_path / "workdir"
|
||||
home.mkdir()
|
||||
workdir.mkdir()
|
||||
|
||||
monkeypatch.setenv("HOME", str(home))
|
||||
monkeypatch.delenv("ASTRBOT_ROOT", raising=False)
|
||||
monkeypatch.chdir(workdir)
|
||||
monkeypatch.setattr(cmd_init, "check_dashboard", fake_check_dashboard)
|
||||
|
||||
result = CliRunner().invoke(cmd_init.init, input="2\n", env={"ASTRBOT_ROOT": ""})
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert (workdir / ".astrbot").exists()
|
||||
assert (workdir / "data" / "config").is_dir()
|
||||
assert not (home / ".astrbot").exists()
|
||||
|
||||
@@ -30,6 +30,7 @@ def _read_config(config_path):
|
||||
|
||||
def test_password_command_changes_dashboard_password(monkeypatch, tmp_path):
|
||||
config_path = _write_config(tmp_path)
|
||||
monkeypatch.delenv("ASTRBOT_ROOT", raising=False)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
runner = CliRunner()
|
||||
@@ -55,6 +56,7 @@ def test_password_command_changes_dashboard_password(monkeypatch, tmp_path):
|
||||
|
||||
def test_password_command_can_update_dashboard_username(monkeypatch, tmp_path):
|
||||
config_path = _write_config(tmp_path)
|
||||
monkeypatch.delenv("ASTRBOT_ROOT", raising=False)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
runner = CliRunner()
|
||||
@@ -71,6 +73,7 @@ def test_password_command_can_update_dashboard_username(monkeypatch, tmp_path):
|
||||
|
||||
def test_conf_set_dashboard_password_updates_password_state(monkeypatch, tmp_path):
|
||||
config_path = _write_config(tmp_path)
|
||||
monkeypatch.delenv("ASTRBOT_ROOT", raising=False)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
Reference in New Issue
Block a user