mirror of
https://github.com/shadowsocks/ShadowsocksX-NG.git
synced 2026-11-08 11:44:50 +00:00
90 lines
4.5 KiB
Python
90 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Behavior checks at the maintainer conversion boundary (no network)."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
|
|
def converter(name):
|
|
path = Path(__file__).with_name(f"convert-{name}.py")
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class SnapshotConverterTests(unittest.TestCase):
|
|
def assert_slim(self, snapshot):
|
|
self.assertEqual(snapshot["schemaVersion"], 2)
|
|
self.assertNotIn("absorbed", snapshot)
|
|
for rule in snapshot["rules"]:
|
|
self.assertEqual(set(rule), {"action", "match"})
|
|
report = snapshot["lossReport"]
|
|
self.assertEqual(set(report), {"convertedCount", "absorbedCount", "skipped", "rejected"})
|
|
self.assertEqual(report["convertedCount"], len(snapshot["rules"]))
|
|
for category in ("skipped", "rejected"):
|
|
self.assertTrue(all(type(count) is int and count >= 0 for count in report[category].values()))
|
|
self.assertNotIn("shadowedException", report["skipped"])
|
|
|
|
def test_cn_absorption_drops_narrow_entries_and_reports_only_totals(self):
|
|
geo = converter("geolocation-cn")
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
snapshot = geo.convert([
|
|
{"type": geo.TYPE_DOMAIN, "value": "foo.cn"},
|
|
{"type": geo.TYPE_FULL, "value": "exact.cn"},
|
|
{"type": geo.TYPE_DOMAIN, "value": "example.com"},
|
|
{"type": geo.TYPE_REGEX, "value": "^example"},
|
|
])
|
|
self.assert_slim(snapshot)
|
|
self.assertEqual([rule["match"]["value"] for rule in snapshot["rules"]], ["cn", "example.com"])
|
|
self.assertEqual(snapshot["lossReport"]["absorbedCount"], 2)
|
|
self.assertEqual(snapshot["lossReport"]["skipped"], {"regexp": 1})
|
|
self.assertIn("synthesized-cn-suffix", output.getvalue())
|
|
self.assertIn("foo.cn", output.getvalue())
|
|
self.assertNotIn("foo.cn", json.dumps(snapshot))
|
|
|
|
def test_gfw_shadowed_exception_counts_once_and_keeps_raw_evidence_on_stdout(self):
|
|
gfw = converter("gfwlist")
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
rules, report = gfw.convert_document(
|
|
"||Example.COM^\n@@||Safe.Example.COM^\n@@||other.example\n|https://path.example/a\n")
|
|
snapshot = {"schemaVersion": gfw.SCHEMA_VERSION, "rules": rules, "lossReport": report}
|
|
self.assert_slim(snapshot)
|
|
self.assertEqual(report["absorbedCount"], 1)
|
|
self.assertEqual(report["skipped"], {"urlPrefix": 1})
|
|
self.assertEqual([rule["match"]["value"] for rule in rules], ["example.com", "other.example"])
|
|
self.assertIn("@@||Safe.Example.COM^ shadowed-by ||Example.COM^", output.getvalue())
|
|
self.assertNotIn("Safe.Example.COM", json.dumps(snapshot))
|
|
|
|
def test_china_ip_build_retains_metadata_normalization_and_duplicate_counts(self):
|
|
china = converter("china-ipv4")
|
|
document = "\n".join(f"1.0.{index}.1/24" for index in range(100)) + "\n1.0.0.2/24\n"
|
|
source = {"kind": "china-ipv4", "upstreamVersion": "fixed", "label": "China"}
|
|
snapshot = china.build_snapshot(document, source, "fixed", "MIT", "author", None)
|
|
self.assert_slim(snapshot)
|
|
self.assertEqual(snapshot["metadata"]["source"], source)
|
|
self.assertEqual(snapshot["metadata"]["converterVersion"], "2.0.0")
|
|
self.assertEqual(snapshot["rules"][0]["match"]["value"], "1.0.0.0/24")
|
|
self.assertEqual(snapshot["lossReport"]["skipped"], {"duplicate": 1})
|
|
|
|
def test_shipped_artifacts_use_compact_schema_and_preserve_pinned_rule_counts(self):
|
|
for name, count, absorbed in [("geolocation-cn", 4241, 1138), ("china-ipv4", 6206, 0), ("gfwlist", 4046, 31)]:
|
|
with self.subTest(source=name):
|
|
data = (Path(__file__).parent.parent / "Vendor" / "rules" / name / "snapshot.json").read_bytes()
|
|
self.assertEqual(data.count(b"\n"), 1)
|
|
snapshot = json.loads(data)
|
|
self.assert_slim(snapshot)
|
|
self.assertEqual(len(snapshot["rules"]), count)
|
|
self.assertEqual(snapshot["lossReport"]["absorbedCount"], absorbed)
|
|
self.assertEqual(snapshot["metadata"]["source"]["kind"], name)
|
|
self.assertEqual(snapshot["metadata"]["converterVersion"], "2.0.0")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|