mirror of
https://github.com/Shu-Ying/main_plugin_farm.git
synced 2026-10-29 05:29:58 +08:00
228 lines
10 KiB
Python
228 lines
10 KiB
Python
import base64
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from zipfile import ZipFile
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
from build_nonebot import build
|
|
from publish_nonebot import publish
|
|
from verify_nonebot import MANIFEST, ROOT_NAME, read_package, verify_files
|
|
|
|
|
|
class NoneBotPackageTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name)
|
|
self.bundle = build(ROOT, self.root / "built")
|
|
self.files = read_package(self.bundle)
|
|
|
|
def test_reproducible_standard_package_and_shared_runtime(self):
|
|
other = build(ROOT, self.root / "second")
|
|
self.assertEqual(self.bundle.read_bytes(), other.read_bytes())
|
|
from format_generated import format_generated
|
|
expected = {"nonebot_plugin_farm/runtime.py": (
|
|
ROOT / "integrations/shared/runtime.py"
|
|
).read_bytes()}
|
|
format_generated(ROOT, expected)
|
|
self.assertEqual(self.files["nonebot_plugin_farm/runtime.py"],
|
|
expected["nonebot_plugin_farm/runtime.py"])
|
|
for suffix in ("farm_core/data/plants.json",
|
|
"farm_render/data/asset-manifest.json",
|
|
"farm_render/data/font-manifest.json",
|
|
"farm_render/data/OFL-Xiaolai.txt"):
|
|
self.assertTrue(any(name.endswith(suffix) for name in self.files))
|
|
self.assertEqual(verify_files(self.files)["target"], "nonebot")
|
|
self.assertFalse(any(name.endswith('.ttf') for name in self.files))
|
|
for name in ("font-manifest.json", "OFL-Xiaolai.txt"):
|
|
self.assertEqual(
|
|
self.files[
|
|
"nonebot_plugin_farm/_vendor/farm_render/data/" + name
|
|
],
|
|
(ROOT / "src/farm_render/data" / name).read_bytes(),
|
|
)
|
|
self.assertIn("pyproject.toml", self.files)
|
|
self.assertIn(".github/workflows/publish.yaml", self.files)
|
|
self.assertFalse(any("legacy/" in name for name in self.files))
|
|
self.assertFalse(any(name.endswith('/assets.zip')
|
|
for name in self.files))
|
|
with self.assertRaises(FileExistsError):
|
|
build(ROOT, self.root / "built")
|
|
|
|
def test_host_entry_has_no_zhenxun_import(self):
|
|
import ast
|
|
for name, content in self.files.items():
|
|
if not name.endswith(".py"):
|
|
continue
|
|
tree = ast.parse(content, feature_version=(3, 10))
|
|
for node in ast.walk(tree):
|
|
imports = ([node.module] if isinstance(node, ast.ImportFrom)
|
|
else [alias.name for alias in node.names]
|
|
if isinstance(node, ast.Import) else [])
|
|
self.assertFalse(any(
|
|
value and value.startswith(("zhenxun", "zhenxun_utils"))
|
|
for value in imports
|
|
), name)
|
|
|
|
def test_tampering_and_paths_rejected(self):
|
|
files = dict(self.files)
|
|
files["pyproject.toml"] += b"# changed\n"
|
|
with self.assertRaisesRegex(ValueError, "checksum"):
|
|
verify_files(files)
|
|
for path in ("../escape", "extra", "C:/escape", "..\\escape"):
|
|
bundle = self.root / "bad.zip"
|
|
bundle.write_bytes(self.bundle.read_bytes())
|
|
with ZipFile(bundle, "a") as archive:
|
|
archive.writestr(path, b"invalid")
|
|
with self.assertRaises(ValueError):
|
|
read_package(bundle)
|
|
|
|
def clean_package(self):
|
|
files = dict(self.files)
|
|
manifest = json.loads(files[MANIFEST])
|
|
manifest["source"]["dirty"] = False
|
|
files[MANIFEST] = json.dumps(manifest).encode()
|
|
bundle = self.root / "clean.zip"
|
|
with ZipFile(bundle, "w") as archive:
|
|
for name, content in files.items():
|
|
archive.writestr(f"{ROOT_NAME}/{name}", content)
|
|
return bundle, manifest
|
|
|
|
def test_default_publish_is_offline_and_dirty_source_is_rejected(self):
|
|
def forbidden(*args):
|
|
self.fail("dry-run must not make requests")
|
|
|
|
report = publish(self.bundle, request=forbidden)
|
|
self.assertFalse(report["published"])
|
|
self.assertEqual(report["repository"], "Shu-Ying/nonebot_plugin_farm")
|
|
self.assertEqual(report["branch"], "main")
|
|
bundle, manifest = self.clean_package()
|
|
manifest["source"]["dirty"] = True
|
|
self.files[MANIFEST] = json.dumps(manifest).encode()
|
|
with ZipFile(bundle, "w") as archive:
|
|
for name, content in self.files.items():
|
|
archive.writestr(f"{ROOT_NAME}/{name}", content)
|
|
with self.assertRaisesRegex(ValueError, "clean source"):
|
|
publish(bundle, True, forbidden)
|
|
|
|
def test_fixed_sync_preserves_history_and_includes_pypi_workflow(self):
|
|
bundle, manifest = self.clean_package()
|
|
calls = []
|
|
|
|
def request(method, path, data):
|
|
calls.append((method, path, data))
|
|
if path == "":
|
|
return {"default_branch": "main"}
|
|
if path.startswith("git/ref/"):
|
|
return {"object": {"sha": "existing"}}
|
|
if path.startswith("contents/"):
|
|
return None
|
|
return {"sha": "generated"}
|
|
|
|
with patch.dict("os.environ", {"FARM_RELEASE_TOKEN": "test-token"}):
|
|
self.assertTrue(publish(bundle, True, request)["published"])
|
|
tree = next(data for method, path, data in calls
|
|
if path == "git/trees")
|
|
self.assertEqual({entry["path"] for entry in tree["tree"]},
|
|
set(self.files))
|
|
commit = next(data for method, path, data in calls
|
|
if path == "git/commits")
|
|
self.assertEqual(commit["parents"], ["existing"])
|
|
self.assertEqual(commit["message"], (
|
|
manifest["source"]["message"] + "\n\n"
|
|
+ "Farm-Source-Commit: " + manifest["source"]["commit"] + "\n"
|
|
+ "Farm-Version: " + manifest["version"]
|
|
))
|
|
self.assertIn(("PATCH", "git/refs/heads/main", {
|
|
"sha": "generated", "force": False,
|
|
}), calls)
|
|
|
|
def test_wrong_owner_and_concurrent_change_are_rejected(self):
|
|
bundle, manifest = self.clean_package()
|
|
for wrong_owner in (True, False):
|
|
calls = []
|
|
|
|
def request(method, path, data):
|
|
calls.append((method, path, data))
|
|
if path == "":
|
|
return {"default_branch": "main"}
|
|
if path.startswith("contents/"):
|
|
previous = {**manifest, "target": (
|
|
"zhenxun" if wrong_owner else "nonebot"
|
|
)}
|
|
return {"encoding": "base64", "content": base64.b64encode(
|
|
json.dumps(previous).encode(),
|
|
).decode()}
|
|
if method == "GET":
|
|
reads = sum(p.startswith("git/ref/") for _, p, _ in calls)
|
|
return {"object": {"sha": (
|
|
"existing" if reads == 1 else "changed"
|
|
)}}
|
|
return {"sha": "generated"}
|
|
|
|
with patch.dict("os.environ", {"FARM_RELEASE_TOKEN": "test"}):
|
|
with self.assertRaises((ValueError, RuntimeError)):
|
|
publish(bundle, True, request)
|
|
self.assertFalse(any(method == "PATCH" for method, _, _ in calls))
|
|
|
|
def test_release_tag_is_created_after_main_sync(self):
|
|
bundle, manifest = self.clean_package()
|
|
calls = []
|
|
def request(method, path, data):
|
|
calls.append((method, path, data))
|
|
if path == '':
|
|
return {'default_branch': 'main'}
|
|
if path.startswith('git/ref/tags/'):
|
|
return None
|
|
if path.startswith('git/ref/heads/'):
|
|
return {'object': {'sha': 'existing'}}
|
|
if path.startswith('contents/'):
|
|
return None
|
|
return {'sha': 'generated'}
|
|
with patch.dict('os.environ', {'FARM_RELEASE_TOKEN': 'test'}):
|
|
result = publish(bundle, True, request, release_tag=True)
|
|
self.assertTrue(result['tag_created'])
|
|
self.assertEqual(calls[-1], ('POST', 'git/refs', {
|
|
'ref': 'refs/tags/v' + manifest['version'], 'sha': 'generated',
|
|
}))
|
|
self.assertLess(next(i for i, call in enumerate(calls)
|
|
if call[0] == 'PATCH'), len(calls) - 1)
|
|
|
|
def test_existing_release_tag_is_reused_or_conflict_blocks_all_writes(self):
|
|
bundle, manifest = self.clean_package()
|
|
for matching in (True, False):
|
|
calls = []
|
|
previous = json.loads(json.dumps(manifest))
|
|
if not matching:
|
|
previous['files']['README.md'] = '0' * 64
|
|
def request(method, path, data):
|
|
calls.append((method, path, data))
|
|
if path == '':
|
|
return {'default_branch': 'main'}
|
|
if path.startswith('git/ref/tags/'):
|
|
return {'object': {'sha': 'released'}}
|
|
if path.startswith('contents/'):
|
|
return {'encoding': 'base64', 'content': base64.b64encode(
|
|
json.dumps(previous).encode(),
|
|
).decode()}
|
|
self.fail('Unexpected request: ' + path)
|
|
with patch.dict('os.environ', {'FARM_RELEASE_TOKEN': 'test'}):
|
|
if matching:
|
|
result = publish(bundle, True, request, release_tag=True)
|
|
self.assertFalse(result['tag_created'])
|
|
else:
|
|
with self.assertRaisesRegex(ValueError, 'version.txt'):
|
|
publish(bundle, True, request, release_tag=True)
|
|
self.assertTrue(all(method == 'GET' for method, _, _ in calls))
|
|
|
|
def test_release_tag_dry_run_does_not_access_remote(self):
|
|
result = publish(self.bundle, release_tag=True,
|
|
request=lambda *args: self.fail('Network in dry run'))
|
|
self.assertEqual(result['tag'], 'v' + result['version'])
|
|
self.assertFalse(result['tag_created'])
|