mirror of
https://github.com/brunopostle/ifcurl.git
synced 2026-07-12 10:18:14 +00:00
bcf_api: implement BCF 3.0 REST API as Python routes in the preview service
Stateless translation layer over Forgejo issues and comments. Auth is forwarded verbatim (Authorization: Bearer), so Forgejo enforces per-user permissions with no service token. Viewpoints are ifc:// URLs stored in comment bodies; GUIDs are deterministically derived from Forgejo IDs and are reversible without a lookup table. Endpoints: GET/POST projects, topics, comments, viewpoints; PUT topic; GET viewpoint snapshot (proxied through preview service). Also adds ifc_url_to_bcf_viewpoint() to bcf.py (symmetric counterpart of bcf_viewpoint_to_ifc_url), and fixes rate-limit state leaking between tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
fea6b37beb
commit
90f003a587
6 changed files with 755 additions and 1 deletions
23
README.md
23
README.md
|
|
@ -137,8 +137,29 @@ ifcurl serve --host 0.0.0.0 --port 9000 --allowed-hosts git.example.com
|
|||
| `GET /select?url=ifc://…` | Resolve a complex selector server-side; returns JSON list of GlobalIds |
|
||||
| `GET /render_diff?base=ifc://…&head=ifc://…` | Render a colour-coded diff PNG (added green, removed red) |
|
||||
| `GET /foundation/versions` | OpenCDE Foundation API discovery — lists BCF and Documents API endpoints (proxy at `/foundation/` on the Forgejo hostname) |
|
||||
| `GET /bcf/3.0/projects` | BCF 3.0 — list repositories as BCF projects |
|
||||
| `GET/POST /bcf/3.0/projects/{owner}/{repo}/topics` | BCF 3.0 — list or create topics (Forgejo issues) |
|
||||
| `GET/PUT /bcf/3.0/projects/{owner}/{repo}/topics/{guid}` | BCF 3.0 — get or update a topic |
|
||||
| `GET/POST /bcf/3.0/projects/{owner}/{repo}/topics/{guid}/comments` | BCF 3.0 — list or create comments |
|
||||
| `GET/POST /bcf/3.0/projects/{owner}/{repo}/topics/{guid}/viewpoints` | BCF 3.0 — list or create viewpoints (ifc:// URLs in comment bodies) |
|
||||
| `GET /bcf/3.0/projects/{owner}/{repo}/topics/{guid}/viewpoints/{guid}` | BCF 3.0 — get a viewpoint |
|
||||
| `GET /bcf/3.0/projects/{owner}/{repo}/topics/{guid}/viewpoints/{guid}/snapshot` | BCF 3.0 — render viewpoint as PNG via preview service |
|
||||
|
||||
The Foundation endpoint derives its public base URL from `X-Forwarded-Host` and `X-Forwarded-Proto` headers set by the reverse proxy, so no extra configuration is needed.
|
||||
The BCF API forwards the client's `Authorization: Bearer` header to Forgejo's REST API verbatim. No service account or machine token is required — Forgejo enforces per-user repository permissions on each call. The Foundation endpoint derives its public base URL from `X-Forwarded-Host` / `X-Forwarded-Proto` headers set by the reverse proxy.
|
||||
|
||||
**Configuration** (environment variables for the preview service):
|
||||
|
||||
| Variable | Default | Description |
|
||||
|---|---|---|
|
||||
| `IFCURL_FORGEJO_URL` | `http://localhost:3000` | Forgejo base URL for BCF API calls |
|
||||
| `IFCURL_PREVIEW_URL` | `http://localhost:8000` | Preview service URL for BCF snapshot endpoint |
|
||||
|
||||
**Proxy configuration** — add to nginx or Caddy on the Forgejo hostname:
|
||||
|
||||
```nginx
|
||||
location /foundation/ { proxy_pass http://localhost:8000/foundation/; }
|
||||
location /bcf/ { proxy_pass http://localhost:8000/bcf/; }
|
||||
```
|
||||
|
||||
**Caching:**
|
||||
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ from __future__ import annotations
|
|||
|
||||
import dataclasses
|
||||
import io
|
||||
import re
|
||||
import uuid
|
||||
import xml.etree.ElementTree as ET
|
||||
import zipfile
|
||||
|
|
@ -37,6 +38,8 @@ from datetime import datetime, timezone
|
|||
|
||||
from ifcurl.url import IfcUrl
|
||||
|
||||
_IFC_GUID_RE = re.compile(r"^[0-9A-Za-z_$]{22}$")
|
||||
|
||||
_VERSION_XML = b"""\
|
||||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<Version VersionId="2.1" xsi:noNamespaceSchemaLocation="https://raw.githubusercontent.com/buildingSMART/BCF-XML/release_2_1/Schemas/version.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
|
||||
|
|
@ -237,3 +240,50 @@ def bcf_viewpoint_to_ifc_url(base: IfcUrl, viewpoint: dict) -> str:
|
|||
selector=selector,
|
||||
visibility=visibility,
|
||||
).to_string()
|
||||
|
||||
|
||||
def ifc_url_to_bcf_viewpoint(ifc_url: IfcUrl, guid: str) -> dict:
|
||||
"""Convert an IfcUrl to a BCF 3.0 REST viewpoint dict.
|
||||
|
||||
Only GUID selectors (bare 22-char IFC GlobalIds joined with '+') are
|
||||
mapped to BCF components/selection. Complex IfcOpenShell expressions
|
||||
(e.g. 'IfcWall') cannot be expanded without the model and are omitted.
|
||||
"""
|
||||
vp: dict = {"guid": guid, "lines": [], "clipping_planes": []}
|
||||
|
||||
if ifc_url.camera is not None:
|
||||
px, py, pz, dx, dy, dz, ux, uy, uz = ifc_url.camera
|
||||
cam = {
|
||||
"camera_view_point": {"x": px, "y": py, "z": pz},
|
||||
"camera_direction": {"x": dx, "y": dy, "z": dz},
|
||||
"camera_up_vector": {"x": ux, "y": uy, "z": uz},
|
||||
}
|
||||
if ifc_url.fov is not None:
|
||||
vp["perspective_camera"] = {**cam, "field_of_view": ifc_url.fov}
|
||||
elif ifc_url.scale is not None:
|
||||
vp["orthogonal_camera"] = {**cam, "view_to_world_scale": ifc_url.scale}
|
||||
|
||||
for px, py, pz, nx, ny, nz in ifc_url.clips:
|
||||
vp["clipping_planes"].append({
|
||||
"location": {"x": px, "y": py, "z": pz},
|
||||
"direction": {"x": nx, "y": ny, "z": nz},
|
||||
})
|
||||
|
||||
selection: list[dict] = []
|
||||
default_visibility = True
|
||||
exceptions: list[dict] = []
|
||||
|
||||
if ifc_url.selector:
|
||||
guid_parts = [p.strip() for p in ifc_url.selector.split("+") if _IFC_GUID_RE.match(p.strip())]
|
||||
if guid_parts:
|
||||
if ifc_url.visibility == "isolate":
|
||||
default_visibility = False
|
||||
exceptions = [{"ifc_guid": g} for g in guid_parts]
|
||||
else:
|
||||
selection = [{"ifc_guid": g, "originating_system": "ifcurl", "authoring_tool_id": "ifcurl"} for g in guid_parts]
|
||||
|
||||
vp["components"] = {
|
||||
"selection": selection,
|
||||
"visibility": {"default_visibility": default_visibility, "exceptions": exceptions},
|
||||
}
|
||||
return vp
|
||||
|
|
|
|||
358
ifcurl/bcf_api.py
Normal file
358
ifcurl/bcf_api.py
Normal file
|
|
@ -0,0 +1,358 @@
|
|||
# IFC URL — BCF 3.0 REST API routes
|
||||
# Copyright (C) 2026 Bruno Postle <bruno@postle.net>
|
||||
#
|
||||
# This file is part of IFC URL.
|
||||
#
|
||||
# IFC URL is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# IFC URL is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Lesser General Public License
|
||||
# along with IFC URL. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""BCF 3.0 REST API — translation layer over Forgejo issues and comments.
|
||||
|
||||
Routes are mounted at /bcf/3.0 in service.py. Authentication is forwarded
|
||||
verbatim: the Authorization: Bearer header from the BCF client is passed to
|
||||
Forgejo's REST API, so Forgejo's own auth middleware enforces per-user
|
||||
permissions with no extra machinery.
|
||||
|
||||
Mapping:
|
||||
BCF Project ← Forgejo repository (project_id = "owner/repo")
|
||||
BCF Topic ← Forgejo issue
|
||||
BCF Comment ← Forgejo comment
|
||||
BCF Viewpoint ← first ifc:// URL found in a comment body
|
||||
|
||||
GUIDs are derived deterministically from the Forgejo resource IDs so that
|
||||
no new storage is required. The issue number / comment ID is embedded in the
|
||||
last 12 hex characters of the UUID and can be recovered without a lookup table.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, Body, HTTPException, Request
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
|
||||
from ifcurl.bcf import bcf_viewpoint_to_ifc_url, ifc_url_to_bcf_viewpoint
|
||||
from ifcurl.url import IfcUrl
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_FORGEJO_URL: str = os.environ.get("IFCURL_FORGEJO_URL", "http://localhost:3000")
|
||||
_PREVIEW_URL: str = os.environ.get("IFCURL_PREVIEW_URL", "http://localhost:8000")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GUID encoding / decoding
|
||||
#
|
||||
# Each GUID embeds the Forgejo numeric ID in the last 12 hex characters so
|
||||
# the mapping is reversible without a lookup table. The first 20 hex chars
|
||||
# come from a SHA-1 hash of "owner/repo". A nibble at position 13 encodes
|
||||
# the resource type: a=topic, b=comment, c=viewpoint.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _repo_prefix(owner: str, repo: str) -> str:
|
||||
h = hashlib.sha1(f"{owner}/{repo}".encode()).digest().hex()
|
||||
variant = (int(h[16:20], 16) & 0x3FFF) | 0x8000
|
||||
return f"{h[:8]}-{h[8:12]}-{{t}}{h[12:15]}-{variant:04x}"
|
||||
|
||||
|
||||
def make_topic_guid(owner: str, repo: str, number: int) -> str:
|
||||
return f"{_repo_prefix(owner, repo).format(t='a')}-{number:012x}"
|
||||
|
||||
|
||||
def make_comment_guid(owner: str, repo: str, comment_id: int) -> str:
|
||||
return f"{_repo_prefix(owner, repo).format(t='b')}-{comment_id:012x}"
|
||||
|
||||
|
||||
def make_viewpoint_guid(owner: str, repo: str, comment_id: int) -> str:
|
||||
return f"{_repo_prefix(owner, repo).format(t='c')}-{comment_id:012x}"
|
||||
|
||||
|
||||
def _id_from_guid(guid: str) -> int:
|
||||
return int(guid.split("-")[-1], 16)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ifc:// URL extraction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_IFC_URL_RE = re.compile(r"ifc://\S+")
|
||||
|
||||
|
||||
def _first_ifc_url(text: str | None) -> str | None:
|
||||
if not text:
|
||||
return None
|
||||
m = _IFC_URL_RE.search(text)
|
||||
return m.group(0) if m else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Forgejo API client
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _headers(auth: str | None) -> dict[str, str]:
|
||||
h: dict[str, str] = {"Content-Type": "application/json"}
|
||||
if auth:
|
||||
h["Authorization"] = auth
|
||||
return h
|
||||
|
||||
|
||||
def _fget(path: str, auth: str | None, params: dict | None = None) -> dict | list:
|
||||
r = httpx.get(f"{_FORGEJO_URL}{path}", headers=_headers(auth), params=params)
|
||||
if r.status_code == 404:
|
||||
raise HTTPException(status_code=404, detail="Not found")
|
||||
if r.status_code in (401, 403):
|
||||
raise HTTPException(status_code=r.status_code, detail="Unauthorized")
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
|
||||
def _fpost(path: str, auth: str | None, body: dict) -> dict:
|
||||
r = httpx.post(f"{_FORGEJO_URL}{path}", headers=_headers(auth), json=body)
|
||||
if r.status_code in (401, 403):
|
||||
raise HTTPException(status_code=r.status_code, detail="Unauthorized")
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
|
||||
def _fpatch(path: str, auth: str | None, body: dict) -> dict:
|
||||
r = httpx.patch(f"{_FORGEJO_URL}{path}", headers=_headers(auth), json=body)
|
||||
if r.status_code in (401, 403):
|
||||
raise HTTPException(status_code=r.status_code, detail="Unauthorized")
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Conversion helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_STATUS_MAP = {"open": "Open", "closed": "Closed"}
|
||||
_STATUS_REVERSE = {"Open": "open", "Closed": "closed"}
|
||||
|
||||
|
||||
def _issue_to_topic(issue: dict, owner: str, repo: str) -> dict:
|
||||
number = issue["number"]
|
||||
return {
|
||||
"guid": make_topic_guid(owner, repo, number),
|
||||
"topic_type": "Issue",
|
||||
"topic_status": _STATUS_MAP.get(issue.get("state", "open"), "Open"),
|
||||
"title": issue.get("title", ""),
|
||||
"description": issue.get("body", "") or "",
|
||||
"creation_date": issue.get("created_at", ""),
|
||||
"creation_author": (issue.get("user") or {}).get("login", ""),
|
||||
"modified_date": issue.get("updated_at", ""),
|
||||
"modified_author": (issue.get("user") or {}).get("login", ""),
|
||||
"assigned_to": ((issue.get("assignees") or [{}])[0] or {}).get("login") or None,
|
||||
"labels": [lbl["name"] for lbl in issue.get("labels", [])],
|
||||
"index": number,
|
||||
}
|
||||
|
||||
|
||||
def _comment_to_bcf(comment: dict, owner: str, repo: str) -> dict:
|
||||
cid = comment["id"]
|
||||
has_viewpoint = bool(_first_ifc_url(comment.get("body")))
|
||||
return {
|
||||
"guid": make_comment_guid(owner, repo, cid),
|
||||
"date": comment.get("created_at", ""),
|
||||
"author": (comment.get("user") or {}).get("login", ""),
|
||||
"comment": comment.get("body", "") or "",
|
||||
"viewpoint_guid": make_viewpoint_guid(owner, repo, cid) if has_viewpoint else None,
|
||||
}
|
||||
|
||||
|
||||
def _comment_to_viewpoint(comment: dict, owner: str, repo: str) -> dict | None:
|
||||
ifc_url_str = _first_ifc_url(comment.get("body"))
|
||||
if not ifc_url_str:
|
||||
return None
|
||||
try:
|
||||
parsed = IfcUrl.parse(ifc_url_str)
|
||||
except ValueError:
|
||||
return None
|
||||
return ifc_url_to_bcf_viewpoint(parsed, make_viewpoint_guid(owner, repo, comment["id"]))
|
||||
|
||||
|
||||
def _find_issue(owner: str, repo: str, tguid: str, auth: str | None) -> dict:
|
||||
number = _id_from_guid(tguid)
|
||||
issue = _fget(f"/api/v1/repos/{owner}/{repo}/issues/{number}", auth)
|
||||
if not isinstance(issue, dict) or make_topic_guid(owner, repo, issue["number"]) != tguid:
|
||||
raise HTTPException(status_code=404, detail="Topic not found")
|
||||
return issue
|
||||
|
||||
|
||||
def _find_comment(owner: str, repo: str, vpguid: str, auth: str | None) -> dict:
|
||||
cid = _id_from_guid(vpguid)
|
||||
comment = _fget(f"/api/v1/repos/{owner}/{repo}/issues/comments/{cid}", auth)
|
||||
if not isinstance(comment, dict) or make_viewpoint_guid(owner, repo, comment["id"]) != vpguid:
|
||||
raise HTTPException(status_code=404, detail="Viewpoint not found")
|
||||
return comment
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Router
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
router = APIRouter(prefix="/bcf/3.0")
|
||||
|
||||
|
||||
def _auth(request: Request) -> str | None:
|
||||
return request.headers.get("authorization")
|
||||
|
||||
|
||||
# --- Projects ---
|
||||
|
||||
@router.get("/projects")
|
||||
def list_projects(request: Request) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
result = _fget("/api/v1/repos/search", auth, params={"limit": 50})
|
||||
items = result if isinstance(result, list) else result.get("data", [])
|
||||
return JSONResponse([{"project_id": r["full_name"], "name": r["name"]} for r in items])
|
||||
|
||||
|
||||
@router.get("/projects/{owner}/{repo}")
|
||||
def get_project(owner: str, repo: str, request: Request) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
r = _fget(f"/api/v1/repos/{owner}/{repo}", auth)
|
||||
return JSONResponse({"project_id": r["full_name"], "name": r["name"]})
|
||||
|
||||
|
||||
@router.get("/projects/{owner}/{repo}/extensions")
|
||||
def get_extensions(owner: str, repo: str) -> JSONResponse:
|
||||
return JSONResponse({
|
||||
"topic_type": ["Issue", "Request", "Fault", "Inquiry"],
|
||||
"topic_status": ["Open", "Closed"],
|
||||
"priority": [],
|
||||
"label": [],
|
||||
"stage": [],
|
||||
"user_id_type": [],
|
||||
})
|
||||
|
||||
|
||||
# --- Topics ---
|
||||
|
||||
@router.get("/projects/{owner}/{repo}/topics")
|
||||
def list_topics(owner: str, repo: str, request: Request) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issues = _fget(f"/api/v1/repos/{owner}/{repo}/issues", auth, params={"type": "issues", "state": "open", "limit": 50})
|
||||
return JSONResponse([_issue_to_topic(i, owner, repo) for i in (issues if isinstance(issues, list) else [])])
|
||||
|
||||
|
||||
@router.post("/projects/{owner}/{repo}/topics", status_code=201)
|
||||
def create_topic(owner: str, repo: str, request: Request, body: dict = Body(default={})) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _fpost(f"/api/v1/repos/{owner}/{repo}/issues", auth, {
|
||||
"title": body.get("title", "BCF Topic"),
|
||||
"body": body.get("description", ""),
|
||||
})
|
||||
return JSONResponse(_issue_to_topic(issue, owner, repo), status_code=201)
|
||||
|
||||
|
||||
@router.get("/projects/{owner}/{repo}/topics/{tguid}")
|
||||
def get_topic(owner: str, repo: str, tguid: str, request: Request) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
return JSONResponse(_issue_to_topic(issue, owner, repo))
|
||||
|
||||
|
||||
@router.put("/projects/{owner}/{repo}/topics/{tguid}")
|
||||
def update_topic(owner: str, repo: str, tguid: str, request: Request, body: dict = Body(default={})) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
patch: dict = {}
|
||||
if "title" in body:
|
||||
patch["title"] = body["title"]
|
||||
if "description" in body:
|
||||
patch["body"] = body["description"]
|
||||
if "topic_status" in body:
|
||||
patch["state"] = _STATUS_REVERSE.get(body["topic_status"], "open")
|
||||
updated = _fpatch(f"/api/v1/repos/{owner}/{repo}/issues/{issue['number']}", auth, patch)
|
||||
return JSONResponse(_issue_to_topic(updated, owner, repo))
|
||||
|
||||
|
||||
# --- Comments ---
|
||||
|
||||
@router.get("/projects/{owner}/{repo}/topics/{tguid}/comments")
|
||||
def list_comments(owner: str, repo: str, tguid: str, request: Request) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
comments = _fget(f"/api/v1/repos/{owner}/{repo}/issues/{issue['number']}/comments", auth)
|
||||
return JSONResponse([_comment_to_bcf(c, owner, repo) for c in (comments if isinstance(comments, list) else [])])
|
||||
|
||||
|
||||
@router.post("/projects/{owner}/{repo}/topics/{tguid}/comments", status_code=201)
|
||||
def create_comment(owner: str, repo: str, tguid: str, request: Request, body: dict = Body(default={})) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
comment = _fpost(f"/api/v1/repos/{owner}/{repo}/issues/{issue['number']}/comments", auth, {
|
||||
"body": body.get("comment", ""),
|
||||
})
|
||||
return JSONResponse(_comment_to_bcf(comment, owner, repo), status_code=201)
|
||||
|
||||
|
||||
# --- Viewpoints ---
|
||||
|
||||
@router.get("/projects/{owner}/{repo}/topics/{tguid}/viewpoints")
|
||||
def list_viewpoints(owner: str, repo: str, tguid: str, request: Request) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
comments = _fget(f"/api/v1/repos/{owner}/{repo}/issues/{issue['number']}/comments", auth)
|
||||
viewpoints = [vp for c in (comments if isinstance(comments, list) else []) if (vp := _comment_to_viewpoint(c, owner, repo))]
|
||||
return JSONResponse(viewpoints)
|
||||
|
||||
|
||||
@router.post("/projects/{owner}/{repo}/topics/{tguid}/viewpoints", status_code=201)
|
||||
def create_viewpoint(owner: str, repo: str, tguid: str, request: Request, body: dict = Body(default={})) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
ifc_url_str = _first_ifc_url(issue.get("body"))
|
||||
if not ifc_url_str:
|
||||
raise HTTPException(status_code=422, detail="Issue body contains no ifc:// URL to use as repo/ref/path context")
|
||||
try:
|
||||
base = IfcUrl.parse(ifc_url_str)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||||
ifc_url = bcf_viewpoint_to_ifc_url(base, body)
|
||||
comment = _fpost(f"/api/v1/repos/{owner}/{repo}/issues/{issue['number']}/comments", auth, {"body": ifc_url})
|
||||
vp = _comment_to_viewpoint(comment, owner, repo)
|
||||
return JSONResponse(vp, status_code=201)
|
||||
|
||||
|
||||
@router.get("/projects/{owner}/{repo}/topics/{tguid}/viewpoints/{vpguid}")
|
||||
def get_viewpoint(owner: str, repo: str, tguid: str, vpguid: str, request: Request) -> JSONResponse:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
comment = _find_comment(owner, repo, vpguid, auth)
|
||||
vp = _comment_to_viewpoint(comment, owner, repo)
|
||||
if not vp:
|
||||
raise HTTPException(status_code=404, detail="Viewpoint not found")
|
||||
return JSONResponse(vp)
|
||||
|
||||
|
||||
@router.get("/projects/{owner}/{repo}/topics/{tguid}/viewpoints/{vpguid}/snapshot")
|
||||
def get_snapshot(owner: str, repo: str, tguid: str, vpguid: str, request: Request) -> Response:
|
||||
auth = _auth(request)
|
||||
issue = _find_issue(owner, repo, tguid, auth)
|
||||
comment = _find_comment(owner, repo, vpguid, auth)
|
||||
ifc_url_str = _first_ifc_url(comment.get("body"))
|
||||
if not ifc_url_str:
|
||||
raise HTTPException(status_code=404, detail="No ifc:// URL in viewpoint comment")
|
||||
headers = {}
|
||||
if auth:
|
||||
headers["Authorization"] = auth
|
||||
r = httpx.get(f"{_PREVIEW_URL}/preview", params={"url": ifc_url_str}, headers=headers, timeout=120)
|
||||
if r.status_code != 200:
|
||||
raise HTTPException(status_code=r.status_code, detail="Preview service error")
|
||||
return Response(content=r.content, media_type="image/png")
|
||||
|
|
@ -52,6 +52,7 @@ from pydantic import BaseModel
|
|||
|
||||
from ifcurl.auth import get_token_for_host
|
||||
from ifcurl.bcf import build_bcf
|
||||
from ifcurl.bcf_api import router as bcf_router
|
||||
from ifcurl.git import diff_text as git_diff_text
|
||||
from ifcurl.git import fetch_ifc
|
||||
from ifcurl.render_service import _sandboxed_diff, _sandboxed_pipeline, _sandboxed_select
|
||||
|
|
@ -172,6 +173,7 @@ app = FastAPI(
|
|||
description="Renders ifc:// URLs to PNG images for embedding in Gitea and other consumers.",
|
||||
version="0.0.0",
|
||||
)
|
||||
app.include_router(bcf_router)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rate limiting (in-process, per client IP, sliding window)
|
||||
|
|
|
|||
321
tests/test_bcf_api.py
Normal file
321
tests/test_bcf_api.py
Normal file
|
|
@ -0,0 +1,321 @@
|
|||
"""Tests for ifcurl.bcf_api — BCF 3.0 REST API routes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from ifcurl.bcf_api import (
|
||||
_comment_to_bcf,
|
||||
_comment_to_viewpoint,
|
||||
_first_ifc_url,
|
||||
_id_from_guid,
|
||||
_issue_to_topic,
|
||||
make_comment_guid,
|
||||
make_topic_guid,
|
||||
make_viewpoint_guid,
|
||||
)
|
||||
from ifcurl.service import _rate_hits, app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_rate_limit():
|
||||
_rate_hits.clear()
|
||||
yield
|
||||
_rate_hits.clear()
|
||||
|
||||
OWNER, REPO = "alice", "myproject"
|
||||
AUTH = "Bearer test-token"
|
||||
|
||||
# Minimal Forgejo API fixture data
|
||||
ISSUE_1 = {
|
||||
"number": 1,
|
||||
"title": "Crack in wall",
|
||||
"body": "ifc://example.com/org/repo@heads/main?path=model.ifc&camera=1.0,2.0,3.0,0.0,0.0,-1.0,0.0,1.0,0.0&fov=60.0",
|
||||
"state": "open",
|
||||
"created_at": "2026-01-01T10:00:00Z",
|
||||
"updated_at": "2026-01-02T10:00:00Z",
|
||||
"user": {"login": "alice"},
|
||||
"assignees": [],
|
||||
"labels": [],
|
||||
}
|
||||
ISSUE_2 = {
|
||||
"number": 2,
|
||||
"title": "Missing door",
|
||||
"body": "Please check",
|
||||
"state": "closed",
|
||||
"created_at": "2026-01-03T10:00:00Z",
|
||||
"updated_at": "2026-01-03T10:00:00Z",
|
||||
"user": {"login": "bob"},
|
||||
"assignees": [],
|
||||
"labels": [],
|
||||
}
|
||||
COMMENT_WITH_VP = {
|
||||
"id": 100,
|
||||
"body": "ifc://example.com/org/repo@heads/main?path=model.ifc&camera=1.0,2.0,3.0,0.0,0.0,-1.0,0.0,1.0,0.0&fov=60.0",
|
||||
"created_at": "2026-01-01T11:00:00Z",
|
||||
"user": {"login": "alice"},
|
||||
}
|
||||
COMMENT_NO_VP = {
|
||||
"id": 101,
|
||||
"body": "Looks like a structural issue.",
|
||||
"created_at": "2026-01-01T12:00:00Z",
|
||||
"user": {"login": "bob"},
|
||||
}
|
||||
REPO_INFO = {
|
||||
"full_name": "alice/myproject",
|
||||
"name": "myproject",
|
||||
}
|
||||
REPO_LIST = {"data": [REPO_INFO]}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GUID functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGuidFunctions:
|
||||
def test_topic_guid_is_stable(self):
|
||||
assert make_topic_guid(OWNER, REPO, 1) == make_topic_guid(OWNER, REPO, 1)
|
||||
|
||||
def test_topic_guid_differs_by_number(self):
|
||||
assert make_topic_guid(OWNER, REPO, 1) != make_topic_guid(OWNER, REPO, 2)
|
||||
|
||||
def test_topic_guid_differs_by_repo(self):
|
||||
assert make_topic_guid("alice", "a", 1) != make_topic_guid("alice", "b", 1)
|
||||
|
||||
def test_id_from_topic_guid_roundtrip(self):
|
||||
for n in (1, 42, 1000, 99999):
|
||||
assert _id_from_guid(make_topic_guid(OWNER, REPO, n)) == n
|
||||
|
||||
def test_id_from_comment_guid_roundtrip(self):
|
||||
for cid in (100, 99999):
|
||||
assert _id_from_guid(make_comment_guid(OWNER, REPO, cid)) == cid
|
||||
|
||||
def test_id_from_viewpoint_guid_roundtrip(self):
|
||||
for cid in (100, 99999):
|
||||
assert _id_from_guid(make_viewpoint_guid(OWNER, REPO, cid)) == cid
|
||||
|
||||
def test_guids_are_uuid_shaped(self):
|
||||
g = make_topic_guid(OWNER, REPO, 1)
|
||||
parts = g.split("-")
|
||||
assert len(parts) == 5
|
||||
assert len(parts[0]) == 8
|
||||
assert len(parts[4]) == 12
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestHelpers:
|
||||
def test_first_ifc_url_found(self):
|
||||
assert _first_ifc_url("see ifc://example.com/o/r@HEAD?path=m.ifc here") == "ifc://example.com/o/r@HEAD?path=m.ifc"
|
||||
|
||||
def test_first_ifc_url_none(self):
|
||||
assert _first_ifc_url("no url here") is None
|
||||
|
||||
def test_first_ifc_url_empty(self):
|
||||
assert _first_ifc_url("") is None
|
||||
|
||||
def test_issue_to_topic_fields(self):
|
||||
t = _issue_to_topic(ISSUE_1, OWNER, REPO)
|
||||
assert t["guid"] == make_topic_guid(OWNER, REPO, 1)
|
||||
assert t["title"] == "Crack in wall"
|
||||
assert t["topic_status"] == "Open"
|
||||
assert t["index"] == 1
|
||||
|
||||
def test_issue_to_topic_closed(self):
|
||||
t = _issue_to_topic(ISSUE_2, OWNER, REPO)
|
||||
assert t["topic_status"] == "Closed"
|
||||
|
||||
def test_comment_to_bcf_with_viewpoint(self):
|
||||
c = _comment_to_bcf(COMMENT_WITH_VP, OWNER, REPO)
|
||||
assert c["guid"] == make_comment_guid(OWNER, REPO, 100)
|
||||
assert c["viewpoint_guid"] == make_viewpoint_guid(OWNER, REPO, 100)
|
||||
|
||||
def test_comment_to_bcf_without_viewpoint(self):
|
||||
c = _comment_to_bcf(COMMENT_NO_VP, OWNER, REPO)
|
||||
assert c["viewpoint_guid"] is None
|
||||
|
||||
def test_comment_to_viewpoint_with_ifc_url(self):
|
||||
vp = _comment_to_viewpoint(COMMENT_WITH_VP, OWNER, REPO)
|
||||
assert vp is not None
|
||||
assert vp["guid"] == make_viewpoint_guid(OWNER, REPO, 100)
|
||||
assert "perspective_camera" in vp
|
||||
|
||||
def test_comment_to_viewpoint_without_ifc_url(self):
|
||||
assert _comment_to_viewpoint(COMMENT_NO_VP, OWNER, REPO) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# BCF API routes (Forgejo calls mocked)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _mock_fget(responses: dict):
|
||||
"""Return a side_effect function that maps path → response."""
|
||||
def side_effect(path, auth, params=None):
|
||||
for key, val in responses.items():
|
||||
if path.endswith(key) or key in path:
|
||||
return val
|
||||
raise Exception(f"Unexpected fget: {path}")
|
||||
return side_effect
|
||||
|
||||
|
||||
class TestProjectRoutes:
|
||||
def test_list_projects(self):
|
||||
with patch("ifcurl.bcf_api._fget", return_value=REPO_LIST):
|
||||
r = client.get("/bcf/3.0/projects", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert any(p["project_id"] == "alice/myproject" for p in data)
|
||||
|
||||
def test_get_project(self):
|
||||
with patch("ifcurl.bcf_api._fget", return_value=REPO_INFO):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["project_id"] == "alice/myproject"
|
||||
|
||||
def test_get_extensions(self):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}/extensions")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert "topic_type" in data
|
||||
assert "topic_status" in data
|
||||
|
||||
|
||||
class TestTopicRoutes:
|
||||
def test_list_topics(self):
|
||||
with patch("ifcurl.bcf_api._fget", return_value=[ISSUE_1, ISSUE_2]):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}/topics", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 200
|
||||
topics = r.json()
|
||||
assert len(topics) == 2
|
||||
assert topics[0]["title"] == "Crack in wall"
|
||||
|
||||
def test_get_topic(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 1)
|
||||
with patch("ifcurl.bcf_api._fget", return_value=ISSUE_1):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["guid"] == tguid
|
||||
|
||||
def test_get_topic_wrong_guid_returns_404(self):
|
||||
# GUID that encodes issue number 999 but issue 1 is returned
|
||||
tguid = make_topic_guid(OWNER, REPO, 999)
|
||||
with patch("ifcurl.bcf_api._fget", return_value=ISSUE_1):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 404
|
||||
|
||||
def test_create_topic(self):
|
||||
with patch("ifcurl.bcf_api._fpost", return_value=ISSUE_1):
|
||||
r = client.post(
|
||||
f"/bcf/3.0/projects/{OWNER}/{REPO}/topics",
|
||||
json={"title": "Crack in wall", "description": ""},
|
||||
headers={"Authorization": AUTH},
|
||||
)
|
||||
assert r.status_code == 201
|
||||
assert r.json()["title"] == "Crack in wall"
|
||||
|
||||
def test_update_topic_status(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 1)
|
||||
closed_issue = {**ISSUE_1, "state": "closed"}
|
||||
with (
|
||||
patch("ifcurl.bcf_api._fget", return_value=ISSUE_1),
|
||||
patch("ifcurl.bcf_api._fpatch", return_value=closed_issue),
|
||||
):
|
||||
r = client.put(
|
||||
f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}",
|
||||
json={"topic_status": "Closed"},
|
||||
headers={"Authorization": AUTH},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert r.json()["topic_status"] == "Closed"
|
||||
|
||||
|
||||
class TestCommentRoutes:
|
||||
def test_list_comments(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 1)
|
||||
with patch("ifcurl.bcf_api._fget", side_effect=[ISSUE_1, [COMMENT_WITH_VP, COMMENT_NO_VP]]):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}/comments", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 200
|
||||
comments = r.json()
|
||||
assert len(comments) == 2
|
||||
assert comments[0]["viewpoint_guid"] is not None
|
||||
assert comments[1]["viewpoint_guid"] is None
|
||||
|
||||
def test_create_comment(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 1)
|
||||
with (
|
||||
patch("ifcurl.bcf_api._fget", return_value=ISSUE_1),
|
||||
patch("ifcurl.bcf_api._fpost", return_value=COMMENT_NO_VP),
|
||||
):
|
||||
r = client.post(
|
||||
f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}/comments",
|
||||
json={"comment": "Looks like a structural issue."},
|
||||
headers={"Authorization": AUTH},
|
||||
)
|
||||
assert r.status_code == 201
|
||||
assert r.json()["author"] == "bob"
|
||||
|
||||
|
||||
class TestViewpointRoutes:
|
||||
def test_list_viewpoints(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 1)
|
||||
with patch("ifcurl.bcf_api._fget", side_effect=[ISSUE_1, [COMMENT_WITH_VP, COMMENT_NO_VP]]):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}/viewpoints", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 200
|
||||
viewpoints = r.json()
|
||||
assert len(viewpoints) == 1
|
||||
assert "perspective_camera" in viewpoints[0]
|
||||
|
||||
def test_get_viewpoint(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 1)
|
||||
vpguid = make_viewpoint_guid(OWNER, REPO, 100)
|
||||
with patch("ifcurl.bcf_api._fget", side_effect=[ISSUE_1, COMMENT_WITH_VP]):
|
||||
r = client.get(f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}/viewpoints/{vpguid}", headers={"Authorization": AUTH})
|
||||
assert r.status_code == 200
|
||||
vp = r.json()
|
||||
assert vp["guid"] == vpguid
|
||||
assert "perspective_camera" in vp
|
||||
|
||||
def test_create_viewpoint_posts_ifc_url(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 1)
|
||||
posted_body = {}
|
||||
|
||||
def mock_fpost(path, auth, body):
|
||||
posted_body.update(body)
|
||||
return {**COMMENT_WITH_VP, "body": body["body"]}
|
||||
|
||||
vp_body = {
|
||||
"perspective_camera": {
|
||||
"camera_view_point": {"x": 5.0, "y": 5.0, "z": 5.0},
|
||||
"camera_direction": {"x": 0.0, "y": 0.0, "z": -1.0},
|
||||
"camera_up_vector": {"x": 0.0, "y": 1.0, "z": 0.0},
|
||||
"field_of_view": 45.0,
|
||||
}
|
||||
}
|
||||
with (
|
||||
patch("ifcurl.bcf_api._fget", return_value=ISSUE_1),
|
||||
patch("ifcurl.bcf_api._fpost", side_effect=mock_fpost),
|
||||
):
|
||||
r = client.post(
|
||||
f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}/viewpoints",
|
||||
json=vp_body,
|
||||
headers={"Authorization": AUTH},
|
||||
)
|
||||
assert r.status_code == 201
|
||||
assert posted_body.get("body", "").startswith("ifc://")
|
||||
|
||||
def test_create_viewpoint_no_ifc_url_in_issue_returns_422(self):
|
||||
tguid = make_topic_guid(OWNER, REPO, 2)
|
||||
with patch("ifcurl.bcf_api._fget", return_value=ISSUE_2):
|
||||
r = client.post(
|
||||
f"/bcf/3.0/projects/{OWNER}/{REPO}/topics/{tguid}/viewpoints",
|
||||
json={},
|
||||
headers={"Authorization": AUTH},
|
||||
)
|
||||
assert r.status_code == 422
|
||||
|
|
@ -10,6 +10,7 @@ import pytest
|
|||
from fastapi.testclient import TestClient
|
||||
|
||||
from ifcurl.service import (
|
||||
_rate_hits,
|
||||
_t2_cache,
|
||||
_t2_get,
|
||||
_t2_put,
|
||||
|
|
@ -36,6 +37,7 @@ def clear_caches(tmp_path, monkeypatch):
|
|||
monkeypatch.setattr("ifcurl.service.user_cache_dir", lambda *a, **kw: str(tmp_path))
|
||||
_t2_cache.clear()
|
||||
_t3_cache.clear()
|
||||
_rate_hits.clear()
|
||||
configure_allowed_hosts(None)
|
||||
yield
|
||||
_t2_cache.clear()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue