83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
"""验证本地运行目录策略，不需要连接设备。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
CLIENT_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(CLIENT_ROOT / "src"))
|
|
|
|
from cmbuyer_client.runtime import LocalStateRuntime, RuntimePaths
|
|
|
|
|
|
class RuntimePathsTests(unittest.TestCase):
|
|
def test_ensure_exists_creates_only_runtime_directories(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
paths = RuntimePaths.from_root(Path(directory) / "runtime")
|
|
|
|
paths.ensure_exists()
|
|
|
|
self.assertTrue(paths.logs.is_dir())
|
|
self.assertTrue(paths.artifacts.is_dir())
|
|
self.assertTrue(paths.state.is_dir())
|
|
self.assertEqual(paths.database, paths.state / "client-state.sqlite3")
|
|
|
|
def test_localstate_runtime_acquires_mutex_before_protector_and_store(self) -> None:
|
|
events: list[str] = []
|
|
|
|
class Mutex:
|
|
def __init__(self, path: Path) -> None:
|
|
events.append("mutex")
|
|
|
|
def close(self) -> None:
|
|
events.append("close")
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
runtime = LocalStateRuntime.open(
|
|
RuntimePaths.from_root(Path(directory)),
|
|
mutex_factory=Mutex,
|
|
protector_factory=lambda: events.append("protector") or object(),
|
|
store_factory=lambda path, protector: events.append("store") or object(),
|
|
)
|
|
runtime.close()
|
|
self.assertEqual(events, ["mutex", "protector", "store", "close"])
|
|
|
|
def test_localstate_runtime_releases_mutex_if_open_fails(self) -> None:
|
|
events: list[str] = []
|
|
|
|
class Mutex:
|
|
def __init__(self, path: Path) -> None:
|
|
events.append("mutex")
|
|
|
|
def close(self) -> None:
|
|
events.append("close")
|
|
|
|
def fail() -> object:
|
|
raise RuntimeError("failed")
|
|
|
|
with tempfile.TemporaryDirectory() as directory, self.assertRaises(RuntimeError):
|
|
LocalStateRuntime.open(
|
|
RuntimePaths.from_root(Path(directory)),
|
|
mutex_factory=Mutex,
|
|
protector_factory=fail,
|
|
)
|
|
self.assertEqual(events, ["mutex", "close"])
|
|
|
|
def test_windows_without_localappdata_fails_instead_of_creating_second_database(self) -> None:
|
|
with mock.patch("cmbuyer_client.runtime.os.name", "nt"), mock.patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(RuntimeError, "local_app_data_required"):
|
|
RuntimePaths.default()
|
|
|
|
def test_runtime_root_is_frozen_absolute_and_relative_localappdata_is_rejected(self) -> None:
|
|
paths = RuntimePaths.from_root(Path("relative-runtime"))
|
|
self.assertTrue(paths.root.is_absolute())
|
|
with mock.patch.dict(os.environ, {"LOCALAPPDATA": "relative-local-app-data"}, clear=True):
|
|
with self.assertRaisesRegex(RuntimeError, "local_app_data_must_be_absolute"):
|
|
RuntimePaths.default()
|