Leonid PershinandClaude Opus 5
60c7141f88
build / build (push) Successful in 24s
Многострочный python3 -c внутри блока run начинался с нулевого отступа
и обрывал YAML-скаляр: Gitea отказывалась читать файл целиком
("line 137: could not find expected ':'").
Генерация markdown уехала в tools/release_notes.py — в CI остался один
вызов без экранирования, а сам рендер покрыт тестами.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
444 lines
19 KiB
Python
444 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Тесты сборщика и синхронизатора: python -m unittest discover -s tools -t tools"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import ipaddress
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import build
|
|
import release_notes
|
|
import sync
|
|
from build import BuildError, Net, Options, Rule
|
|
|
|
|
|
def net(cidr: str) -> Net:
|
|
parsed = ipaddress.ip_network(cidr)
|
|
return Net(parsed.network_address.packed, parsed.prefixlen)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def data(ip: dict[str, str] | None = None, **files: str):
|
|
"""Подменяет data/ временным каталогом с заданными категориями."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / build.IP_SUBDIR).mkdir()
|
|
for name, content in files.items():
|
|
(root / f"{name.replace('_', '-')}.txt").write_text(content, encoding="utf-8")
|
|
for name, content in (ip or {}).items():
|
|
path = root / build.IP_SUBDIR / f"{name.replace('_', '-')}.txt"
|
|
path.write_text(content, encoding="utf-8")
|
|
original = build.DATA_DIR
|
|
build.DATA_DIR = root
|
|
try:
|
|
yield root
|
|
finally:
|
|
build.DATA_DIR = original
|
|
|
|
|
|
class ParseTest(unittest.TestCase):
|
|
def test_default_type_is_domain(self):
|
|
with data(a="example.com\n"):
|
|
self.assertEqual(build.resolve("a"), [Rule("domain", "example.com")])
|
|
|
|
def test_all_prefixes(self):
|
|
source = "domain:a.com\nfull:b.com\nkeyword:cc\nregexp:^d[0-9]+\\.com$\n"
|
|
with data(a=source):
|
|
self.assertEqual(
|
|
build.resolve("a"),
|
|
[
|
|
Rule("domain", "a.com"),
|
|
Rule("full", "b.com"),
|
|
Rule("keyword", "cc"),
|
|
Rule("regexp", r"^d[0-9]+\.com$"),
|
|
],
|
|
)
|
|
|
|
def test_comments_blank_lines_and_attributes(self):
|
|
with data(a="# заголовок\n\na.com @ads @cn # хвост\n"):
|
|
self.assertEqual(build.resolve("a"), [Rule("domain", "a.com", ("ads", "cn"))])
|
|
|
|
def test_normalisation(self):
|
|
with data(a="Example.COM.\nDOMAIN:пример.рф\n"):
|
|
values = [r.value for r in build.resolve("a")]
|
|
self.assertIn("example.com", values)
|
|
self.assertIn("xn--e1afmkfd.xn--p1ai", values)
|
|
|
|
def test_include_dedupes_and_sorts(self):
|
|
with data(a="include:b\nz.com\na.com\n", b="a.com\nfull:m.com\n"):
|
|
self.assertEqual(
|
|
build.resolve("a"),
|
|
[Rule("domain", "a.com"), Rule("domain", "z.com"), Rule("full", "m.com")],
|
|
)
|
|
|
|
def test_rejects_bad_input(self):
|
|
cases = {
|
|
"битый домен": "not_a_domain\n",
|
|
"битый regexp": "regexp:[unclosed\n",
|
|
"атрибут без @": "a.com ads\n",
|
|
"неизвестный include": "include:missing\n",
|
|
}
|
|
for label, source in cases.items():
|
|
with self.subTest(label), data(a=source):
|
|
with self.assertRaises(BuildError):
|
|
build.resolve("a")
|
|
|
|
def test_detects_include_cycle(self):
|
|
with data(a="include:b\n", b="include:a\n"):
|
|
with self.assertRaisesRegex(BuildError, "цикл"):
|
|
build.resolve("a")
|
|
|
|
|
|
class IPParseTest(unittest.TestCase):
|
|
def test_bare_address_becomes_host_route(self):
|
|
with data(ip={"a": "1.2.3.4\n2001:db8::1\n"}):
|
|
nets, options = build.resolve_ip("a")
|
|
self.assertEqual([str(n) for n in nets], ["1.2.3.4/32", "2001:db8::1/128"])
|
|
self.assertFalse(options.restricted)
|
|
|
|
def test_directives(self):
|
|
with data(ip={"a": "!network: udp\n!port: 50000-50099, 443\n1.2.3.0/24\n"}):
|
|
_, options = build.resolve_ip("a")
|
|
self.assertEqual(options.networks, ("udp",))
|
|
self.assertEqual(options.ports, ((443, 443), (50000, 50099)))
|
|
self.assertEqual(options.xray_ports, "443,50000-50099")
|
|
|
|
def test_adjacent_port_ranges_merge(self):
|
|
self.assertEqual(build.merge_ports([(10, 20), (21, 30), (100, 100)]), ((10, 30), (100, 100)))
|
|
|
|
def test_rejects_bad_input(self):
|
|
cases = {
|
|
"не сеть": "1.2.3.999\n",
|
|
"заданы хостовые биты": "1.2.3.4/24\n",
|
|
"чужой протокол": "!network: icmp\n1.2.3.4\n",
|
|
"порт вне диапазона": "!port: 70000\n1.2.3.4\n",
|
|
"неизвестная директива": "!ttl: 5\n1.2.3.4\n",
|
|
}
|
|
for label, source in cases.items():
|
|
with self.subTest(label), data(ip={"a": source}):
|
|
with self.assertRaises(BuildError):
|
|
build.resolve_ip("a")
|
|
|
|
def test_include_requires_matching_options(self):
|
|
with data(ip={"a": "!network: udp\ninclude:b\n", "b": "1.2.3.4\n"}):
|
|
with self.assertRaisesRegex(BuildError, "ограничения не совпадают"):
|
|
build.resolve_ip("a")
|
|
|
|
def test_include_with_same_options_is_allowed(self):
|
|
source = "!network: udp\n!port: 100-200\n"
|
|
with data(ip={"a": source + "include:b\n1.1.1.1\n", "b": source + "2.2.2.2\n"}):
|
|
nets, options = build.resolve_ip("a")
|
|
self.assertEqual([str(n) for n in nets], ["1.1.1.1/32", "2.2.2.2/32"])
|
|
self.assertEqual(options, Options(("udp",), ((100, 200),)))
|
|
|
|
|
|
class CollapseTest(unittest.TestCase):
|
|
def test_subdomains_are_dropped_under_suffix(self):
|
|
rules = [Rule("domain", "a.com"), Rule("full", "x.a.com"), Rule("domain", "deep.sub.a.com")]
|
|
self.assertEqual(build.collapse_rules(rules), [Rule("domain", "a.com")])
|
|
|
|
def test_attributed_rules_survive(self):
|
|
rules = [Rule("domain", "a.com"), Rule("full", "x.a.com", ("ads",))]
|
|
self.assertEqual(len(build.collapse_rules(rules)), 2)
|
|
|
|
def test_nested_networks_are_dropped(self):
|
|
nets = [net("10.0.0.0/8"), net("10.1.2.3/32"), net("11.0.0.0/8"), net("2001:db8::/32")]
|
|
collapsed = [str(n) for n in build.collapse_nets(nets)]
|
|
self.assertEqual(collapsed, ["10.0.0.0/8", "11.0.0.0/8", "2001:db8::/32"])
|
|
|
|
def test_collapse_happens_across_includes(self):
|
|
with data(a="include:b\nfull:node1.a.com\n", b="a.com\n"):
|
|
self.assertEqual(build.resolve("a"), [Rule("domain", "a.com")])
|
|
|
|
|
|
class ProtobufTest(unittest.TestCase):
|
|
def test_known_geosite_byte_layout(self):
|
|
blob = build.encode_geosite_dat({"t": [Rule("domain", "a.com")]})
|
|
# 0a 0e GeoSiteList.entry, 14 байт
|
|
# 0a 01 54 country_code = "T" (в .dat теги в верхнем регистре)
|
|
# 12 09 GeoSite.domain, 9 байт
|
|
# 08 02 type = Domain(2)
|
|
# 12 05 612e636f6d value = "a.com"
|
|
self.assertEqual(blob.hex(), "0a0e" "0a0154" "1209" "0802" "1205" "612e636f6d")
|
|
|
|
def test_known_geoip_byte_layout(self):
|
|
blob = build.encode_geoip_dat({"t": [net("1.2.3.0/24")]})
|
|
# 0a 0d GeoIPList.entry, 13 байт
|
|
# 0a 01 54 country_code = "T"
|
|
# 12 08 GeoIP.cidr, 8 байт
|
|
# 0a 04 01020300 ip = 1.2.3.0
|
|
# 10 18 prefix = 24
|
|
self.assertEqual(blob.hex(), "0a0d" "0a0154" "1208" "0a04" "01020300" "1018")
|
|
|
|
def test_keyword_type_is_omitted_as_proto3_default(self):
|
|
blob = build.encode_geosite_dat({"t": [Rule("keyword", "ip")]})
|
|
self.assertNotIn(b"\x08\x00", blob)
|
|
self.assertEqual(build.decode_geosite_dat(blob)["T"], [Rule("keyword", "ip")])
|
|
|
|
def test_round_trip_with_attributes(self):
|
|
categories = {
|
|
"one": [Rule("domain", "a.com", ("ads",)), Rule("full", "b.com")],
|
|
"two": [Rule("keyword", "ip"), Rule("regexp", r"^x\d+$")],
|
|
}
|
|
decoded = build.decode_geosite_dat(build.encode_geosite_dat(categories))
|
|
self.assertEqual(decoded, {k.upper(): v for k, v in categories.items()})
|
|
|
|
def test_geoip_round_trip_mixed_families(self):
|
|
categories = {"one": [net("0.0.0.0/0"), net("1.2.3.4/32")], "two": [net("2001:db8::/32")]}
|
|
decoded = build.decode_geoip_dat(build.encode_geoip_dat(categories))
|
|
self.assertEqual(decoded, {k.upper(): v for k, v in categories.items()})
|
|
|
|
|
|
class RenderTest(unittest.TestCase):
|
|
def test_singbox_expands_suffix_to_apex_plus_dot(self):
|
|
rendered = json.loads(build.render_singbox([Rule("domain", "a.com"), Rule("full", "b.com")]))
|
|
self.assertEqual(rendered["rules"][0]["domain"], ["a.com", "b.com"])
|
|
self.assertEqual(rendered["rules"][0]["domain_suffix"], [".a.com"])
|
|
|
|
def test_clash_yaml_skips_unsupported_types(self):
|
|
rules = [Rule("domain", "a.com"), Rule("full", "b.com"), Rule("keyword", "ip")]
|
|
payload = build.render_clash_yaml("t", rules)
|
|
self.assertIn("- '+.a.com'", payload)
|
|
self.assertIn("- 'b.com'", payload)
|
|
self.assertNotIn("ip", payload.split("payload:")[1])
|
|
|
|
def test_clash_list_keeps_all_types(self):
|
|
rules = [Rule("domain", "a.com"), Rule("keyword", "ip"), Rule("regexp", "^x$")]
|
|
body = build.render_clash_list("t", rules)
|
|
self.assertIn("DOMAIN-SUFFIX,a.com", body)
|
|
self.assertIn("DOMAIN-KEYWORD,ip", body)
|
|
self.assertIn("DOMAIN-REGEX,^x$", body)
|
|
|
|
def test_ip_singbox_carries_network_and_ports(self):
|
|
options = Options(("udp",), ((443, 443), (50000, 50099)))
|
|
rendered = json.loads(build.render_ip_singbox([net("1.2.3.0/24")], options))
|
|
rule = rendered["rules"][0]
|
|
self.assertEqual(rule["ip_cidr"], ["1.2.3.0/24"])
|
|
self.assertEqual(rule["network"], ["udp"])
|
|
self.assertEqual(rule["port"], [443])
|
|
self.assertEqual(rule["port_range"], ["50000:50099"])
|
|
|
|
def test_ip_clash_list_marks_family_and_no_resolve(self):
|
|
body = build.render_ip_clash_list("t", [net("1.2.3.0/24"), net("2001:db8::/32")], Options())
|
|
self.assertIn("IP-CIDR,1.2.3.0/24,no-resolve", body)
|
|
self.assertIn("IP-CIDR6,2001:db8::/32,no-resolve", body)
|
|
|
|
def test_xray_routing_orders_general_rules_before_restricted(self):
|
|
payload = json.loads(
|
|
build.render_xray_routing(
|
|
{"site": [Rule("domain", "a.com")]},
|
|
{
|
|
"voice": ([net("1.2.3.0/24")], Options(("udp",), ((50000, 50099),))),
|
|
"plain": ([net("4.4.4.0/24")], Options()),
|
|
},
|
|
"site.dat",
|
|
"ip.dat",
|
|
"proxy",
|
|
)
|
|
)
|
|
rules = payload["routing"]["rules"]
|
|
self.assertEqual(rules[0]["domain"], ["ext:site.dat:site"])
|
|
self.assertEqual(rules[1]["ip"], ["ext:ip.dat:plain"])
|
|
self.assertEqual(rules[2]["network"], "udp")
|
|
self.assertEqual(rules[2]["port"], "50000-50099")
|
|
self.assertTrue(all(r["outboundTag"] == "proxy" for r in rules))
|
|
|
|
def test_xray_routing_skips_categories_covered_by_another(self):
|
|
payload = json.loads(
|
|
build.render_xray_routing(
|
|
{"all": [Rule("domain", "a.com")], "part": [Rule("full", "x.a.com")]},
|
|
{
|
|
"all": ([net("10.0.0.0/8")], Options()),
|
|
"part": ([net("10.1.2.3/32")], Options()),
|
|
# Те же сети, но под ограничением — это другое правило.
|
|
"voice": ([net("10.0.0.0/8")], Options(("udp",), ((50000, 50099),))),
|
|
},
|
|
"site.dat",
|
|
"ip.dat",
|
|
"proxy",
|
|
)
|
|
)
|
|
rules = payload["routing"]["rules"]
|
|
self.assertEqual(
|
|
[r.get("domain", r.get("ip"))[0] for r in rules],
|
|
["ext:site.dat:all", "ext:ip.dat:all", "ext:ip.dat:voice"],
|
|
)
|
|
|
|
def test_essential_keeps_one_of_two_identical_categories(self):
|
|
items = {"b": [Rule("domain", "a.com")], "a": [Rule("domain", "a.com")]}
|
|
self.assertEqual(build.essential(items, build.collapse_rules), ["a"])
|
|
|
|
|
|
class SyncTest(unittest.TestCase):
|
|
def test_listable_fields_are_accepted(self):
|
|
payload = {
|
|
"version": 3,
|
|
"rules": [{"network": "udp", "ip_cidr": "1.2.3.0/24", "port_range": "50000:50099"}],
|
|
}
|
|
parsed = sync.parse_ruleset(payload, "t")
|
|
(options, nets), = parsed.groups.items()
|
|
self.assertEqual(options, Options(("udp",), ((50000, 50099),)))
|
|
self.assertEqual([str(n) for n in nets], ["1.2.3.0/24"])
|
|
|
|
def test_domain_fields_map_to_geosite_types(self):
|
|
payload = {
|
|
"version": 1,
|
|
"rules": [
|
|
{
|
|
"domain": ["exact.com"],
|
|
"domain_suffix": [".suffix.com", "bare.com"],
|
|
"domain_keyword": ["kw"],
|
|
"domain_regex": ["^r$"],
|
|
}
|
|
],
|
|
}
|
|
parsed = sync.parse_ruleset(payload, "t")
|
|
self.assertEqual(
|
|
sorted(str(r) for r in parsed.domains),
|
|
["bare.com", "full:exact.com", "keyword:kw", "regexp:^r$", "suffix.com"],
|
|
)
|
|
|
|
def test_logical_rules_are_flattened_and_invert_skipped(self):
|
|
payload = {
|
|
"version": 2,
|
|
"rules": [
|
|
{"type": "logical", "mode": "or", "rules": [{"domain_suffix": "inner.com"}]},
|
|
{"ip_cidr": "9.9.9.9/32", "invert": True},
|
|
],
|
|
}
|
|
parsed = sync.parse_ruleset(payload, "t")
|
|
self.assertEqual([str(r) for r in parsed.domains], ["inner.com"])
|
|
self.assertEqual(parsed.groups, {})
|
|
self.assertTrue(any("invert" in s for s in parsed.skipped))
|
|
|
|
def test_group_names_disambiguate_by_ports(self):
|
|
first = Options(("udp",), ((50000, 50099),))
|
|
second = Options(("udp",), ((19000, 20000),))
|
|
names = sync.group_names([first, second], "src")
|
|
self.assertEqual(names[first], "src-udp-50000-50099")
|
|
self.assertEqual(names[second], "src-udp-19000-20000")
|
|
|
|
def test_single_group_keeps_short_name(self):
|
|
options = Options(("udp",), ((50000, 50099),))
|
|
self.assertEqual(sync.group_names([options], "src"), {options: "src-udp"})
|
|
|
|
def test_handwritten_files_are_never_overwritten(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "manual.txt"
|
|
path.write_text("# мой список\na.com\n", encoding="utf-8")
|
|
self.assertFalse(sync.is_generated(path))
|
|
with self.assertRaisesRegex(sync.SyncError, "вручную"):
|
|
sync.write_generated(path, [], ["b.com"])
|
|
sync.write_generated(Path(tmp) / "new.txt", ["nota"], ["b.com"])
|
|
self.assertTrue(sync.is_generated(Path(tmp) / "new.txt"))
|
|
|
|
|
|
class ReleaseNotesTest(unittest.TestCase):
|
|
META = {
|
|
"geosite_dat_sha256": "aa",
|
|
"geoip_dat_sha256": "bb",
|
|
"categories": {"discord": 34},
|
|
"ip_categories": {
|
|
"discord": {"count": 140},
|
|
"discord-voice": {"count": 13, "network": ["udp"], "ports": "50000-50099"},
|
|
},
|
|
}
|
|
|
|
def test_renders_both_tables(self):
|
|
notes = release_notes.render(self.META, "abc1234")
|
|
self.assertIn("Собрано из `abc1234`.", notes)
|
|
self.assertIn("| discord | 34 |", notes)
|
|
self.assertIn("| discord | 140 | — |", notes)
|
|
self.assertIn("| discord-voice | 13 | udp 50000-50099 |", notes)
|
|
self.assertIn("sha256(discord-geoip.dat) = `bb`", notes)
|
|
|
|
def test_reads_real_metadata(self):
|
|
with tempfile.TemporaryDirectory() as out:
|
|
out_dir = Path(out) / "dist"
|
|
with data(a="a.com\n", ip={"n": "1.2.3.0/24\n"}):
|
|
build.build(out_dir, None, None, "proxy")
|
|
meta = json.loads((out_dir / "metadata.json").read_text(encoding="utf-8"))
|
|
self.assertIn("| n | 1 | — |", release_notes.render(meta, "abc1234"))
|
|
|
|
|
|
class CheckTest(unittest.TestCase):
|
|
def test_clean_sources_pass(self):
|
|
with data(a="a.com\nb.com\n", ip={"a": "1.2.3.0/24\n"}):
|
|
self.assertEqual(build.check(), 0)
|
|
|
|
def test_duplicate_is_reported(self):
|
|
with data(a="a.com\na.com\n"):
|
|
self.assertEqual(build.check(), 1)
|
|
|
|
def test_subdomain_covered_by_parent_is_reported(self):
|
|
with data(a="a.com\ndeep.sub.a.com\n"):
|
|
self.assertEqual(build.check(), 1)
|
|
|
|
def test_duplicate_network_is_reported(self):
|
|
with data(a="a.com\n", ip={"a": "1.2.3.4\n1.2.3.4/32\n"}):
|
|
self.assertEqual(build.check(), 1)
|
|
|
|
def test_network_covered_by_wider_one_is_reported(self):
|
|
with data(a="a.com\n", ip={"a": "10.0.0.0/8\n10.1.2.3\n"}):
|
|
self.assertEqual(build.check(), 1)
|
|
|
|
def test_real_repository_data_is_clean(self):
|
|
self.assertEqual(build.check(), 0)
|
|
|
|
|
|
class BuildTest(unittest.TestCase):
|
|
def test_produces_every_format(self):
|
|
expected = [
|
|
"discord-geosite.dat",
|
|
"discord-geoip.dat",
|
|
"geosite.dat",
|
|
"geoip.dat",
|
|
"discord-geoip.dat.sha256sum",
|
|
"metadata.json",
|
|
"xray/routing.json",
|
|
"txt/site/b.txt",
|
|
"src/site/b.txt",
|
|
"json/site/b.json",
|
|
"sing-box/site/b.json",
|
|
"clash/site/b.yaml",
|
|
"clash/site/b.list",
|
|
"adguard/b.txt",
|
|
"txt/ip/v.txt",
|
|
"src/ip/v.txt",
|
|
"json/ip/v.json",
|
|
"sing-box/ip/v.json",
|
|
"clash/ip/v.yaml",
|
|
"clash/ip/v.list",
|
|
]
|
|
sources = {"a": "a.com\n", "b": "include:a\nfull:c.com\n"}
|
|
nets = {"n": "1.2.3.0/24\n", "v": "!network: udp\n!port: 50000-50099\n5.6.7.0/24\n"}
|
|
with data(ip=nets, **sources), tempfile.TemporaryDirectory() as out:
|
|
out_dir = Path(out) / "dist"
|
|
self.assertEqual(build.build(out_dir, None, None, "proxy"), 0)
|
|
for rel in expected:
|
|
self.assertTrue((out_dir / rel).is_file(), rel)
|
|
self.assertEqual(
|
|
(out_dir / "geoip.dat").read_bytes(), (out_dir / "discord-geoip.dat").read_bytes()
|
|
)
|
|
meta = json.loads((out_dir / "metadata.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(meta["categories"], {"a": 1, "b": 2})
|
|
self.assertEqual(meta["ip_categories"]["n"], {"count": 1})
|
|
self.assertEqual(
|
|
meta["ip_categories"]["v"], {"count": 1, "network": ["udp"], "ports": "50000-50099"}
|
|
)
|
|
|
|
def test_source_dump_round_trips_through_parser(self):
|
|
"""src/ip/*.txt должен снова читаться сборщиком — вместе с директивами."""
|
|
options = Options(("udp",), ((50000, 50099),))
|
|
dumped = build.render_ip_source("v", [net("5.6.7.0/24")], options)
|
|
with data(a="a.com\n", ip={"v": dumped}):
|
|
self.assertEqual(build.resolve_ip("v"), ([net("5.6.7.0/24")], options))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|