asf/scripts/seed-extra-rels.py

#!/usr/bin/env python3
"""POST extra artifacts + 4 relazioni per densificare Traceability + Context Explorer.

Creati in tenant t-demo-003 / unico project esistente:
  1. DOC "Product Requirements.md" (DOCUMENT) --references--> REQ-103
  2. COMMIT "0abc123 feat: REQ-103" (CODE_COMMIT) --generated_from--> PR#88
  3. SEC "SEC-025 SQL injection input sanitization" (SECURITY_FINDING) --mitigated_by--> PR#88
  4. INC "INC-77 Login timeout" (INCIDENT) --caused_by--> DEP-44 deployment

Rileva artifact IDs esistenti via LIST + ext_id exact match.
"""
from __future__ import annotations

import json
import sys
import time
import urllib.request
import urllib.error
import urllib.parse

BASE = "http://localhost:8000/api/v1"
H = {
    "X-ASF-Tenant-Id": "t-demo-003",
    "X-ASF-User-Id": "u-mario",
    "Content-Type": "application/json",
}

LOOKUP = {
    "REQ": "REQ-103",
    "PR": "PR#88",
    "DEP": "DEP-44",
}


def req(method: str, path: str, body=None) -> tuple[int, object]:
    data = None
    if body is not None:
        data = json.dumps(body).encode()
    r = urllib.request.Request(BASE + path, data=data, method=method, headers=H)
    try:
        with urllib.request.urlopen(r, timeout=10) as resp:
            text = resp.read().decode()
            return resp.status, json.loads(text) if text else None
    except urllib.error.HTTPError as e:
        body = e.read().decode()
        print(f"  !! {method} {path} -> {e.code}: {body[:300]}", file=sys.stderr)
        return e.code, body


def list_artifacts() -> list[dict]:
    s, d = req("GET", "/artifacts?limit=500")
    assert s == 200 and isinstance(d, list), f"list artifacts failed: {s}"
    return d


def find_project_id(arts: list[dict]) -> str:
    """Unico project del tenant: take artifact project_id non-None first."""
    for a in arts:
        if a.get("project_id"):
            return a["project_id"]
    s, pl = req("GET", "/projects?limit=50")
    assert s == 200 and isinstance(pl, list) and pl
    return pl[0]["id"]


def create_artifact(payload: dict) -> dict:
    s, d = req("POST", "/artifacts", payload)
    if s != 201 or not isinstance(d, dict):
        raise RuntimeError(f"CREATE {payload.get('type')} {payload.get('external_id')} HTTP {s}")
    return d


def create_rel(src: str, tgt: str, rel_type: str) -> int:
    body = {
        "source_artifact_id": src,
        "target_artifact_id": tgt,
        "relationship_type": rel_type,
    }
    s, _ = req("POST", "/relationships", body)
    return s


def main() -> int:
    arts = list_artifacts()
    existing: dict[str, str] = {}
    for k, ext in LOOKUP.items():
        match = next((a for a in arts if a.get("external_id") == ext), None)
        if match is None:
            print(f"  !! missing {ext}, aborting.")
            return 2
        existing[k] = match["id"]
        print(f"  {k:3s} {ext!r:12s} -> {match['id'][:10]}…")

    pid = find_project_id(arts)
    print(f"  project id = {pid[:10]}…")

    print("\n== Creating 4 extra artifacts ==")
    extra = [
        {
            "project_id": pid,
            "type": "document",
            "external_id": "DOC-PRD-2026",
            "title": "Product Requirements v2.3",
            "status": "active",
            "content": "# Product Requirements 2.3\n\nLiving PRD covering authentication, billing and reporting.\n\n## Goals\nProvide clear tenancy and billing.\n",
            "metadata": {"format": "markdown", "page_count": 18, "owner": "pm"},
        },
        {
            "project_id": pid,
            "type": "commit",
            "external_id": "COMMIT-0abc123",
            "title": "feat(auth): implement REQ-103 SSO flow",
            "content": "commit 0abc123df\nAuthor: mario\nDate: today\n\nAdds Okta SAML callback + session refresh route.\n",
            "metadata": {"sha": "0abc123df", "author": "mario@asf.test", "repo": "asf/mono", "branch": "main"},
        },
        {
            "project_id": pid,
            "type": "security_finding",
            "external_id": "SEC-025",
            "title": "SQL injection risk in profile search (low confidence)",
            "status": "open",
            "content": "Input param q passed directly to $regex without sanitization. Scanner: semgrep. Confidence 82%.",
            "metadata": {"cwe": "CWE-89", "severity": "high", "scanner": "semgrep", "confidence_pct": 82},
        },
        {
            "project_id": pid,
            "type": "incident",
            "external_id": "INC-77",
            "title": "Login timeout spike 05:00-05:14 UTC",
            "status": "mitigated",
            "content": "Session refresh endpoint returned 5xx for 4% of users. MTTR 14 minutes. 27 customers impacted.",
            "metadata": {"severity": "sev2", "mttr_min": 14, "customers_impacted": 27},
        },
    ]
    ids: list[str] = []
    for pl in extra:
        d = create_artifact(pl)
        print(f"  [201] {pl['type']:15s} {pl['external_id']:18s} id={d['id'][:8]}…")
        ids.append(d["id"])

    doc_id, commit_id, sec_id, inc_id = ids
    print("\n== Creating 4 extra relationships ==")
    edges = [
        (doc_id, existing["REQ"], "documented_by", "DOC --documented_by--> REQ"),
        (commit_id, existing["PR"], "generated_from", "COMMIT --generated_from--> PR"),
        (sec_id, existing["PR"], "related_to", "SEC --related_to--> PR"),
        (inc_id, existing["DEP"], "caused_by", "INC --caused_by--> DEP"),
    ]
    all_ok = True
    for src, tgt, rt, label in edges:
        s = create_rel(src, tgt, rt)
        chk = "OK" if s == 201 else "FAIL"
        print(f"  [{s:3d}] {chk:4s} {label}")
        if s != 201:
            all_ok = False

    time.sleep(0.3)
    s, g = req("GET", f"/traceability/graph?project_id={pid}")
    if s == 200 and isinstance(g, dict):
        print(f"\n== Traceability graph now: nodes={len(g.get('nodes', []))} edges={len(g.get('edges', []))} ==")
    return 0 if all_ok else 5


if __name__ == "__main__":
    raise SystemExit(main())