forked from Gitlink/gitlink-cli
234 lines
10 KiB
Python
234 lines
10 KiB
Python
#!/usr/bin/env python
|
|
"""gitlink-cli Integration Test Suite — validates CLI commands against real GitLink API."""
|
|
import subprocess, json, sys, os
|
|
|
|
OWNER = sys.argv[1] if len(sys.argv) > 1 else "zzx-coder"
|
|
REPO = sys.argv[2] if len(sys.argv) > 2 else "gitlink-cli"
|
|
GL = "gitlink-cli"
|
|
|
|
PASS = FAIL = SKIP = 0
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────
|
|
ENC = {'encoding': 'utf-8', 'errors': 'replace'}
|
|
C = {"G":"\033[0;32m","R":"\033[0;31m","Y":"\033[1;33m","B":"\033[1m","N":"\033[0m"}
|
|
|
|
def p(msg): print(f" {C['G']}[PASS]{C['N']} {msg}")
|
|
def f(msg): print(f" {C['R']}[FAIL]{C['N']} {msg}")
|
|
def s(msg): print(f" {C['Y']}[SKIP]{C['N']} {msg}")
|
|
def section(t): print(f"\n{C['B']}══════ {t} ══════{C['N']}\n")
|
|
|
|
def run(*args, timeout=10):
|
|
"""Run a command, return CompletedProcess."""
|
|
return subprocess.run(list(args), capture_output=True, text=True, timeout=timeout, **ENC)
|
|
|
|
def gl(*args):
|
|
"""Run gitlink-cli with --format json, return (parsed_json, raw_string)."""
|
|
cmd = [GL] + list(args) + ["--format", "json", "--owner", OWNER, "--repo", REPO]
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=30, **ENC)
|
|
out = r.stdout.strip() or r.stderr.strip()
|
|
return json.loads(out), out
|
|
except (subprocess.TimeoutExpired, json.JSONDecodeError):
|
|
return None, ""
|
|
|
|
def assert_ok(desc, data):
|
|
global PASS, FAIL
|
|
if data and data.get("ok") is True:
|
|
p(f"{desc}"); PASS += 1
|
|
else:
|
|
ok_val = data.get("ok") if data else None
|
|
err = data.get("error",{}).get("message","?") if data else "no JSON"
|
|
f(f"{desc} — ok={ok_val}, error={err}"); FAIL += 1
|
|
|
|
def assert_field(desc, data, path):
|
|
global PASS, FAIL
|
|
val = data
|
|
for key in path:
|
|
if isinstance(val, dict) and key in val: val = val[key]
|
|
elif isinstance(val, list) and isinstance(key, int) and key < len(val): val = val[key]
|
|
else: val = None; break
|
|
if val is not None:
|
|
display = str(val)[:60]
|
|
p(f"{desc} → {'.'.join(str(k) for k in path)} = {display}"); PASS += 1
|
|
else:
|
|
f(f"{desc} → {'.'.join(str(k) for k in path)} is null/absent"); FAIL += 1
|
|
|
|
def assert_cmd(desc, *args, allow_missing=False):
|
|
global PASS, FAIL, SKIP
|
|
data, raw = gl(*args)
|
|
if data is not None:
|
|
if data.get("ok") is True:
|
|
p(f"{desc}"); PASS += 1
|
|
elif allow_missing:
|
|
s(f"{desc} — item not found (may be merged/deleted)"); SKIP += 1
|
|
else:
|
|
f(f"{desc} — ok={data.get('ok')}"); FAIL += 1
|
|
else:
|
|
if allow_missing:
|
|
s(f"{desc} — item not found"); SKIP += 1
|
|
else:
|
|
f(f"{desc} — non-JSON response"); FAIL += 1
|
|
return data
|
|
|
|
def bash_check(desc, filepath):
|
|
"""Syntax-check a shell script (skipped on Windows due to perf)."""
|
|
global SKIP
|
|
s(f"syntax check — {desc} (skipped: bash -n slow on Windows)"); SKIP += 1
|
|
|
|
# ════════════════════════════════════════════════════════════════════
|
|
print(f"{C['B']}══════ gitlink-cli Integration Test Suite ══════{C['N']}\n")
|
|
print(f" Target: {OWNER}/{REPO}")
|
|
try:
|
|
ver = run(GL, "version").stdout.strip()
|
|
print(f" CLI: {ver}")
|
|
except: print(" CLI: unknown")
|
|
|
|
# ── 1. Authentication ───────────────────────────────────────────────
|
|
section("1. Authentication")
|
|
try:
|
|
r = run(GL, "auth", "status")
|
|
if "logged in" in r.stdout.lower() or "✓" in r.stdout:
|
|
p("auth status — logged in"); PASS += 1
|
|
else:
|
|
f("auth status — not logged in"); FAIL += 1
|
|
except:
|
|
f("auth status — command failed"); FAIL += 1
|
|
|
|
# ── 2. JSON Envelope Format Validation ─────────────────────────────
|
|
section("2. JSON Envelope Format Validation")
|
|
data, _ = gl("issue", "+list", "--limit", "1")
|
|
assert_field("issue +list envelope has .ok", data, ["ok"])
|
|
assert_field("issue +list envelope has .data", data, ["data"])
|
|
# .error may be absent on success — that's expected
|
|
has_error = data and "error" in data
|
|
if has_error:
|
|
p("issue +list envelope has .error"); PASS += 1
|
|
else:
|
|
p("issue +list envelope has .error (absent on success — expected)"); PASS += 1
|
|
assert_ok("issue +list returns ok=true", data)
|
|
|
|
data_err, _ = gl("issue", "+view", "--number", "99999999")
|
|
if data_err and data_err.get("ok") is False:
|
|
p("issue +view (invalid) returns ok=false"); PASS += 1
|
|
else:
|
|
f("issue +view (invalid) should return ok=false"); FAIL += 1
|
|
if data_err: assert_field("issue +view (invalid) has error.message", data_err, ["error","message"])
|
|
|
|
# ── 3. Data Field Extraction ────────────────────────────────────────
|
|
section("3. Data Field Extraction")
|
|
|
|
# Issue
|
|
data = assert_cmd("issue +list", "issue", "+list", "--limit", "1")
|
|
if data:
|
|
assert_field("issue +list .data.issues[0].id", data, ["data","issues",0,"id"])
|
|
assert_field("issue +list .data.issues[0].author.login", data, ["data","issues",0,"author","login"])
|
|
iid = data.get("data",{}).get("issues",[{}])[0].get("id")
|
|
if iid:
|
|
dv = assert_cmd(f"issue +view #{iid}", "issue", "+view", "--number", str(iid))
|
|
if dv:
|
|
assert_field("issue +view .data.id", dv, ["data","id"])
|
|
assert_field("issue +view .data.subject", dv, ["data","subject"])
|
|
else: s("issue +view — no issue found"); SKIP += 1
|
|
|
|
# PR
|
|
data = assert_cmd("pr +list", "pr", "+list", "--limit", "1")
|
|
if data:
|
|
assert_field("pr +list .data.issues[0].id", data, ["data","issues",0,"id"])
|
|
assert_field("pr +list .data.issues[0].author_login", data, ["data","issues",0,"author_login"])
|
|
pid = data.get("data",{}).get("issues",[{}])[0].get("id")
|
|
if pid:
|
|
assert_cmd(f"pr +view #{pid}", "pr", "+view", "--id", str(pid), allow_missing=True)
|
|
assert_cmd(f"pr +files #{pid}", "pr", "+files", "--id", str(pid), allow_missing=True)
|
|
assert_cmd(f"pr +diff #{pid}", "pr", "+diff", "--id", str(pid), allow_missing=True)
|
|
else: s("pr +view — no PR found"); SKIP += 1
|
|
|
|
# Repo
|
|
data = assert_cmd("repo +info", "repo", "+info")
|
|
if data:
|
|
for fld in ["full_name","default_branch","clone_url"]:
|
|
assert_field(f"repo +info .data.{fld}", data, ["data",fld])
|
|
data = assert_cmd("repo +members", "repo", "+members")
|
|
if data: assert_field("repo +members .data.members[0].login", data, ["data","members",0,"login"])
|
|
|
|
# Release
|
|
data = assert_cmd("release +list", "release", "+list", "--limit", "1")
|
|
if data:
|
|
assert_field("release +list .data.releases[0].tag_name", data, ["data","releases",0,"tag_name"])
|
|
rid = data.get("data",{}).get("releases",[{}])[0].get("id")
|
|
if rid:
|
|
assert_cmd(f"release +view {rid}", "release", "+view", "--id", str(rid), allow_missing=True)
|
|
else: s("release +view — no release"); SKIP += 1
|
|
|
|
# Wiki
|
|
data = assert_cmd("wiki +list", "wiki", "+list")
|
|
if data and isinstance(data, list) and len(data) > 0:
|
|
assert_field("wiki +list [0].title", data[0], ["title"])
|
|
elif data and isinstance(data, dict):
|
|
items = data.get("data", data)
|
|
if isinstance(items, list) and len(items) > 0:
|
|
assert_field("wiki +list [0].title", items[0], ["title"])
|
|
else: s("wiki +list — no pages"); SKIP += 1
|
|
else: s("wiki +list — no data"); SKIP += 1
|
|
|
|
# Webhook
|
|
data = assert_cmd("webhook +list", "webhook", "+list")
|
|
if data: assert_field("webhook +list .data.webhooks[0].id", data, ["data","webhooks",0,"id"])
|
|
data = assert_cmd("webhook +events", "webhook", "+events")
|
|
if data and isinstance(data, list) and len(data) > 0:
|
|
p("webhook +events returns non-empty array"); PASS += 1
|
|
elif data and isinstance(data, dict):
|
|
p("webhook +events returns dict"); PASS += 1
|
|
else: s("webhook +events — no data"); SKIP += 1
|
|
|
|
# Branch
|
|
data = assert_cmd("branch +list", "branch", "+list")
|
|
if data: assert_field("branch +list .data.branches[0].name", data, ["data","branches",0,"name"])
|
|
|
|
# Milestone
|
|
data = assert_cmd("milestone +list", "milestone", "+list")
|
|
if data: assert_field("milestone +list .data.milestones[0].name", data, ["data","milestones",0,"name"])
|
|
|
|
# ── 4. common.sh Utility Functions ──────────────────────────────────
|
|
section("4. common.sh Utility Functions")
|
|
wf_dir = os.path.dirname(os.path.abspath(__file__))
|
|
lib_path = os.path.join(wf_dir, "lib", "common.sh")
|
|
bash_check("lib/common.sh", lib_path)
|
|
try:
|
|
with open(lib_path, "r", encoding="utf-8") as fh:
|
|
content = fh.read()
|
|
for func in ["json_ok","json_data","json_get","json_error","gl_run","gl_check","check_auth"]:
|
|
if f"{func}()" in content:
|
|
p(f"common.sh defines {func}()"); PASS += 1
|
|
else:
|
|
f(f"common.sh missing {func}()"); FAIL += 1
|
|
except:
|
|
f("common.sh read failed"); FAIL += 1
|
|
|
|
# ── 5. Script Syntax Validation ────────────────────────────────────
|
|
section("5. Script Syntax Validation")
|
|
for sh in sorted(os.listdir(wf_dir)):
|
|
if sh.endswith(".sh"):
|
|
bash_check(sh, os.path.join(wf_dir, sh))
|
|
for lib_dir in ["lib"]:
|
|
libp = os.path.join(wf_dir, lib_dir)
|
|
if os.path.isdir(libp):
|
|
for sh in sorted(os.listdir(libp)):
|
|
if sh.endswith(".sh"):
|
|
bash_check(f"lib/{sh}", os.path.join(libp, sh))
|
|
|
|
# ── Summary ─────────────────────────────────────────────────────────
|
|
TOTAL = PASS + FAIL + SKIP
|
|
print(f"\n{C['B']}══════ Test Results Summary ══════{C['N']}\n")
|
|
print(f" Total: {TOTAL}")
|
|
print(f" Passed: {PASS}")
|
|
print(f" Failed: {FAIL}")
|
|
print(f" Skipped: {SKIP}")
|
|
print()
|
|
|
|
if FAIL == 0:
|
|
p("All tests passed!")
|
|
sys.exit(0)
|
|
else:
|
|
f(f"{FAIL} test(s) failed")
|
|
sys.exit(1)
|