196 lines
7.2 KiB
Python
196 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from fastapi.testclient import TestClient
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parent.parent
|
|
if str(ROOT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(ROOT_DIR))
|
|
|
|
|
|
def _write_sample_source(path: Path) -> None:
|
|
path.write_text(
|
|
"\n".join(
|
|
[
|
|
"proxies:",
|
|
" - name: hk-demo",
|
|
" type: ss",
|
|
" server: 1.1.1.1",
|
|
" port: 443",
|
|
" cipher: aes-128-gcm",
|
|
" password: demo-pass",
|
|
" - name: us-demo",
|
|
" type: ss",
|
|
" server: 2.2.2.2",
|
|
" port: 443",
|
|
" cipher: aes-128-gcm",
|
|
" password: demo-pass",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _load_yaml(text: str) -> dict[str, Any]:
|
|
data = yaml.safe_load(text)
|
|
if not isinstance(data, dict):
|
|
raise ValueError("response is not a yaml mapping")
|
|
return data
|
|
|
|
|
|
def _list_names(items: list[dict[str, Any]] | None) -> list[str]:
|
|
names: list[str] = []
|
|
for item in items or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
name = item.get("name")
|
|
if isinstance(name, str) and name.strip():
|
|
names.append(name)
|
|
return names
|
|
|
|
|
|
def _mapping_keys(mapping: dict[str, Any] | None) -> list[str]:
|
|
if not isinstance(mapping, dict):
|
|
return []
|
|
return sorted(mapping.keys())
|
|
|
|
|
|
def _group_map(groups: list[dict[str, Any]] | None) -> dict[str, dict[str, Any]]:
|
|
result: dict[str, dict[str, Any]] = {}
|
|
for group in groups or []:
|
|
if not isinstance(group, dict):
|
|
continue
|
|
name = group.get("name")
|
|
if isinstance(name, str) and name.strip():
|
|
result[name] = group
|
|
return result
|
|
|
|
|
|
def _extract_summary(data: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"proxy_provider_keys": _mapping_keys(data.get("proxy-providers")),
|
|
"rule_provider_keys": _mapping_keys(data.get("rule-providers")),
|
|
"proxy_group_names": _list_names(data.get("proxy-groups")),
|
|
"proxy_names": _list_names(data.get("proxies")),
|
|
"rules_count": len(data.get("rules") or []),
|
|
"rules_head": list(data.get("rules") or [])[:5],
|
|
"rules_tail": list(data.get("rules") or [])[-5:],
|
|
}
|
|
|
|
|
|
def _compare_group_members(legacy_groups: list[dict[str, Any]], conf_groups: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
mismatches: list[dict[str, Any]] = []
|
|
legacy_map = _group_map(legacy_groups)
|
|
conf_map = _group_map(conf_groups)
|
|
for name in sorted(set(legacy_map) & set(conf_map)):
|
|
legacy = legacy_map[name]
|
|
conf = conf_map[name]
|
|
legacy_members = legacy.get("proxies") or legacy.get("use") or []
|
|
conf_members = conf.get("proxies") or conf.get("use") or []
|
|
if legacy_members != conf_members:
|
|
mismatches.append(
|
|
{
|
|
"group": name,
|
|
"legacy_members": legacy_members,
|
|
"conf_members": conf_members,
|
|
}
|
|
)
|
|
return mismatches
|
|
|
|
|
|
def compare_outputs(client_type: str, profile_key: str, sources: str, *, sample_source: Path | None) -> dict[str, Any]:
|
|
if sample_source is not None:
|
|
os.environ["AIRPORT_A_URL"] = str(sample_source)
|
|
os.environ["AIRPORT_B_URL"] = str(sample_source)
|
|
os.environ["AIRPORT_C_URL"] = str(sample_source)
|
|
|
|
from app.main import PUBLIC_PREFIX, app
|
|
|
|
client = TestClient(app)
|
|
legacy_thin = client.get(f"{PUBLIC_PREFIX}/clients/{client_type}.yaml?sources={sources}")
|
|
legacy_bundle = client.get(f"{PUBLIC_PREFIX}/bundle/{client_type}.yaml?sources={sources}&force_refresh=true")
|
|
conf_thin = client.get(f"{PUBLIC_PREFIX}/conf/clients/{client_type}/{profile_key}.yaml")
|
|
conf_bundle = client.get(f"{PUBLIC_PREFIX}/conf/bundle/{client_type}/{profile_key}.yaml?force_refresh=true")
|
|
|
|
responses = {
|
|
"legacy_thin": legacy_thin,
|
|
"legacy_bundle": legacy_bundle,
|
|
"conf_thin": conf_thin,
|
|
"conf_bundle": conf_bundle,
|
|
}
|
|
for name, response in responses.items():
|
|
if response.status_code != 200:
|
|
raise RuntimeError(f"{name} returned {response.status_code}: {response.text}")
|
|
|
|
legacy_thin_yaml = _load_yaml(legacy_thin.text)
|
|
legacy_bundle_yaml = _load_yaml(legacy_bundle.text)
|
|
conf_thin_yaml = _load_yaml(conf_thin.text)
|
|
conf_bundle_yaml = _load_yaml(conf_bundle.text)
|
|
|
|
result = {
|
|
"inputs": {
|
|
"client_type": client_type,
|
|
"profile_key": profile_key,
|
|
"sources": sources,
|
|
},
|
|
"thin": {
|
|
"legacy": _extract_summary(legacy_thin_yaml),
|
|
"conf": _extract_summary(conf_thin_yaml),
|
|
"group_member_mismatches": _compare_group_members(
|
|
legacy_thin_yaml.get("proxy-groups") or [],
|
|
conf_thin_yaml.get("proxy-groups") or [],
|
|
),
|
|
},
|
|
"bundle": {
|
|
"legacy": _extract_summary(legacy_bundle_yaml),
|
|
"conf": _extract_summary(conf_bundle_yaml),
|
|
"group_member_mismatches": _compare_group_members(
|
|
legacy_bundle_yaml.get("proxy-groups") or [],
|
|
conf_bundle_yaml.get("proxy-groups") or [],
|
|
),
|
|
},
|
|
}
|
|
result["thin"]["proxy_provider_match"] = result["thin"]["legacy"]["proxy_provider_keys"] == result["thin"]["conf"]["proxy_provider_keys"]
|
|
result["thin"]["rule_provider_match"] = result["thin"]["legacy"]["rule_provider_keys"] == result["thin"]["conf"]["rule_provider_keys"]
|
|
result["thin"]["group_name_match"] = result["thin"]["legacy"]["proxy_group_names"] == result["thin"]["conf"]["proxy_group_names"]
|
|
result["bundle"]["group_name_match"] = result["bundle"]["legacy"]["proxy_group_names"] == result["bundle"]["conf"]["proxy_group_names"]
|
|
result["bundle"]["proxy_count_match"] = len(result["bundle"]["legacy"]["proxy_names"]) == len(result["bundle"]["conf"]["proxy_names"])
|
|
result["bundle"]["rules_count_match"] = result["bundle"]["legacy"]["rules_count"] == result["bundle"]["conf"]["rules_count"]
|
|
return result
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Compare legacy YAML routes with conf routes")
|
|
parser.add_argument("--client-type", default="mihomo")
|
|
parser.add_argument("--profile", default="default")
|
|
parser.add_argument("--sources", default="airport-a,airport-b")
|
|
parser.add_argument("--sample-source", default="")
|
|
args = parser.parse_args()
|
|
|
|
if args.sample_source:
|
|
sample_source = Path(args.sample_source).resolve()
|
|
else:
|
|
sample_source = None
|
|
|
|
if sample_source is None:
|
|
with TemporaryDirectory() as tmp:
|
|
generated = Path(tmp) / "sample-proxies.yaml"
|
|
_write_sample_source(generated)
|
|
result = compare_outputs(args.client_type, args.profile, args.sources, sample_source=generated)
|
|
else:
|
|
result = compare_outputs(args.client_type, args.profile, args.sources, sample_source=sample_source)
|
|
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|