#!/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 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 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()