feat(skills): 补充 gitlink-deps 配套脚本与测试
依审阅意见，将 SKILL.md 方式A引用的脚本（scripts/）与单元测试（tests/）一并纳入，使 PR 自包含可运行。
This commit is contained in:
parent
70e137c04f
commit
f5b378a42b
|
|
@ -0,0 +1,294 @@
|
|||
"""gitlink-deps：项目依赖追踪。
|
||||
|
||||
扫描一个 GitLink 仓库的依赖声明文件（go.mod / package.json /
|
||||
requirements.txt / pom.xml / Cargo.toml / pyproject.toml 等），
|
||||
解析出依赖清单、数量统计、技术栈识别与潜在风险提示，生成依赖报告。
|
||||
|
||||
数据来自 GitLink 公开 API（只读），无需登录。
|
||||
|
||||
用法：
|
||||
python deps.py --owner Gitlink --repo gitlink-cli
|
||||
python deps.py --owner Gitlink --repo gitlink-cli --format json
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from glapi import GitLinkClient, GitLinkError, split_owner_repo
|
||||
|
||||
# Windows 控制台默认 GBK，直接打印含 emoji 的 Markdown 会抛 UnicodeEncodeError。
|
||||
# 重配置 stdout 为 UTF-8，确保跨平台正常输出。
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 依赖文件 → 生态映射
|
||||
MANIFESTS = {
|
||||
"go.mod": "Go",
|
||||
"package.json": "Node.js",
|
||||
"requirements.txt": "Python",
|
||||
"pyproject.toml": "Python",
|
||||
"Pipfile": "Python",
|
||||
"pom.xml": "Java (Maven)",
|
||||
"build.gradle": "Java (Gradle)",
|
||||
"Cargo.toml": "Rust",
|
||||
"composer.json": "PHP",
|
||||
"Gemfile": "Ruby",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 各类清单解析器（纯函数，输入文本，输出依赖列表）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_go_mod(text: str) -> list[dict[str, str]]:
|
||||
"""解析 go.mod 的 require 块。"""
|
||||
deps: list[dict[str, str]] = []
|
||||
in_block = False
|
||||
for line in text.splitlines():
|
||||
s = line.strip()
|
||||
if s.startswith("require ("):
|
||||
in_block = True
|
||||
continue
|
||||
if in_block and s == ")":
|
||||
in_block = False
|
||||
continue
|
||||
# require 块内，或单行 require
|
||||
m = re.match(r"(?:require\s+)?([\w./\-]+)\s+(v[\w.\-+]+)", s)
|
||||
if m and ("/" in m.group(1)):
|
||||
deps.append({
|
||||
"name": m.group(1),
|
||||
"version": m.group(2),
|
||||
"indirect": "// indirect" in s,
|
||||
})
|
||||
return deps
|
||||
|
||||
|
||||
def parse_package_json(text: str) -> list[dict[str, str]]:
|
||||
"""解析 package.json 的 dependencies 与 devDependencies。"""
|
||||
deps: list[dict[str, str]] = []
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return deps
|
||||
for field, dev in (("dependencies", False), ("devDependencies", True)):
|
||||
block = data.get(field)
|
||||
if isinstance(block, dict):
|
||||
for name, ver in block.items():
|
||||
deps.append({"name": name, "version": str(ver), "indirect": dev})
|
||||
return deps
|
||||
|
||||
|
||||
def parse_requirements(text: str) -> list[dict[str, str]]:
|
||||
"""解析 requirements.txt。"""
|
||||
deps: list[dict[str, str]] = []
|
||||
for line in text.splitlines():
|
||||
s = line.strip()
|
||||
if not s or s.startswith("#") or s.startswith("-"):
|
||||
continue
|
||||
m = re.match(r"([A-Za-z0-9_.\-]+)\s*([=<>!~]=?.*)?", s)
|
||||
if m:
|
||||
deps.append({
|
||||
"name": m.group(1),
|
||||
"version": (m.group(2) or "").strip() or "*",
|
||||
"indirect": False,
|
||||
})
|
||||
return deps
|
||||
|
||||
|
||||
def parse_cargo_toml(text: str) -> list[dict[str, str]]:
|
||||
"""解析 Cargo.toml 的 [dependencies] 段（简化）。"""
|
||||
deps: list[dict[str, str]] = []
|
||||
in_deps = False
|
||||
for line in text.splitlines():
|
||||
s = line.strip()
|
||||
if s.startswith("["):
|
||||
in_deps = "dependencies" in s
|
||||
continue
|
||||
if in_deps and "=" in s and not s.startswith("#"):
|
||||
name = s.split("=", 1)[0].strip()
|
||||
ver_part = s.split("=", 1)[1].strip().strip('"')
|
||||
if name:
|
||||
deps.append({"name": name, "version": ver_part or "*", "indirect": False})
|
||||
return deps
|
||||
|
||||
|
||||
def parse_pom_xml(text: str) -> list[dict[str, str]]:
|
||||
"""解析 pom.xml 的 <dependency> 块（正则简化）。"""
|
||||
deps: list[dict[str, str]] = []
|
||||
for block in re.findall(r"<dependency>(.*?)</dependency>", text, re.DOTALL):
|
||||
gid = re.search(r"<groupId>(.*?)</groupId>", block)
|
||||
aid = re.search(r"<artifactId>(.*?)</artifactId>", block)
|
||||
ver = re.search(r"<version>(.*?)</version>", block)
|
||||
if aid:
|
||||
name = f"{gid.group(1)}:{aid.group(1)}" if gid else aid.group(1)
|
||||
deps.append({"name": name.strip(),
|
||||
"version": ver.group(1).strip() if ver else "*",
|
||||
"indirect": False})
|
||||
return deps
|
||||
|
||||
|
||||
PARSERS = {
|
||||
"go.mod": parse_go_mod,
|
||||
"package.json": parse_package_json,
|
||||
"requirements.txt": parse_requirements,
|
||||
"Cargo.toml": parse_cargo_toml,
|
||||
"pom.xml": parse_pom_xml,
|
||||
}
|
||||
|
||||
|
||||
def scan(owner: str, repo: str, ref: str = "master",
|
||||
client: GitLinkClient | None = None) -> dict[str, Any]:
|
||||
"""扫描仓库根目录的依赖文件并解析。"""
|
||||
client = client or GitLinkClient()
|
||||
|
||||
# 列根目录，找出存在的清单文件
|
||||
try:
|
||||
root_entries = client.list_dir(owner, repo, "", ref)
|
||||
except GitLinkError:
|
||||
root_entries = []
|
||||
root_names = {str(e.get("name", "")) for e in root_entries}
|
||||
|
||||
manifests_found: list[dict[str, Any]] = []
|
||||
all_deps: list[dict[str, Any]] = []
|
||||
ecosystems: set[str] = set()
|
||||
|
||||
for fname, eco in MANIFESTS.items():
|
||||
if fname not in root_names:
|
||||
continue
|
||||
ecosystems.add(eco)
|
||||
parser = PARSERS.get(fname)
|
||||
deps: list[dict[str, str]] = []
|
||||
if parser:
|
||||
content = client.file_content(owner, repo, fname, ref)
|
||||
if content:
|
||||
deps = parser(content)
|
||||
for d in deps:
|
||||
d["manifest"] = fname
|
||||
d["ecosystem"] = eco
|
||||
all_deps.extend(deps)
|
||||
manifests_found.append({
|
||||
"file": fname, "ecosystem": eco, "parsed": parser is not None,
|
||||
"count": len(deps),
|
||||
})
|
||||
|
||||
direct = [d for d in all_deps if not d.get("indirect")]
|
||||
indirect = [d for d in all_deps if d.get("indirect")]
|
||||
|
||||
return {
|
||||
"owner": owner, "repo": repo,
|
||||
"ecosystems": sorted(ecosystems),
|
||||
"manifests": manifests_found,
|
||||
"total_deps": len(all_deps),
|
||||
"direct_count": len(direct),
|
||||
"indirect_count": len(indirect),
|
||||
"dependencies": all_deps,
|
||||
"risks": _assess_risks(all_deps, manifests_found),
|
||||
}
|
||||
|
||||
|
||||
def _assess_risks(deps: list[dict[str, Any]], manifests: list[dict[str, Any]]) -> list[str]:
|
||||
"""基于依赖清单给出风险与改进提示。"""
|
||||
risks: list[str] = []
|
||||
if not manifests:
|
||||
risks.append("未发现依赖声明文件，无法分析依赖（可能是纯文档/资源仓库，或依赖文件不在根目录）。")
|
||||
return risks
|
||||
|
||||
# 未锁定版本的依赖
|
||||
unpinned = [d for d in deps if d.get("version") in ("*", "", "latest")
|
||||
or str(d.get("version", "")).startswith("^")
|
||||
or str(d.get("version", "")).startswith("~")]
|
||||
if unpinned:
|
||||
risks.append(f"有 {len(unpinned)} 个依赖未锁定精确版本（使用 ^ / ~ / * / latest），"
|
||||
"可能导致构建不可复现，建议在锁文件中固定版本。")
|
||||
|
||||
# 依赖数量过多
|
||||
direct = [d for d in deps if not d.get("indirect")]
|
||||
if len(direct) > 50:
|
||||
risks.append(f"直接依赖较多（{len(direct)} 个），建议定期审查是否都必要，减少供应链攻击面。")
|
||||
|
||||
if not risks:
|
||||
risks.append("未发现明显的依赖风险，依赖声明较为规范。")
|
||||
return risks
|
||||
|
||||
|
||||
def render_report(result: dict[str, Any]) -> str:
|
||||
"""渲染依赖报告（Markdown）。"""
|
||||
owner, repo = result["owner"], result["repo"]
|
||||
lines = [
|
||||
f"# 依赖追踪报告 — {owner}/{repo}",
|
||||
"",
|
||||
f"- 技术栈：{', '.join(result['ecosystems']) or '未识别'}",
|
||||
f"- 依赖声明文件：{len(result['manifests'])} 个",
|
||||
f"- 依赖总数：{result['total_deps']}（直接 {result['direct_count']} / 间接 {result['indirect_count']}）",
|
||||
"",
|
||||
]
|
||||
if result["manifests"]:
|
||||
lines += ["## 依赖声明文件", "", "| 文件 | 生态 | 解析依赖数 |", "|------|------|:----------:|"]
|
||||
for m in result["manifests"]:
|
||||
lines.append(f"| `{m['file']}` | {m['ecosystem']} | {m['count']} |")
|
||||
lines.append("")
|
||||
|
||||
direct = [d for d in result["dependencies"] if not d.get("indirect")]
|
||||
if direct:
|
||||
lines += ["## 直接依赖（前 30）", "", "| 依赖 | 版本 | 生态 |", "|------|------|------|"]
|
||||
for d in direct[:30]:
|
||||
lines.append(f"| `{d['name']}` | {d['version']} | {d['ecosystem']} |")
|
||||
if len(direct) > 30:
|
||||
lines.append(f"| … | 其余 {len(direct) - 30} 个 | |")
|
||||
lines.append("")
|
||||
|
||||
lines += ["## 风险与建议", ""]
|
||||
for i, r in enumerate(result["risks"], 1):
|
||||
lines.append(f"{i}. {r}")
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
p = argparse.ArgumentParser(prog="gitlink-deps", description="项目依赖追踪")
|
||||
p.add_argument("--owner", help="仓库所有者")
|
||||
p.add_argument("--repo", help="仓库名称")
|
||||
p.add_argument("--slug", help="owner/repo 或完整 URL")
|
||||
p.add_argument("--ref", default="master", help="分支或标签，默认 master")
|
||||
p.add_argument("--format", choices=["markdown", "json"], default="markdown")
|
||||
p.add_argument("--output", type=Path, help="报告输出文件")
|
||||
args = p.parse_args(argv)
|
||||
|
||||
if args.slug:
|
||||
owner, repo = split_owner_repo(args.slug)
|
||||
elif args.owner and args.repo:
|
||||
owner, repo = args.owner, args.repo
|
||||
else:
|
||||
print("错误：请用 --owner/--repo 或 --slug 指定仓库。", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
try:
|
||||
result = scan(owner, repo, ref=args.ref)
|
||||
except GitLinkError as exc:
|
||||
print(f"采集失败：{exc}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
out = (json.dumps(result, ensure_ascii=False, indent=2)
|
||||
if args.format == "json" else render_report(result))
|
||||
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(out, encoding="utf-8")
|
||||
print(f"已写入 {args.output}")
|
||||
else:
|
||||
print(out)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -0,0 +1,241 @@
|
|||
"""GitLink 公开 API 共享客户端。
|
||||
|
||||
供 gitlink-skills-pack 下各 Skill 的脚本复用。仅依赖 Python 标准库，
|
||||
无需第三方包，便于在受限环境或 Agent 沙箱中运行。
|
||||
|
||||
数据全部来自 GitLink 平台公开接口（https://www.gitlink.org.cn/api），
|
||||
默认无需 token；如需访问私有仓库，可传入 token。
|
||||
|
||||
所有方法均为只读，不修改任何远程数据。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
API_BASE = "https://www.gitlink.org.cn/api"
|
||||
USER_AGENT = "gitlink-skills-pack/1.0 (+https://www.gitlink.org.cn)"
|
||||
DEFAULT_TIMEOUT = 30
|
||||
COMMIT_PAGE_SIZE = 50 # GitLink commits 接口每页硬上限
|
||||
|
||||
|
||||
class GitLinkError(RuntimeError):
|
||||
"""API 调用中不可恢复的错误。"""
|
||||
|
||||
|
||||
class GitLinkClient:
|
||||
"""GitLink 公开数据接口客户端。
|
||||
|
||||
带可选文件缓存：同一资源重复读取不重复打网，对平台友好。
|
||||
"""
|
||||
|
||||
def __init__(self, base: str = API_BASE, token: str | None = None,
|
||||
timeout: int = DEFAULT_TIMEOUT, cache_dir: Path | None = None) -> None:
|
||||
self.base = base.rstrip("/")
|
||||
self.token = token
|
||||
self.timeout = timeout
|
||||
self.cache_dir = cache_dir
|
||||
if self.cache_dir:
|
||||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 底层请求
|
||||
# ------------------------------------------------------------------
|
||||
def _cache_path(self, url: str) -> Path | None:
|
||||
if not self.cache_dir:
|
||||
return None
|
||||
safe = urllib.parse.quote(url, safe="")
|
||||
return self.cache_dir / f"{safe}.json"
|
||||
|
||||
def get(self, path: str, query: dict[str, Any] | None = None) -> Any:
|
||||
"""GET 请求，返回解析后的 JSON（dict/list）或 None。"""
|
||||
url = f"{self.base}/{path.lstrip('/')}"
|
||||
if query:
|
||||
url = f"{url}?{urllib.parse.urlencode(query)}"
|
||||
|
||||
cache_path = self._cache_path(url)
|
||||
if cache_path and cache_path.exists():
|
||||
return json.loads(cache_path.read_text(encoding="utf-8"))
|
||||
|
||||
headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
|
||||
if self.token:
|
||||
headers["Authorization"] = f"Bearer {self.token}"
|
||||
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
||||
raw = resp.read().decode("utf-8", errors="replace")
|
||||
except urllib.error.HTTPError as exc:
|
||||
raise GitLinkError(f"HTTP {exc.code}: {url}") from exc
|
||||
except urllib.error.URLError as exc:
|
||||
raise GitLinkError(f"网络错误: {url} -> {exc.reason}") from exc
|
||||
|
||||
text = raw.strip()
|
||||
if not text or text in ("null", "{}", "[]"):
|
||||
data: Any = None
|
||||
elif text[0] in "{[":
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise GitLinkError(f"响应非 JSON: {url}") from exc
|
||||
else:
|
||||
raise GitLinkError(f"响应非 JSON（可能是 HTML）: {url}")
|
||||
|
||||
if cache_path is not None:
|
||||
cache_path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
||||
return data
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 资源访问（高层封装）
|
||||
# ------------------------------------------------------------------
|
||||
def repo_info(self, owner: str, repo: str) -> dict[str, Any]:
|
||||
"""仓库元信息。"""
|
||||
data = self.get(f"{owner}/{repo}.json")
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
def issues(self, owner: str, repo: str, limit: int = 50,
|
||||
page: int = 1) -> list[dict[str, Any]]:
|
||||
"""Issue 列表。"""
|
||||
data = self.get(f"{owner}/{repo}/issues.json", {"page": page, "limit": limit})
|
||||
return _extract_list(data, ("issues",))
|
||||
|
||||
def issue_detail(self, owner: str, repo: str, number: int) -> dict[str, Any]:
|
||||
"""单个 Issue 详情（含完整字段）。"""
|
||||
data = self.get(f"{owner}/{repo}/issues/{number}.json")
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
def pulls(self, owner: str, repo: str, limit: int = 50,
|
||||
page: int = 1) -> list[dict[str, Any]]:
|
||||
"""PR 列表。"""
|
||||
data = self.get(f"{owner}/{repo}/pulls.json", {"page": page, "limit": limit})
|
||||
return _extract_list(data, ("issues", "pulls"))
|
||||
|
||||
def contributors(self, owner: str, repo: str) -> list[dict[str, Any]]:
|
||||
"""贡献者列表。"""
|
||||
data = self.get(f"{owner}/{repo}/contributors.json")
|
||||
return _extract_list(data, ("list",))
|
||||
|
||||
def commits(self, owner: str, repo: str, max_pages: int = 4) -> list[dict[str, Any]]:
|
||||
"""提交列表（按需翻页，每页 50 条，以 total_count 为终止依据）。"""
|
||||
out: list[dict[str, Any]] = []
|
||||
total: int | None = None
|
||||
for page in range(1, max(1, max_pages) + 1):
|
||||
data = self.get(f"{owner}/{repo}/commits.json",
|
||||
{"page": page, "limit": COMMIT_PAGE_SIZE})
|
||||
if total is None and isinstance(data, dict):
|
||||
total = _safe_int(data.get("total_count")) or None
|
||||
page_items = _extract_list(data, ("commits",))
|
||||
if not page_items:
|
||||
break
|
||||
out.extend(page_items)
|
||||
if total is not None and len(out) >= total:
|
||||
break
|
||||
return out
|
||||
|
||||
def list_dir(self, owner: str, repo: str, path: str = "",
|
||||
ref: str = "master") -> list[dict[str, Any]]:
|
||||
"""列出目录下的条目（文件与子目录）。
|
||||
|
||||
返回的每个 entry 含 name / path / type(file|dir) / sha / size，
|
||||
文件类型的 entry 还可能直接带明文 content。
|
||||
"""
|
||||
data = self.get(f"{owner}/{repo}/sub_entries.json",
|
||||
{"filepath": path, "ref": ref})
|
||||
# 查询目录时 entries 为 list；查询单文件时 entries 为单个 dict。
|
||||
# 统一归一化为 list，便于下游处理。
|
||||
if isinstance(data, dict):
|
||||
entries = data.get("entries")
|
||||
if isinstance(entries, dict):
|
||||
return [entries]
|
||||
if isinstance(entries, list):
|
||||
return entries
|
||||
return _extract_list(data, ("entries",))
|
||||
|
||||
def file_content(self, owner: str, repo: str, filepath: str,
|
||||
ref: str = "master") -> str | None:
|
||||
"""读取单个文件的文本内容。
|
||||
|
||||
GitLink 的 sub_entries 接口对单文件查询会在 entries 中返回明文 content，
|
||||
据此取出。文件不存在或无内容时返回 None。
|
||||
"""
|
||||
entries = self.list_dir(owner, repo, filepath, ref)
|
||||
target = filepath.rsplit("/", 1)[-1]
|
||||
for entry in entries:
|
||||
if entry.get("type") == "file" and entry.get("name") == target:
|
||||
content = entry.get("content")
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
# 回退：部分情况下单文件查询 entries 仅一项
|
||||
if len(entries) == 1 and entries[0].get("type") == "file":
|
||||
content = entries[0].get("content")
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
return None
|
||||
|
||||
def readme(self, owner: str, repo: str, ref: str = "master") -> str | None:
|
||||
"""读取仓库 README（自动 base64 解码）。"""
|
||||
data = self.get(f"{owner}/{repo}/readme.json", {"ref": ref})
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
content = data.get("content")
|
||||
if not isinstance(content, str):
|
||||
return None
|
||||
# 注意：GitLink 的 readme.json 虽然 encoding 标为 base64，
|
||||
# 实测 content 多为明文 Markdown。先探测明文特征，命中则直接返回；
|
||||
# 否则再尝试 base64 解码。
|
||||
stripped = content.lstrip()
|
||||
if stripped.startswith(("#", "<", "[", "-", "*", "本", "这", "项")) or "\n" in content[:200]:
|
||||
return content
|
||||
try:
|
||||
raw = base64.b64decode(content.encode("ascii", "ignore"))
|
||||
decoded = raw.decode("utf-8", errors="replace")
|
||||
# 解码结果若不像文本（大量替换符），回退为原文
|
||||
if decoded.count("\ufffd") > len(decoded) * 0.1:
|
||||
return content
|
||||
return decoded
|
||||
except (ValueError, TypeError):
|
||||
return content
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------
|
||||
# 辅助
|
||||
# ----------------------------------------------------------------------------
|
||||
|
||||
def _extract_list(payload: Any, keys: tuple[str, ...]) -> list[Any]:
|
||||
"""从可能嵌套的响应中提取第一个匹配键的列表。"""
|
||||
if isinstance(payload, list):
|
||||
return payload
|
||||
if isinstance(payload, dict):
|
||||
for key in keys:
|
||||
value = payload.get(key)
|
||||
if isinstance(value, list):
|
||||
return value
|
||||
return []
|
||||
|
||||
|
||||
def _safe_int(value: Any, default: int = 0) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
|
||||
def split_owner_repo(slug: str) -> tuple[str, str]:
|
||||
"""把 'owner/repo' 或完整 URL 解析为 (owner, repo)。"""
|
||||
s = slug.strip()
|
||||
if s.startswith("http"):
|
||||
parts = urllib.parse.urlparse(s).path.strip("/").split("/")
|
||||
if len(parts) >= 2:
|
||||
return parts[0], parts[1].replace(".git", "")
|
||||
raise GitLinkError(f"无法从 URL 解析 owner/repo: {slug}")
|
||||
if "/" in s:
|
||||
owner, repo = s.split("/", 1)
|
||||
return owner, repo.replace(".git", "")
|
||||
raise GitLinkError(f"格式应为 owner/repo: {slug}")
|
||||
|
|
@ -0,0 +1,140 @@
|
|||
"""gitlink-deps 单元测试。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
||||
|
||||
import pytest
|
||||
|
||||
from deps import (
|
||||
parse_go_mod, parse_package_json, parse_requirements,
|
||||
parse_cargo_toml, parse_pom_xml, scan, render_report, _assess_risks,
|
||||
)
|
||||
|
||||
GO_MOD = """module github.com/example/proj
|
||||
|
||||
go 1.26.1
|
||||
|
||||
require (
|
||||
\tgithub.com/spf13/cobra v1.10.2
|
||||
\tgopkg.in/yaml.v3 v3.0.1
|
||||
)
|
||||
|
||||
require (
|
||||
\tgithub.com/danieljoos/wincred v1.2.3 // indirect
|
||||
)
|
||||
"""
|
||||
|
||||
PACKAGE_JSON = """{
|
||||
"name": "x",
|
||||
"dependencies": {"react": "^18.0.0", "axios": "1.6.0"},
|
||||
"devDependencies": {"jest": "^29.0.0"}
|
||||
}"""
|
||||
|
||||
REQUIREMENTS = """# comment
|
||||
requests==2.31.0
|
||||
flask>=2.0
|
||||
numpy
|
||||
-e .
|
||||
"""
|
||||
|
||||
|
||||
class TestGoMod:
|
||||
def test_parses_direct_and_indirect(self):
|
||||
deps = parse_go_mod(GO_MOD)
|
||||
names = {d["name"] for d in deps}
|
||||
assert "github.com/spf13/cobra" in names
|
||||
assert "gopkg.in/yaml.v3" in names
|
||||
indirect = [d for d in deps if d["indirect"]]
|
||||
assert any(d["name"] == "github.com/danieljoos/wincred" for d in indirect)
|
||||
|
||||
def test_version_extracted(self):
|
||||
deps = parse_go_mod(GO_MOD)
|
||||
cobra = next(d for d in deps if "cobra" in d["name"])
|
||||
assert cobra["version"] == "v1.10.2"
|
||||
|
||||
|
||||
class TestPackageJson:
|
||||
def test_deps_and_devdeps(self):
|
||||
deps = parse_package_json(PACKAGE_JSON)
|
||||
names = {d["name"] for d in deps}
|
||||
assert "react" in names and "axios" in names and "jest" in names
|
||||
jest = next(d for d in deps if d["name"] == "jest")
|
||||
assert jest["indirect"] is True # devDependency
|
||||
|
||||
def test_invalid_json(self):
|
||||
assert parse_package_json("{not json") == []
|
||||
|
||||
|
||||
class TestRequirements:
|
||||
def test_parses_pins(self):
|
||||
deps = parse_requirements(REQUIREMENTS)
|
||||
names = {d["name"] for d in deps}
|
||||
assert "requests" in names and "flask" in names and "numpy" in names
|
||||
req = next(d for d in deps if d["name"] == "requests")
|
||||
assert "2.31.0" in req["version"]
|
||||
|
||||
def test_skips_comments_and_flags(self):
|
||||
deps = parse_requirements(REQUIREMENTS)
|
||||
names = {d["name"] for d in deps}
|
||||
assert "-e" not in names
|
||||
|
||||
|
||||
class TestCargoToml:
|
||||
def test_parses_deps(self):
|
||||
text = '[package]\nname="x"\n[dependencies]\nserde = "1.0"\ntokio = "1.35"\n'
|
||||
deps = parse_cargo_toml(text)
|
||||
names = {d["name"] for d in deps}
|
||||
assert "serde" in names and "tokio" in names
|
||||
|
||||
|
||||
class TestPomXml:
|
||||
def test_parses_dependencies(self):
|
||||
text = """<project><dependencies>
|
||||
<dependency><groupId>org.junit</groupId><artifactId>junit</artifactId><version>5.0</version></dependency>
|
||||
</dependencies></project>"""
|
||||
deps = parse_pom_xml(text)
|
||||
assert deps[0]["name"] == "org.junit:junit"
|
||||
assert deps[0]["version"] == "5.0"
|
||||
|
||||
|
||||
class TestRisks:
|
||||
def test_no_manifest(self):
|
||||
risks = _assess_risks([], [])
|
||||
assert any("未发现依赖声明" in r for r in risks)
|
||||
|
||||
def test_unpinned_flagged(self):
|
||||
deps = [{"name": "react", "version": "^18.0.0", "indirect": False}]
|
||||
manifests = [{"file": "package.json", "count": 1}]
|
||||
risks = _assess_risks(deps, manifests)
|
||||
assert any("未锁定" in r for r in risks)
|
||||
|
||||
|
||||
class TestScan:
|
||||
class FakeClient:
|
||||
def list_dir(self, owner, repo, path, ref):
|
||||
if path == "":
|
||||
return [{"name": "go.mod", "type": "file"}]
|
||||
return []
|
||||
|
||||
def file_content(self, owner, repo, filepath, ref):
|
||||
return GO_MOD if filepath == "go.mod" else None
|
||||
|
||||
def test_scan_go_repo(self):
|
||||
r = scan("o", "r", client=self.FakeClient())
|
||||
assert "Go" in r["ecosystems"]
|
||||
assert r["total_deps"] >= 3
|
||||
assert r["direct_count"] >= 2
|
||||
|
||||
def test_report_renders(self):
|
||||
r = scan("o", "r", client=self.FakeClient())
|
||||
report = render_report(r)
|
||||
assert "依赖追踪报告" in report
|
||||
assert "cobra" in report
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(pytest.main([__file__, "-v"]))
|
||||
Loading…
Reference in New Issue