"""amane 桥脚本本身：一次子进程、一行 JSON、退出码分档。不装 amane，塞假的 Runtime 进去。""" from __future__ import annotations import importlib.util import io import json from pathlib import Path import subprocess import sys import tempfile import unittest from unittest import mock import zipfile ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "tools" / "amane-bridge" / "bridge.py" def load_bridge(): spec = importlib.util.spec_from_file_location("amane_bridge_under_test", SCRIPT) module = importlib.util.module_from_spec(spec) # dataclass 解析 `from __future__ import annotations` 下的字段类型要在 sys.modules 里找到它。 sys.modules[spec.name] = module spec.loader.exec_module(module) return module bridge = load_bridge() class FakeSourceError(Exception): def __init__(self, reason, *, http_status=None, detail="", url=""): super().__init__(detail) self.reason, self.http_status, self.detail, self.url = reason, http_status, detail, url class FakeMetadata: def __init__(self, **fields): self.fields = fields def model_dump(self, mode="json"): return dict(self.fields) def make_runtime(outcomes): """`outcomes[站名]` 是要返回的资料、None（没有）或要抛的异常。""" class Crawler: site = "" def __init__(self, *, client, config): self.client, self.config = client, config async def fetch(self, query, options): outcome = outcomes[self.site] if isinstance(outcome, BaseException): raise outcome return outcome def crawler(site): return type(f"{site}Crawler", (Crawler,), {"site": site}) return bridge.Runtime( crawler=crawler, make_client=lambda proxy, timeout, rate: {"proxy": proxy}, query=lambda number: {"number": number}, options=lambda language: {"language": language}, source_error=FakeSourceError, version="0.16.1") def run_main(argv, runtime): out = io.StringIO() code = bridge.main(argv, runtime=runtime, out=out) lines = [line for line in out.getvalue().splitlines() if line.strip()] return code, lines class BridgeContractTests(unittest.TestCase): def test_cli_outputs_utf8_even_when_redirected_streams_start_as_gbk(self): wrapper = ( "import runpy, sys; " "sys.stdout.reconfigure(encoding='gbk'); " "sys.stderr.reconfigure(encoding='gbk'); " "sys.argv = sys.argv[1:]; " "runpy.run_path(sys.argv[0], run_name='__main__')" ) for arguments in ([], ['--未知参数']): with self.subTest(arguments=arguments): result = subprocess.run([sys.executable, '-I', '-c', wrapper, str(SCRIPT), *arguments], capture_output=True, check=False, timeout=30) self.assertEqual(result.returncode, bridge.EXIT_USAGE) if not arguments: self.assertEqual(json.loads(result.stdout.decode('utf-8')), {'error': '番号与站名都不能为空'}) else: self.assertIn('--未知参数', result.stderr.decode('utf-8')) def test_self_check_loads_every_site_without_creating_a_client(self): runtime = make_runtime({}) factory = mock.Mock(side_effect=AssertionError("离线自检不能创建 HTTP 客户端")) crawler = mock.Mock(side_effect=runtime.crawler) runtime = bridge.Runtime(crawler=crawler, make_client=factory, query=runtime.query, options=runtime.options, source_error=runtime.source_error, version="0.20.0") code, lines = run_main(["--self-check"], runtime) self.assertEqual(code, 0) self.assertEqual(len(lines), 1) self.assertEqual(json.loads(lines[0]), { "ok": True, "amane": {"version": "0.20.0"}, "sites": list(bridge.SITES)}) self.assertEqual(crawler.call_args_list, [mock.call(site) for site in bridge.SITES]) factory.assert_not_called() def test_self_check_reports_missing_site_dependencies(self): runtime = make_runtime({}) runtime = bridge.Runtime(crawler=mock.Mock(side_effect=ImportError("missing parser")), make_client=runtime.make_client, query=runtime.query, options=runtime.options, source_error=runtime.source_error) code, lines = run_main(["--self-check"], runtime) self.assertEqual(code, bridge.EXIT_FAILED) self.assertFalse(json.loads(lines[0])["ok"]) self.assertIn("ImportError: missing parser", json.loads(lines[0])["error"]) def test_mixed_site_networks_use_the_requested_client(self): seen = {} class Crawler: def __init__(self, *, client, config): self.client = client async def fetch(self, query, options): seen[self.site] = self.client['proxy'] return None runtime = bridge.Runtime( crawler=lambda site: type(site, (Crawler,), {"site": site}), make_client=lambda proxy, timeout, rate: {"proxy": proxy}, query=lambda number: number, options=lambda language: None, source_error=FakeSourceError) code, _ = run_main(["--number", "ABW-032", "--sites", "airav,freejavbt", "--proxy", "http://127.0.0.1:7890", "--direct-sites", "airav"], runtime) self.assertEqual(code, bridge.EXIT_NOT_FOUND) self.assertEqual(seen, {"airav": None, "freejavbt": "http://127.0.0.1:7890"}) def test_failure_reasons_have_unique_values(self): self.assertEqual(len(bridge.FAILURE_REASONS), 18) self.assertEqual(len(set(bridge.FAILURE_REASONS)), 18) def test_a_login_requirement_keeps_its_failure_reason(self): runtime = make_runtime({"airav": FakeSourceError("login_required", detail="login page")}) code, lines = run_main(["--number", "ABW-032", "--sites", "airav"], runtime) self.assertEqual(code, bridge.EXIT_FAILED) record = json.loads(lines[0])["sites"]["airav"] self.assertEqual(record["status"], "failed") self.assertEqual(record["reason"], "login_required") self.assertEqual(record["detail"], "login page") def test_a_hit_prints_one_json_line_and_exits_zero(self): runtime = make_runtime({ "fc2club": FakeMetadata(number="FC2-PPV-1", title="標題", actors=[{"name": "A", "gender": "female"}]), "avsox": None, }) code, lines = run_main(["--number", "FC2-PPV-1", "--sites", "fc2club,avsox"], runtime) self.assertEqual(code, bridge.EXIT_FOUND) self.assertEqual(len(lines), 1) report = json.loads(lines[0]) self.assertEqual(report["number"], "FC2-PPV-1") self.assertEqual(report["amane"], {"version": "0.16.1"}) self.assertEqual(report["sites"]["fc2club"]["status"], "found") self.assertEqual(report["sites"]["fc2club"]["metadata"]["title"], "標題") self.assertEqual(report["sites"]["avsox"], {"status": "not_found", "reason": "not_found", "elapsed_s": report["sites"]["avsox"]["elapsed_s"]}) def test_all_misses_exit_two_and_any_error_without_a_hit_exits_three(self): runtime = make_runtime({"avsox": None, "airav": None}) code, _ = run_main(["--number", "ABW-1", "--sites", "avsox,airav"], runtime) self.assertEqual(code, bridge.EXIT_NOT_FOUND) runtime = make_runtime({ "avsox": None, "airav": FakeSourceError("cloudflare_blocked", http_status=403, detail="blocked", url="https://x/y"), }) code, lines = run_main(["--number", "ABW-1", "--sites", "avsox,airav"], runtime) self.assertEqual(code, bridge.EXIT_FAILED) record = json.loads(lines[0])["sites"]["airav"] self.assertEqual(record["status"], "failed") self.assertEqual(record["reason"], "cloudflare_blocked") self.assertEqual(record["http_status"], 403) self.assertEqual(record["url"], "https://x/y") def test_an_unknown_reason_is_kept_in_detail_under_unexpected(self): record = bridge.failure_record(FakeSourceError("brand_new_reason", detail="x" * 600)) self.assertEqual(record["reason"], "unexpected") self.assertTrue(record["detail"].startswith("brand_new_reason: xxx")) self.assertLessEqual(len(record["detail"]), 500) def test_a_crash_inside_one_site_does_not_take_the_batch_down(self): runtime = make_runtime({"avsox": RuntimeError("boom"), "freejavbt": FakeMetadata(number="ABW-1", title="t")}) code, lines = run_main(["--number", "ABW-1", "--sites", "avsox,freejavbt"], runtime) self.assertEqual(code, bridge.EXIT_FOUND) report = json.loads(lines[0]) self.assertEqual(report["sites"]["avsox"]["reason"], "unexpected") self.assertIn("RuntimeError: boom", report["sites"]["avsox"]["detail"]) def test_usage_errors_exit_four_with_a_json_error(self): runtime = make_runtime({}) code, lines = run_main(["--number", "ABW-1", "--sites", "nosuchsite"], runtime) self.assertEqual(code, bridge.EXIT_USAGE) self.assertEqual(json.loads(lines[0]), {"error": "未知站名：nosuchsite"}) code, lines = run_main(["--number", " ", "--sites", "avsox"], runtime) self.assertEqual(code, bridge.EXIT_USAGE) self.assertIn("error", json.loads(lines[0])) code, _ = run_main(["--sites", "avsox"], runtime) self.assertEqual(code, bridge.EXIT_USAGE) def test_the_proxy_reaches_the_client_factory(self): seen = {} def make_client(proxy, timeout, rate): seen.update(proxy=proxy, timeout=timeout, rate=rate) return None runtime = make_runtime({"avsox": None}) runtime = bridge.Runtime(crawler=runtime.crawler, make_client=make_client, query=runtime.query, options=runtime.options, source_error=runtime.source_error) run_main(["--number", "ABW-1", "--sites", "avsox", "--proxy", "http://127.0.0.1:7897", "--timeout", "12", "--rate", "1.5"], runtime) self.assertEqual(seen, {"proxy": "http://127.0.0.1:7897", "timeout": 12.0, "rate": 1.5}) class OfficialSiteTests(unittest.TestCase): """官方档的五站（ADR-0048）走同一份 JSON 契约；`makers` 是 amane 的 `official` 模块。""" def test_the_official_sites_name_their_amane_crawlers(self): self.assertEqual(bridge.SITES["makers"], ("official", "OfficialCrawler")) self.assertEqual(bridge.SITES["prestige"], ("prestige", "PrestigeCrawler")) self.assertEqual(bridge.SITES["faleno"], ("faleno", "FalenoCrawler")) self.assertEqual(bridge.SITES["dahlia"], ("dahlia", "DahliaCrawler")) self.assertEqual(bridge.SITES["mgstage"], ("mgstage", "MGStageCrawler")) self.assertNotIn("official", bridge.SITES) def test_each_official_site_reports_hit_miss_and_refusal_on_its_own(self): runtime = make_runtime({ "makers": FakeMetadata(number="SSIS-057", title="標題", studio="S1 NO.1 STYLE", release="2021-05-07"), "prestige": FakeSourceError("http_error", http_status=403, detail="Forbidden", url="https://www.prestige-av.com/api/product/search"), "faleno": None, "dahlia": FakeSourceError("geo_restricted", http_status=403, detail="not available in your region"), "mgstage": FakeSourceError("age_verification", detail="adc cookie"), }) code, lines = run_main(["--number", "SSIS-057", "--sites", "makers,prestige,faleno,dahlia,mgstage"], runtime) self.assertEqual(code, bridge.EXIT_FOUND) sites = json.loads(lines[0])["sites"] self.assertEqual(sites["makers"]["metadata"]["studio"], "S1 NO.1 STYLE") self.assertEqual((sites["prestige"]["reason"], sites["prestige"]["http_status"]), ("http_error", 403)) self.assertEqual(sites["faleno"]["status"], "not_found") self.assertEqual(sites["dahlia"]["reason"], "geo_restricted") self.assertEqual(sites["mgstage"]["reason"], "age_verification") def test_a_refused_maker_site_alone_is_a_failure_not_a_miss(self): """被挡与没有分开报：只问 Prestige 且被 403 挡住时退出码是 3，不是 2。""" runtime = make_runtime({"prestige": FakeSourceError("http_error", http_status=403)}) code, _ = run_main(["--number", "ABW-032", "--sites", "prestige"], runtime) self.assertEqual(code, bridge.EXIT_FAILED) class RuntimePackagingTests(unittest.TestCase): def setUp(self): spec = importlib.util.spec_from_file_location( "amane_runtime_builder_under_test", ROOT / "scripts/build_amane_runtime.py") self.builder = importlib.util.module_from_spec(spec) spec.loader.exec_module(self.builder) directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) self.root = Path(directory.name).resolve() def manifest(self, kind): facts = {**self.builder.locked_amane(), "kind": kind, "executable": "runtime.exe" if kind == "executable" else ".venv/bin/python"} (self.root / "manifest.json").write_text(json.dumps(facts), encoding="utf-8") return facts def test_locked_runtime_uses_the_existing_revision_and_build_toolchain(self): facts = self.builder.locked_amane() self.assertEqual(facts, {"revision": "fd9294fbe5b4aa88ce1df84d534f6816ad9460d5", "version": "0.22.1"}) requirements = self.builder.build_requirements() self.assertIn("pyinstaller==6.22.3", requirements) self.assertTrue(all("==" in item for item in requirements)) self.assertNotIn("packaging", [item.split("==")[0] for item in requirements]) def test_runtime_self_check_dispatches_both_packaging_kinds(self): for kind in ("python", "executable"): facts = self.manifest(kind) report = {"ok": True, "amane": {"version": facts["version"]}, "sites": list(bridge.SITES)} with mock.patch.object(self.builder.subprocess, "run", return_value=mock.Mock( stdout=json.dumps(report))) as run: self.assertEqual(self.builder.verify(self.root), report) command = run.call_args.args[0] self.assertEqual(command[0], str(self.root / facts["executable"])) self.assertEqual(command[-1], "--self-check") self.assertEqual("-X" in command, kind == "python") self.assertNotIn("PYTHONPATH", run.call_args.kwargs["env"]) self.assertEqual(run.call_args.kwargs["env"]["PATH"], "") self.assertEqual(run.call_args.kwargs["env"]["PYINSTALLER_RESET_ENVIRONMENT"], "1") def test_desktop_smoke_requires_the_service_to_use_the_bundled_runtime(self): from scripts import smoke_desktop manifest = self.manifest('executable') facts = {'installed': True, 'problem': '', 'revision': manifest['revision'], 'installed_version': manifest['version'], 'python': str(self.root / manifest['executable'])} opener = mock.Mock() with mock.patch.dict(sys.modules, {'build_amane_runtime': self.builder}), \ mock.patch.object(self.builder, 'verify') as verify: opener.open.return_value = io.StringIO(json.dumps(facts)) smoke_desktop.assert_amane_runtime(opener, 'http://127.0.0.1:8900', self.root) verify.assert_called_once_with(self.root) opener.open.assert_called_once_with('http://127.0.0.1:8900/api/scraping/amane-bridge', timeout=15) for change in ({'installed': False}, {'revision': 'foreign'}, {'installed_version': 'unknown'}, {'python': str(self.root / 'user-python')}, {'problem': '缺少运行环境'}): with self.subTest(change=change): opener.open.return_value = io.StringIO(json.dumps({**facts, **change})) with self.assertRaises(AssertionError): smoke_desktop.assert_amane_runtime(opener, 'http://127.0.0.1:8900', self.root) def test_runtime_self_check_rejects_wrong_versions_and_process_failures(self): self.manifest("executable") with mock.patch.object(self.builder.subprocess, "run", return_value=mock.Mock( stdout=json.dumps({"ok": True, "amane": {"version": "unknown"}}))): with self.assertRaises(ValueError): self.builder.verify(self.root) with mock.patch.object(self.builder.subprocess, "run", side_effect= subprocess.CalledProcessError(3, ["runtime.exe"])): with self.assertRaises(subprocess.CalledProcessError): self.builder.verify(self.root) def test_runtime_self_check_rejects_foreign_revision_and_path_escape(self): for change in ({"revision": "0" * 40}, {"executable": "../outside.exe"}, {"kind": "shell"}): facts = {**self.manifest("executable"), **change} (self.root / "manifest.json").write_text(json.dumps(facts), encoding="utf-8") with mock.patch.object(self.builder.subprocess, "run") as run: with self.assertRaises(ValueError): self.builder.verify(self.root) run.assert_not_called() def test_build_rejects_work_inside_the_distribution(self): with mock.patch.object(self.builder.subprocess, "run") as run: with self.assertRaises(ValueError): self.builder.build(self.root, work=self.root / "work") run.assert_not_called() def test_build_preserves_an_unowned_nonempty_destination(self): private_file = self.root / "keep.txt" private_file.write_text("keep", encoding="utf-8") with mock.patch.object(self.builder.subprocess, "run") as run: with self.assertRaises(ValueError): self.builder.build(self.root, work=self.root.parent / "work") run.assert_not_called() self.assertEqual(private_file.read_text(encoding="utf-8"), "keep") def test_docker_smoke_checks_the_bundled_runtime_and_matching_api_facts(self): from scripts import smoke_docker manifest = {**self.builder.locked_amane(), "kind": "python", "executable": ".venv/bin/python"} report = {"ok": True, "amane": {"version": manifest["version"]}, "sites": list(bridge.SITES)} description = {"python": "/app/amane-runtime/.venv/bin/python", "installed": True, "revision": manifest["revision"]} with mock.patch.object(smoke_docker, "docker", side_effect=[json.dumps(manifest), json.dumps(report)]) as docker, \ mock.patch.object(smoke_docker, "request", return_value=json.dumps(description)): smoke_docker.amane_checks("fixture") self.assertEqual(docker.call_args_list[1], mock.call( "exec", "fixture", "/app/amane-runtime/.venv/bin/python", "-X", "utf8", "/app/tools/amane-bridge/bridge.py", "--self-check")) def test_windows_application_and_lite_payloads_keep_the_complete_bridge(self): from peach.update_packages import TOOL_PATHS from scripts.package_windows_release import package, build_lite_installer payload = self.root / "Peach" (payload / "_internal").mkdir(parents=True) (payload / "Peach.exe").write_bytes(b"application") (payload / "_internal/standalone.txt").touch() runtime_files = {"amane-bridge.exe", "manifest.json", "LICENSE-amane.txt", "amane-corresponding-source.zip", "_internal/python314.dll"} runtime = payload / "_internal/tools/amane-runtime" for name in runtime_files: target = runtime / name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(name.encode()) for prefix, required in TOOL_PATHS.values(): for name in required: target = payload / prefix / name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"tool") release = self.root / "release" package(payload, release, "1.0.0") expected = {"Peach/_internal/tools/amane-runtime/" + name for name in runtime_files} for suffix in (".zip", "-update.zip"): with zipfile.ZipFile(release / f"Peach-1.0.0-windows-x64{suffix}") as archive: self.assertTrue(expected <= set(archive.namelist())) for tool in TOOL_PATHS: with zipfile.ZipFile(release / f"Peach-1.0.0-windows-x64-tools-{tool}.zip") as archive: self.assertFalse(expected & set(archive.namelist())) def compile_installer(command, **kwargs): source = Path(next(arg.removeprefix("/DSourceDir=") for arg in command if arg.startswith("/DSourceDir="))) for name in runtime_files: self.assertEqual((source / "_internal/tools/amane-runtime" / name).read_bytes(), name.encode()) self.assertIn("/DLiteInstaller=1", command) (release / "Peach-1.0.0-windows-x64-lite-setup.exe").write_bytes(b"installer") with mock.patch("scripts.package_windows_release.subprocess.run", side_effect=compile_installer): self.assertTrue(build_lite_installer(release, "1.0.0", self.root / "ISCC.exe").is_file()) if __name__ == "__main__": unittest.main()