mirror of
https://github.com/Shu-Ying/main_plugin_farm.git
synced 2026-10-28 21:19:58 +08:00
add # 新增对zhenxun的工作流
This commit is contained in:
@@ -0,0 +1,160 @@
|
||||
import importlib.util
|
||||
import base64
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from zipfile import ZipFile
|
||||
from unittest.mock import patch
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
try:
|
||||
from build_zhenxun import build
|
||||
from install_zhenxun import install
|
||||
from verify_zhenxun import verify_package
|
||||
except ImportError:
|
||||
build = None
|
||||
|
||||
|
||||
class ZhenxunPackageTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.assertIsNotNone(build, "真寻打包功能未实现")
|
||||
self.temporary = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temporary.cleanup)
|
||||
self.root = Path(self.temporary.name)
|
||||
self.output = self.root / "bundle"
|
||||
self.bundle = build(ROOT, self.output, "0.1.0-preview.1")
|
||||
|
||||
def test_reproducible_private_complete_package(self):
|
||||
other = build(ROOT, self.root / "second", "0.1.0-preview.1")
|
||||
self.assertEqual(self.bundle.read_bytes(), other.read_bytes())
|
||||
manifest = verify_package(self.bundle)
|
||||
self.assertEqual(manifest["target"], "zhenxun")
|
||||
self.assertEqual(manifest["version"], "0.1.0-preview.1")
|
||||
self.assertIn("dirty", manifest["source"])
|
||||
with ZipFile(self.bundle) as archive:
|
||||
paths = archive.namelist()
|
||||
self.assertTrue(any(p.endswith("assets.zip") for p in paths))
|
||||
self.assertFalse(any("astrbot" in p or "legacy/" in p
|
||||
for p in paths))
|
||||
source = archive.read(next(
|
||||
p for p in paths if p.endswith("farm_render/models.py")
|
||||
)).decode()
|
||||
self.assertIn("from ..farm_core.domain", source)
|
||||
self.assertNotIn("from farm_core.domain", source)
|
||||
with self.assertRaises(FileExistsError):
|
||||
build(ROOT, self.output, "0.1.0-preview.1")
|
||||
|
||||
def test_install_and_existing_target_are_safe(self):
|
||||
plugins = self.root / "plugins"
|
||||
plugins.mkdir()
|
||||
target = install(self.bundle, plugins)
|
||||
sentinel = target / "do-not-overwrite.txt"
|
||||
sentinel.write_text("preserved")
|
||||
with self.assertRaises(FileExistsError):
|
||||
install(self.bundle, plugins)
|
||||
self.assertEqual(sentinel.read_text(), "preserved")
|
||||
self.assertTrue((target / "__init__.py").is_file())
|
||||
|
||||
def test_tampering_extra_path_and_traversal_rejected(self):
|
||||
for name in ("../outside.txt", "C:/escape.txt", "extra.txt",
|
||||
"..\\outside.txt"):
|
||||
with self.subTest(name=name):
|
||||
bad = self.root / (str(len(name)) + ".zip")
|
||||
bad.write_bytes(self.bundle.read_bytes())
|
||||
with ZipFile(bad, "a") as archive:
|
||||
archive.writestr(name, "bad")
|
||||
with self.assertRaises(ValueError):
|
||||
verify_package(bad)
|
||||
bad = self.root / "corrupt.zip"
|
||||
with ZipFile(self.bundle) as original, ZipFile(bad, "w") as changed:
|
||||
for item in original.infolist():
|
||||
content = original.read(item.filename)
|
||||
if item.filename.endswith("requirements.txt"):
|
||||
content += b"bad\n"
|
||||
changed.writestr(item, content)
|
||||
with self.assertRaises(ValueError):
|
||||
install(bad, self.root)
|
||||
self.assertFalse((self.root / "zhenxun_plugin_farm_preview").exists())
|
||||
|
||||
def test_publisher_is_fixed_target_and_dry_by_default(self):
|
||||
from publish_zhenxun import publish
|
||||
report = publish(self.bundle)
|
||||
self.assertFalse(report["published"])
|
||||
self.assertEqual(report["branch"], "farm-preview")
|
||||
self.assertEqual(report["repository"], "Shu-Ying/zhenxun_plugin_farm")
|
||||
self.assertNotIn("token", json.dumps(report))
|
||||
with self.assertRaises(ValueError):
|
||||
build(ROOT, self.root / "bad", "bad; echo secret")
|
||||
|
||||
def clean_package(self):
|
||||
from verify_zhenxun import read_package, MANIFEST, ROOT_NAME
|
||||
files = read_package(self.bundle)
|
||||
manifest = json.loads(files[MANIFEST])
|
||||
manifest["source"]["dirty"] = False
|
||||
files[MANIFEST] = json.dumps(manifest).encode()
|
||||
path = self.root / "clean.zip"
|
||||
with ZipFile(path, "w") as archive:
|
||||
for name, content in files.items():
|
||||
archive.writestr(f"{ROOT_NAME}/{name}", content)
|
||||
return path, manifest
|
||||
|
||||
def test_publish_uses_nonforced_managed_branch_and_rejects_race(self):
|
||||
from publish_zhenxun import publish
|
||||
package, manifest = self.clean_package()
|
||||
for race in (False, True):
|
||||
calls = []
|
||||
reads = []
|
||||
|
||||
def request(method, path, data):
|
||||
calls.append((method, path, data))
|
||||
if path == "":
|
||||
return {"default_branch":"main"}
|
||||
if path.startswith("contents/"):
|
||||
return {"encoding":"base64", "content":base64.b64encode(
|
||||
json.dumps(manifest).encode(),
|
||||
).decode()}
|
||||
if method == "GET":
|
||||
reads.append(1)
|
||||
return {"object":{"sha":(
|
||||
"raced" if race and len(reads) > 1 else "parent"
|
||||
)}}
|
||||
return {"sha":"created"}
|
||||
|
||||
with patch.dict("os.environ", {"FARM_RELEASE_TOKEN":"unit-test"}):
|
||||
if race:
|
||||
with self.assertRaises(RuntimeError):
|
||||
publish(package, True, request)
|
||||
self.assertFalse(any(c[0] == "PATCH" for c in calls))
|
||||
else:
|
||||
self.assertTrue(publish(
|
||||
package, True, request,
|
||||
)["published"])
|
||||
self.assertIn(("PATCH", "git/refs/heads/farm-preview", {
|
||||
"sha":"created", "force":False,
|
||||
}), calls)
|
||||
|
||||
def test_symlink_duplicate_and_default_branch_rejected(self):
|
||||
from publish_zhenxun import publish
|
||||
from zipfile import ZipInfo
|
||||
package, _ = self.clean_package()
|
||||
with patch.dict("os.environ", {"FARM_RELEASE_TOKEN":"unit-test"}):
|
||||
with self.assertRaises(ValueError):
|
||||
publish(package, True, lambda *args: {
|
||||
"default_branch":"farm-preview",
|
||||
})
|
||||
for kind in ("duplicate", "link"):
|
||||
bad = self.root / (kind + ".zip")
|
||||
bad.write_bytes(self.bundle.read_bytes())
|
||||
with ZipFile(bad, "a") as archive:
|
||||
item = ZipInfo("zhenxun_plugin_farm_preview/extra")
|
||||
if kind == "duplicate":
|
||||
item.filename = archive.namelist()[0]
|
||||
else:
|
||||
item.create_system = 3
|
||||
item.external_attr = 0o120777 << 16
|
||||
archive.writestr(item, b"outside")
|
||||
with self.assertRaises(ValueError):
|
||||
verify_package(bad)
|
||||
@@ -0,0 +1,153 @@
|
||||
import asyncio
|
||||
import importlib
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import tempfile
|
||||
from types import ModuleType, SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
from build_zhenxun import build
|
||||
from ci_font import choose_font
|
||||
|
||||
|
||||
class ZhenxunRuntimeTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def asyncSetUp(self):
|
||||
self.temporary = tempfile.TemporaryDirectory()
|
||||
self.root = Path(self.temporary.name)
|
||||
build(ROOT, self.root / "built", "0.1.0-preview.1")
|
||||
self.package = "farm_fixture"
|
||||
module = ModuleType(self.package)
|
||||
module.__path__ = [str(self.root / "built/repository")]
|
||||
sys.modules[self.package] = module
|
||||
self.module = importlib.import_module(self.package + ".runtime")
|
||||
self.runtime = self.module.Runtime(self.root / "data", choose_font())
|
||||
await self.runtime.start()
|
||||
self.context = dict(bot_id="bot", platform="OneBot V11", user_id="1",
|
||||
message_id="registration", display_name="离线玩家")
|
||||
|
||||
async def asyncTearDown(self):
|
||||
await self.runtime.close()
|
||||
for name in tuple(sys.modules):
|
||||
if name == self.package or name.startswith(self.package + "."):
|
||||
del sys.modules[name]
|
||||
self.temporary.cleanup()
|
||||
|
||||
async def call(self, action, arguments="", message="event", **changes):
|
||||
return await self.runtime.execute(
|
||||
action, arguments,
|
||||
**{**self.context, "message_id":message, **changes},
|
||||
)
|
||||
|
||||
async def test_original_commands_transactions_and_png(self):
|
||||
self.assertNotIn("测试农场", self.module.COMMANDS)
|
||||
self.assertIn("开通农场", self.module.COMMANDS)
|
||||
self.assertIn("我的农场", self.module.COMMANDS)
|
||||
self.assertIsInstance(await self.call("register"), str)
|
||||
await self.call("buy", "白萝卜 1", "buy")
|
||||
before = await self.call("inventory")
|
||||
await self.call("buy", "白萝卜 1", "buy")
|
||||
self.assertEqual(before, await self.call("inventory"))
|
||||
await self.call("sow", "白萝卜 1", "sow")
|
||||
result = await self.call("farm", "高清")
|
||||
self.assertIsInstance(result, bytes)
|
||||
await self.call("farm", "my-seed")
|
||||
self.assertIn("成熟", await self.call("harvest", message="harvest"))
|
||||
self.assertTrue((self.root / "data/farm-preview.log").exists())
|
||||
|
||||
async def test_missing_message_id_and_other_identity(self):
|
||||
self.assertIn("未执行", await self.call("register", message=None))
|
||||
await self.call("register")
|
||||
response = await self.call("inventory", user_id="other")
|
||||
self.assertIsInstance(response, str)
|
||||
self.assertTrue("注册" in response or "开通" in response)
|
||||
self.assertNotEqual(
|
||||
self.module.operation_id("b", "p", "u", "1", "buy"),
|
||||
self.module.operation_id("b", "p", "u", "2", "buy"),
|
||||
)
|
||||
|
||||
async def test_invalid_arguments_and_missing_font_are_explicit(self):
|
||||
await self.call("register")
|
||||
for action, arguments in (("buy", "白萝卜 nope"),
|
||||
("sow", "白萝卜 -2"),
|
||||
("lock", ""), ("farm", "unknown")):
|
||||
self.assertIsInstance(await self.call(action, arguments), str)
|
||||
other = self.module.Runtime(self.root / "without-font")
|
||||
await other.start()
|
||||
try:
|
||||
response = await other.execute("farm", "", **self.context)
|
||||
self.assertIn("FARM_PREVIEW_FONT", response)
|
||||
finally:
|
||||
await other.close()
|
||||
|
||||
async def test_insufficient_funds_has_chinese_message_and_no_debit(self):
|
||||
await self.call("register")
|
||||
before = await self.call("inventory")
|
||||
self.assertEqual(
|
||||
await self.call("buy", "白萝卜 100000", "too-expensive"),
|
||||
"余额不足。",
|
||||
)
|
||||
self.assertEqual(before, await self.call("inventory"))
|
||||
|
||||
async def test_mock_host_load_registers_hooks_and_original_matchers(self):
|
||||
hooks = {"start":[], "close":[], "commands":{}}
|
||||
driver = SimpleNamespace(
|
||||
config=SimpleNamespace(farm_preview_data_dir=self.root / "host"),
|
||||
on_startup=hooks["start"].append,
|
||||
on_shutdown=hooks["close"].append,
|
||||
)
|
||||
nonebot = ModuleType("nonebot")
|
||||
nonebot.get_driver = lambda: driver
|
||||
nonebot.require = lambda _: None
|
||||
|
||||
def register(name, **kwargs):
|
||||
return SimpleNamespace(handle=lambda: (
|
||||
lambda handler: hooks["commands"].update({name:handler})
|
||||
))
|
||||
|
||||
nonebot.on_command = register
|
||||
adapters = ModuleType("nonebot.adapters")
|
||||
adapters.Bot = adapters.Event = adapters.Message = type(
|
||||
"HostType", (), {},
|
||||
)
|
||||
params = ModuleType("nonebot.params")
|
||||
params.CommandArg = lambda: None
|
||||
uninfo = ModuleType("nonebot_plugin_uninfo")
|
||||
uninfo.Uninfo = type("Uninfo", (), {})
|
||||
paths = ModuleType("zhenxun.configs.path_config")
|
||||
paths.DATA_PATH = self.root
|
||||
messages = ModuleType("zhenxun.utils.message")
|
||||
sent = []
|
||||
|
||||
async def send(**kwargs):
|
||||
return None
|
||||
|
||||
messages.MessageUtils = SimpleNamespace(build_message=lambda content: (
|
||||
sent.append(content) or SimpleNamespace(send=send)
|
||||
))
|
||||
platform = ModuleType("zhenxun.utils.platform")
|
||||
platform.PlatformUtils = SimpleNamespace()
|
||||
modules = {m.__name__:m for m in (
|
||||
nonebot, adapters, params, uninfo, paths, messages, platform,
|
||||
)}
|
||||
with patch.dict(sys.modules, modules):
|
||||
imported = importlib.import_module(self.package + ".commands")
|
||||
self.assertEqual(set(hooks["commands"]), set(self.module.COMMANDS))
|
||||
self.assertEqual(len(hooks["start"]), 1)
|
||||
await hooks["start"][0]()
|
||||
try:
|
||||
bot = SimpleNamespace(self_id="bot", adapter=SimpleNamespace(
|
||||
get_name=lambda:"OneBot V11",
|
||||
))
|
||||
await hooks["commands"]["开通农场"](
|
||||
bot, SimpleNamespace(message_id="m1"),
|
||||
SimpleNamespace(user=SimpleNamespace(id="u", name="用户")),
|
||||
SimpleNamespace(extract_plain_text=lambda:""),
|
||||
)
|
||||
self.assertTrue(sent)
|
||||
finally:
|
||||
await hooks["close"][0]()
|
||||
self.assertFalse(imported.runtime.started)
|
||||
Reference in New Issue
Block a user