mirror of
https://github.com/brunopostle/ifcurl.git
synced 2026-07-12 10:18:14 +00:00
Authentication and test coverage
- ifcurl/auth.py: token config (~/.config/ifcurl/tokens.json), get_token_for_host(), inject_token() for HTTPS URL credential injection - ifcurl/git.py: thread token= through fetch_ifc/fetch_ifc_bytes/_open_remote; authenticated fetch uses 'git fetch <auth_url> +refs/*:refs/*' so the token is never written to on-disk git config - ifcurl/service.py: optional token field in PreviewRequest; request token takes precedence over config-file token; get_token_for_host imported at module level for clean monkeypatching in tests - tests Generated with the assistance of an AI coding tool.
This commit is contained in:
parent
018ee446b5
commit
7f92cadc50
9 changed files with 887 additions and 35 deletions
53
README.md
53
README.md
|
|
@ -8,8 +8,6 @@ A single URL encodes the model source (git host, repository, ref), an optional e
|
||||||
ifcurl render "ifc://github.com/brunopostle/simple-ifc@heads/main?path=_test_simple.ifc"
|
ifcurl render "ifc://github.com/brunopostle/simple-ifc@heads/main?path=_test_simple.ifc"
|
||||||
```
|
```
|
||||||
|
|
||||||

|
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## URL format
|
## URL format
|
||||||
|
|
@ -102,35 +100,36 @@ ifcurl render "ifc://..." -o output.png
|
||||||
## Installation
|
## Installation
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install ifcurl
|
pip install "ifcurl[render]" # CLI rendering tool
|
||||||
```
|
pip install "ifcurl[service]" # preview service (includes render)
|
||||||
|
|
||||||
For rendering support (required for `ifcurl render`):
|
|
||||||
|
|
||||||
```bash
|
|
||||||
pip install "ifcurl[render]"
|
|
||||||
```
|
```
|
||||||
|
|
||||||
**Dependencies:**
|
**Dependencies:**
|
||||||
|
|
||||||
- [ifcopenshell](https://ifcopenshell.org) — IFC parsing and selector execution
|
- [ifcopenshell](https://ifcopenshell.org) — IFC parsing and selector execution
|
||||||
- [GitPython](https://gitpython.readthedocs.io) — git repository access
|
- [GitPython](https://gitpython.readthedocs.io) — git repository access
|
||||||
- [platformdirs](https://platformdirs.readthedocs.io) — OS-appropriate cache directory
|
- [platformdirs](https://platformdirs.readthedocs.io) — OS-appropriate cache and config directories
|
||||||
- [pyvista](https://pyvista.org) + [numpy](https://numpy.org) — 3D rendering (optional, required for `render`)
|
- [pyvista](https://pyvista.org) + [numpy](https://numpy.org) — 3D rendering (`[render]` and `[service]` extras)
|
||||||
|
- [FastAPI](https://fastapi.tiangolo.com) + [uvicorn](https://www.uvicorn.org) — HTTP service (`[service]` extra)
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## Python library
|
## Python library
|
||||||
|
|
||||||
```python
|
```python
|
||||||
from ifcurl import IfcUrl, fetch_ifc_bytes
|
from ifcurl import IfcUrl, fetch_ifc
|
||||||
|
from ifcurl import render as render_mod
|
||||||
import ifcopenshell
|
import ifcopenshell
|
||||||
import tempfile, os
|
import tempfile, os
|
||||||
from ifcurl import render
|
|
||||||
|
|
||||||
url = IfcUrl.parse("ifc://github.com/brunopostle/simple-ifc@heads/main?path=_test_simple.ifc&selector=IfcWall&visibility=ghost")
|
url = IfcUrl.parse(
|
||||||
|
"ifc://github.com/brunopostle/simple-ifc@heads/main"
|
||||||
|
"?path=_test_simple.ifc&selector=IfcWall&visibility=ghost"
|
||||||
|
)
|
||||||
|
|
||||||
ifc_bytes = fetch_ifc_bytes(url)
|
# fetch_ifc returns (commit_hexsha, ifc_bytes) — the hexsha is useful
|
||||||
|
# as a cache key even when the URL uses a mutable ref like a branch
|
||||||
|
hexsha, ifc_bytes = fetch_ifc(url)
|
||||||
|
|
||||||
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".ifc")
|
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".ifc")
|
||||||
os.write(tmp_fd, ifc_bytes)
|
os.write(tmp_fd, ifc_bytes)
|
||||||
|
|
@ -138,7 +137,7 @@ os.close(tmp_fd)
|
||||||
model = ifcopenshell.open(tmp_path)
|
model = ifcopenshell.open(tmp_path)
|
||||||
os.unlink(tmp_path)
|
os.unlink(tmp_path)
|
||||||
|
|
||||||
png_bytes = render.render(
|
png_bytes = render_mod.render(
|
||||||
model,
|
model,
|
||||||
selector=url.selector,
|
selector=url.selector,
|
||||||
camera=url.camera,
|
camera=url.camera,
|
||||||
|
|
@ -159,7 +158,6 @@ with open("output.png", "wb") as f:
|
||||||
ifcurl includes an HTTP preview service for use by Gitea and other consumers.
|
ifcurl includes an HTTP preview service for use by Gitea and other consumers.
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install "ifcurl[service]"
|
|
||||||
ifcurl serve # 127.0.0.1:8000
|
ifcurl serve # 127.0.0.1:8000
|
||||||
ifcurl serve --host 0.0.0.0 --port 9000
|
ifcurl serve --host 0.0.0.0 --port 9000
|
||||||
```
|
```
|
||||||
|
|
@ -170,11 +168,28 @@ ifcurl serve --host 0.0.0.0 --port 9000
|
||||||
POST /preview
|
POST /preview
|
||||||
Content-Type: application/json
|
Content-Type: application/json
|
||||||
|
|
||||||
{"url": "ifc://..."}
|
{"url": "ifc://...", "token": "optional-git-token"}
|
||||||
```
|
```
|
||||||
|
|
||||||
Returns `image/png`.
|
Returns `image/png`.
|
||||||
|
|
||||||
|
The optional `token` field is a bearer token for git authentication. When provided it takes precedence over any token in the config file. Intended for co-located Gitea deployments that pass the requesting user's session token.
|
||||||
|
|
||||||
|
### Authentication
|
||||||
|
|
||||||
|
For private repositories, configure a token per host in `~/.config/ifcurl/tokens.json` (Linux/macOS) or `%APPDATA%\ifcurl\tokens.json` (Windows):
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"hosts": {
|
||||||
|
"github.com": "ghp_your_token_here",
|
||||||
|
"gitlab.example.com": "glpat_your_token_here"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
Tokens are injected into HTTPS remote URLs (`https://<token>@host/path`) for clone and fetch operations and are never written to the on-disk git config. SSH transport uses the platform key store and ignores this config.
|
||||||
|
|
||||||
### Service caching
|
### Service caching
|
||||||
|
|
||||||
| Tier | Key | Contents | Notes |
|
| Tier | Key | Contents | Notes |
|
||||||
|
|
@ -198,7 +213,7 @@ Remote repositories are cloned as bare repos to the OS cache directory (`~/.cach
|
||||||
```bash
|
```bash
|
||||||
git clone https://github.com/brunopostle/ifcurl
|
git clone https://github.com/brunopostle/ifcurl
|
||||||
cd ifcurl
|
cd ifcurl
|
||||||
pip install -e ".[render]"
|
pip install -e ".[service]"
|
||||||
python -m pytest tests/
|
python -m pytest tests/
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|
|
||||||
78
ifcurl/auth.py
Normal file
78
ifcurl/auth.py
Normal file
|
|
@ -0,0 +1,78 @@
|
||||||
|
# IFC URL — resolve and render ifc:// URLs
|
||||||
|
# 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/>.
|
||||||
|
|
||||||
|
"""Token-based authentication for git hosts.
|
||||||
|
|
||||||
|
Configuration file
|
||||||
|
------------------
|
||||||
|
``~/.config/ifcurl/tokens.json`` (Linux/macOS) or
|
||||||
|
``%APPDATA%\\ifcurl\\tokens.json`` (Windows):
|
||||||
|
|
||||||
|
.. code-block:: json
|
||||||
|
|
||||||
|
{
|
||||||
|
"hosts": {
|
||||||
|
"github.com": "ghp_your_token_here",
|
||||||
|
"gitlab.example.com": "glpat_your_token_here"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
The token is injected into HTTPS remote URLs as ``https://<token>@host/path``.
|
||||||
|
SSH transport is unaffected; SSH authentication uses the platform key store.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
from urllib.parse import urlparse, urlunparse
|
||||||
|
|
||||||
|
from platformdirs import user_config_dir
|
||||||
|
|
||||||
|
|
||||||
|
def config_path() -> Path:
|
||||||
|
"""Return the path to the ifcurl tokens config file."""
|
||||||
|
return Path(user_config_dir("ifcurl")) / "tokens.json"
|
||||||
|
|
||||||
|
|
||||||
|
def get_token_for_host(host: str) -> str | None:
|
||||||
|
"""Return the configured token for *host*, or ``None`` if not set.
|
||||||
|
|
||||||
|
Reads ``tokens.json`` from the OS config directory. Silently returns
|
||||||
|
``None`` on any read or parse error so callers fall back to anonymous.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
data = json.loads(config_path().read_text())
|
||||||
|
return data.get("hosts", {}).get(host) or None
|
||||||
|
except (FileNotFoundError, json.JSONDecodeError, KeyError, TypeError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def inject_token(https_url: str, token: str) -> str:
|
||||||
|
"""Return *https_url* with *token* injected as the username credential.
|
||||||
|
|
||||||
|
``https://host/org/repo`` → ``https://<token>@host/org/repo``
|
||||||
|
|
||||||
|
This format is accepted by GitHub, GitLab, Gitea, and most other git
|
||||||
|
hosting platforms. The token is never stored — it only appears in the
|
||||||
|
git command line arguments used for clone/fetch operations.
|
||||||
|
"""
|
||||||
|
parsed = urlparse(https_url)
|
||||||
|
port_suffix = f":{parsed.port}" if parsed.port else ""
|
||||||
|
netloc = f"{token}@{parsed.hostname}{port_suffix}"
|
||||||
|
return urlunparse(parsed._replace(netloc=netloc))
|
||||||
|
|
@ -19,9 +19,15 @@
|
||||||
"""Fetch IFC file bytes from a git repository described by an IfcUrl.
|
"""Fetch IFC file bytes from a git repository described by an IfcUrl.
|
||||||
|
|
||||||
Remote repositories are cloned as bare repos to a per-URL cache directory
|
Remote repositories are cloned as bare repos to a per-URL cache directory
|
||||||
under ~/.cache/ifcurl/. Mutable refs (HEAD, branches) trigger a fetch on
|
under the OS cache directory. Mutable refs (HEAD, branches) trigger a fetch
|
||||||
each call to pick up upstream changes; immutable refs (commit hashes, tags)
|
on each call to pick up upstream changes; immutable refs (commit hashes, tags)
|
||||||
reuse the cached clone without fetching.
|
reuse the cached clone without fetching.
|
||||||
|
|
||||||
|
Authentication
|
||||||
|
--------------
|
||||||
|
Pass a *token* string to inject it into HTTPS remote URLs as a credential.
|
||||||
|
SSH transport uses the platform key store and ignores the token.
|
||||||
|
See :mod:`ifcurl.auth` for loading tokens from the config file.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
@ -46,13 +52,15 @@ except ImportError:
|
||||||
from ifcurl.url import IfcUrl
|
from ifcurl.url import IfcUrl
|
||||||
|
|
||||||
|
|
||||||
def fetch_ifc(ifc_url: IfcUrl) -> tuple[str, bytes]:
|
def fetch_ifc(ifc_url: IfcUrl, token: str | None = None) -> tuple[str, bytes]:
|
||||||
"""Return ``(commit_hexsha, ifc_bytes)`` for the file addressed by *ifc_url*.
|
"""Return ``(commit_hexsha, ifc_bytes)`` for the file addressed by *ifc_url*.
|
||||||
|
|
||||||
The commit hexsha is the resolved, immutable identifier for the ref —
|
The commit hexsha is the resolved, immutable identifier for the ref —
|
||||||
useful as a cache key even when the URL uses a mutable ref like a branch.
|
useful as a cache key even when the URL uses a mutable ref like a branch.
|
||||||
|
|
||||||
:param ifc_url: A parsed :class:`IfcUrl`.
|
:param ifc_url: A parsed :class:`IfcUrl`.
|
||||||
|
:param token: Optional bearer token for HTTPS authentication. Injected
|
||||||
|
as ``https://<token>@host/path``. Ignored for SSH and local repos.
|
||||||
:raises ImportError: If GitPython is not installed.
|
:raises ImportError: If GitPython is not installed.
|
||||||
:raises ValueError: If ``ifc_url.path`` is unset, the repo cannot be
|
:raises ValueError: If ``ifc_url.path`` is unset, the repo cannot be
|
||||||
reached, or the file is not found at the specified ref.
|
reached, or the file is not found at the specified ref.
|
||||||
|
|
@ -62,19 +70,20 @@ def fetch_ifc(ifc_url: IfcUrl) -> tuple[str, bytes]:
|
||||||
if ifc_url.path is None:
|
if ifc_url.path is None:
|
||||||
raise ValueError("URL has no 'path' parameter — cannot fetch IFC file")
|
raise ValueError("URL has no 'path' parameter — cannot fetch IFC file")
|
||||||
|
|
||||||
repo = _get_repo(ifc_url)
|
repo = _get_repo(ifc_url, token=token)
|
||||||
return _read_commit_blob(repo, ifc_url.git_ref(), ifc_url.path)
|
return _read_commit_blob(repo, ifc_url.git_ref(), ifc_url.path)
|
||||||
|
|
||||||
|
|
||||||
def fetch_ifc_bytes(ifc_url: IfcUrl) -> bytes:
|
def fetch_ifc_bytes(ifc_url: IfcUrl, token: str | None = None) -> bytes:
|
||||||
"""Return the raw bytes of the IFC file addressed by *ifc_url*.
|
"""Return the raw bytes of the IFC file addressed by *ifc_url*.
|
||||||
|
|
||||||
:param ifc_url: A parsed :class:`IfcUrl`.
|
:param ifc_url: A parsed :class:`IfcUrl`.
|
||||||
|
:param token: Optional bearer token for HTTPS authentication.
|
||||||
:raises ImportError: If GitPython is not installed.
|
:raises ImportError: If GitPython is not installed.
|
||||||
:raises ValueError: If ``ifc_url.path`` is unset, the repo cannot be
|
:raises ValueError: If ``ifc_url.path`` is unset, the repo cannot be
|
||||||
reached, or the file is not found at the specified ref.
|
reached, or the file is not found at the specified ref.
|
||||||
"""
|
"""
|
||||||
_, data = fetch_ifc(ifc_url)
|
_, data = fetch_ifc(ifc_url, token=token)
|
||||||
return data
|
return data
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -83,11 +92,11 @@ def fetch_ifc_bytes(ifc_url: IfcUrl) -> bytes:
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
def _get_repo(ifc_url: IfcUrl) -> git.Repo:
|
def _get_repo(ifc_url: IfcUrl, token: str | None = None) -> git.Repo:
|
||||||
"""Open or fetch the repository for *ifc_url*."""
|
"""Open or fetch the repository for *ifc_url*."""
|
||||||
if ifc_url.transport == "local":
|
if ifc_url.transport == "local":
|
||||||
return _open_local(ifc_url.repo_path)
|
return _open_local(ifc_url.repo_path)
|
||||||
return _open_remote(ifc_url.git_remote_url(), ifc_url.is_mutable_ref())
|
return _open_remote(ifc_url.git_remote_url(), ifc_url.is_mutable_ref(), token=token)
|
||||||
|
|
||||||
|
|
||||||
def _open_local(repo_path: str) -> git.Repo:
|
def _open_local(repo_path: str) -> git.Repo:
|
||||||
|
|
@ -107,6 +116,9 @@ def _cache_dir_for(remote_url: str) -> Path:
|
||||||
Linux ~/.cache/ifcurl/<hash>
|
Linux ~/.cache/ifcurl/<hash>
|
||||||
macOS ~/Library/Caches/ifcurl/<hash>
|
macOS ~/Library/Caches/ifcurl/<hash>
|
||||||
Windows %LOCALAPPDATA%\\ifcurl\\Cache\\<hash>
|
Windows %LOCALAPPDATA%\\ifcurl\\Cache\\<hash>
|
||||||
|
|
||||||
|
The cache key is derived from the clean URL without any token so the same
|
||||||
|
cached clone is shared regardless of which credential was used to fetch it.
|
||||||
"""
|
"""
|
||||||
url_hash = hashlib.sha256(remote_url.encode()).hexdigest()[:24]
|
url_hash = hashlib.sha256(remote_url.encode()).hexdigest()[:24]
|
||||||
cache_dir = Path(user_cache_dir("ifcurl")) / url_hash
|
cache_dir = Path(user_cache_dir("ifcurl")) / url_hash
|
||||||
|
|
@ -114,28 +126,45 @@ def _cache_dir_for(remote_url: str) -> Path:
|
||||||
return cache_dir
|
return cache_dir
|
||||||
|
|
||||||
|
|
||||||
def _open_remote(remote_url: str, is_mutable: bool) -> git.Repo:
|
def _auth_url(remote_url: str, token: str | None) -> str:
|
||||||
|
"""Return *remote_url* with the token injected, or the original URL."""
|
||||||
|
if token and remote_url.startswith("https://"):
|
||||||
|
from ifcurl.auth import inject_token
|
||||||
|
return inject_token(remote_url, token)
|
||||||
|
return remote_url
|
||||||
|
|
||||||
|
|
||||||
|
def _open_remote(remote_url: str, is_mutable: bool, token: str | None = None) -> git.Repo:
|
||||||
"""Return a GitPython Repo for *remote_url*, cloning it if necessary.
|
"""Return a GitPython Repo for *remote_url*, cloning it if necessary.
|
||||||
|
|
||||||
Bare clones are stored under ~/.cache/ifcurl/<url-hash>. For mutable refs
|
Bare clones are stored under the OS cache dir keyed on the clean URL.
|
||||||
(branches, HEAD) the remote is fetched on every call so the cache stays
|
For mutable refs (branches, HEAD) the remote is fetched on every call.
|
||||||
current. Immutable refs (commit hashes, tags) skip the fetch.
|
Immutable refs (commit hashes, tags) skip the fetch.
|
||||||
|
|
||||||
|
If *token* is provided it is injected into the HTTPS URL for clone and
|
||||||
|
fetch operations but is never written to the on-disk git config.
|
||||||
"""
|
"""
|
||||||
cache_dir = _cache_dir_for(remote_url)
|
cache_dir = _cache_dir_for(remote_url)
|
||||||
git_dir = cache_dir / "repo.git"
|
git_dir = cache_dir / "repo.git"
|
||||||
|
auth = _auth_url(remote_url, token)
|
||||||
|
|
||||||
if not git_dir.exists():
|
if not git_dir.exists():
|
||||||
try:
|
try:
|
||||||
repo = git.Repo.clone_from(remote_url, str(git_dir), bare=True)
|
repo = git.Repo.clone_from(auth, str(git_dir), bare=True)
|
||||||
except git.exc.GitCommandError as exc:
|
except git.exc.GitCommandError as exc:
|
||||||
raise ValueError(f"Failed to clone {remote_url!r}: {exc.stderr.strip()}") from exc
|
raise ValueError(f"Failed to clone {remote_url!r}: {exc.stderr.strip()}") from exc
|
||||||
else:
|
else:
|
||||||
repo = git.Repo(str(git_dir))
|
repo = git.Repo(str(git_dir))
|
||||||
if is_mutable:
|
if is_mutable:
|
||||||
try:
|
try:
|
||||||
|
if auth != remote_url:
|
||||||
|
# Fetch from the authenticated URL directly so the token
|
||||||
|
# is never stored in the repo config.
|
||||||
|
repo.git.fetch(auth, "+refs/*:refs/*")
|
||||||
|
else:
|
||||||
repo.git.fetch("origin")
|
repo.git.fetch("origin")
|
||||||
except git.exc.GitCommandError:
|
except git.exc.GitCommandError:
|
||||||
pass # offline or no remote — use cached data
|
pass # offline — use cached data
|
||||||
|
|
||||||
return repo
|
return repo
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -50,6 +50,7 @@ from platformdirs import user_cache_dir
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
|
|
||||||
from ifcurl import render as render_mod
|
from ifcurl import render as render_mod
|
||||||
|
from ifcurl.auth import get_token_for_host
|
||||||
from ifcurl.git import fetch_ifc
|
from ifcurl.git import fetch_ifc
|
||||||
from ifcurl.url import IfcUrl
|
from ifcurl.url import IfcUrl
|
||||||
|
|
||||||
|
|
@ -173,6 +174,13 @@ def _load_model(ifc_bytes: bytes) -> ifcopenshell.file:
|
||||||
|
|
||||||
class PreviewRequest(BaseModel):
|
class PreviewRequest(BaseModel):
|
||||||
url: str
|
url: str
|
||||||
|
token: str | None = None
|
||||||
|
"""Optional bearer token for git authentication.
|
||||||
|
|
||||||
|
When provided, takes precedence over any token configured in
|
||||||
|
``~/.config/ifcurl/tokens.json``. Intended for co-located Gitea
|
||||||
|
deployments that pass the requesting user's session token.
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
@ -201,9 +209,14 @@ def preview(request: PreviewRequest) -> Response:
|
||||||
if cached_png is not None:
|
if cached_png is not None:
|
||||||
return Response(content=cached_png, media_type="image/png")
|
return Response(content=cached_png, media_type="image/png")
|
||||||
|
|
||||||
|
# --- Resolve authentication token ---
|
||||||
|
token = request.token
|
||||||
|
if token is None and ifc_url.host:
|
||||||
|
token = get_token_for_host(ifc_url.host)
|
||||||
|
|
||||||
# --- Fetch IFC bytes + commit hexsha ---
|
# --- Fetch IFC bytes + commit hexsha ---
|
||||||
try:
|
try:
|
||||||
hexsha, ifc_bytes = fetch_ifc(ifc_url)
|
hexsha, ifc_bytes = fetch_ifc(ifc_url, token=token)
|
||||||
except ImportError as exc:
|
except ImportError as exc:
|
||||||
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
|
|
|
||||||
98
tests/conftest.py
Normal file
98
tests/conftest.py
Normal file
|
|
@ -0,0 +1,98 @@
|
||||||
|
"""Shared pytest fixtures for ifcurl tests."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import ifcopenshell.api.aggregate
|
||||||
|
import ifcopenshell.api.context
|
||||||
|
import ifcopenshell.api.geometry
|
||||||
|
import ifcopenshell.api.owner.settings
|
||||||
|
import ifcopenshell.api.project
|
||||||
|
import ifcopenshell.api.root
|
||||||
|
import ifcopenshell.api.spatial
|
||||||
|
import ifcopenshell.api.unit
|
||||||
|
import numpy as np
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# IFC model fixtures
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def _suppress_owner(f):
|
||||||
|
ifcopenshell.api.owner.settings.get_user = lambda ifc: (ifc.by_type("IfcPersonAndOrganization") or [None])[0]
|
||||||
|
ifcopenshell.api.owner.settings.get_application = lambda ifc: (ifc.by_type("IfcApplication") or [None])[0]
|
||||||
|
return f
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
def model_with_geometry():
|
||||||
|
"""IFC4 model with two walls that have geometric representations."""
|
||||||
|
f = ifcopenshell.api.project.create_file()
|
||||||
|
_suppress_owner(f)
|
||||||
|
|
||||||
|
project = ifcopenshell.api.root.create_entity(f, ifc_class="IfcProject", name="TestProject")
|
||||||
|
ifcopenshell.api.unit.assign_unit(f)
|
||||||
|
|
||||||
|
site = ifcopenshell.api.root.create_entity(f, ifc_class="IfcSite", name="TestSite")
|
||||||
|
building = ifcopenshell.api.root.create_entity(f, ifc_class="IfcBuilding", name="TestBuilding")
|
||||||
|
storey = ifcopenshell.api.root.create_entity(f, ifc_class="IfcBuildingStorey", name="Ground Floor")
|
||||||
|
|
||||||
|
ifcopenshell.api.aggregate.assign_object(f, products=[site], relating_object=project)
|
||||||
|
ifcopenshell.api.aggregate.assign_object(f, products=[building], relating_object=site)
|
||||||
|
ifcopenshell.api.aggregate.assign_object(f, products=[storey], relating_object=building)
|
||||||
|
|
||||||
|
model_ctx = ifcopenshell.api.context.add_context(f, context_type="Model")
|
||||||
|
body = ifcopenshell.api.context.add_context(
|
||||||
|
f, context_type="Model", context_identifier="Body",
|
||||||
|
target_view="MODEL_VIEW", parent=model_ctx,
|
||||||
|
)
|
||||||
|
|
||||||
|
wall1 = ifcopenshell.api.root.create_entity(f, ifc_class="IfcWall", name="Wall001")
|
||||||
|
rep1 = ifcopenshell.api.geometry.add_wall_representation(f, context=body, length=5, height=3, thickness=0.2)
|
||||||
|
ifcopenshell.api.geometry.assign_representation(f, product=wall1, representation=rep1)
|
||||||
|
ifcopenshell.api.spatial.assign_container(f, products=[wall1], relating_structure=storey)
|
||||||
|
|
||||||
|
wall2 = ifcopenshell.api.root.create_entity(f, ifc_class="IfcWall", name="Wall002")
|
||||||
|
rep2 = ifcopenshell.api.geometry.add_wall_representation(f, context=body, length=4, height=3, thickness=0.2)
|
||||||
|
ifcopenshell.api.geometry.assign_representation(f, product=wall2, representation=rep2)
|
||||||
|
ifcopenshell.api.spatial.assign_container(f, products=[wall2], relating_structure=storey)
|
||||||
|
matrix = np.eye(4)
|
||||||
|
matrix[1, 3] = 3.0
|
||||||
|
ifcopenshell.api.geometry.edit_object_placement(f, product=wall2, matrix=matrix)
|
||||||
|
|
||||||
|
return f
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Local git repo fixture
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
def local_ifc_repo(tmp_path_factory, model_with_geometry):
|
||||||
|
"""A local bare-ish git repo containing model.ifc committed at HEAD.
|
||||||
|
|
||||||
|
Returns a dict with keys:
|
||||||
|
path — absolute path to the working-tree repo
|
||||||
|
hexsha — the commit hexsha
|
||||||
|
bytes — raw IFC file bytes as committed
|
||||||
|
"""
|
||||||
|
import git as gitpkg
|
||||||
|
|
||||||
|
repo_dir = tmp_path_factory.mktemp("ifc_repo")
|
||||||
|
ifc_path = repo_dir / "model.ifc"
|
||||||
|
model_with_geometry.write(str(ifc_path))
|
||||||
|
|
||||||
|
repo = gitpkg.Repo.init(str(repo_dir))
|
||||||
|
with repo.config_writer() as cw:
|
||||||
|
cw.set_value("user", "name", "Test")
|
||||||
|
cw.set_value("user", "email", "test@example.com")
|
||||||
|
repo.index.add(["model.ifc"])
|
||||||
|
commit = repo.index.commit("Add model")
|
||||||
|
repo.create_tag("v1.0")
|
||||||
|
|
||||||
|
return {
|
||||||
|
"path": str(repo_dir),
|
||||||
|
"hexsha": commit.hexsha,
|
||||||
|
"bytes": ifc_path.read_bytes(),
|
||||||
|
}
|
||||||
74
tests/test_auth.py
Normal file
74
tests/test_auth.py
Normal file
|
|
@ -0,0 +1,74 @@
|
||||||
|
"""Tests for ifcurl.auth — token config and URL injection."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from ifcurl.auth import get_token_for_host, inject_token
|
||||||
|
|
||||||
|
|
||||||
|
class TestGetTokenForHost:
|
||||||
|
def test_returns_token_for_known_host(self, tmp_path, monkeypatch):
|
||||||
|
config = tmp_path / "tokens.json"
|
||||||
|
config.write_text(json.dumps({"hosts": {"github.com": "ghp_abc123"}}))
|
||||||
|
monkeypatch.setattr("ifcurl.auth.config_path", lambda: config)
|
||||||
|
assert get_token_for_host("github.com") == "ghp_abc123"
|
||||||
|
|
||||||
|
def test_returns_none_for_unknown_host(self, tmp_path, monkeypatch):
|
||||||
|
config = tmp_path / "tokens.json"
|
||||||
|
config.write_text(json.dumps({"hosts": {"github.com": "ghp_abc123"}}))
|
||||||
|
monkeypatch.setattr("ifcurl.auth.config_path", lambda: config)
|
||||||
|
assert get_token_for_host("gitlab.com") is None
|
||||||
|
|
||||||
|
def test_returns_none_when_file_missing(self, tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr("ifcurl.auth.config_path", lambda: tmp_path / "nonexistent.json")
|
||||||
|
assert get_token_for_host("github.com") is None
|
||||||
|
|
||||||
|
def test_returns_none_on_malformed_json(self, tmp_path, monkeypatch):
|
||||||
|
config = tmp_path / "tokens.json"
|
||||||
|
config.write_text("not json {{{")
|
||||||
|
monkeypatch.setattr("ifcurl.auth.config_path", lambda: config)
|
||||||
|
assert get_token_for_host("github.com") is None
|
||||||
|
|
||||||
|
def test_returns_none_for_empty_token(self, tmp_path, monkeypatch):
|
||||||
|
config = tmp_path / "tokens.json"
|
||||||
|
config.write_text(json.dumps({"hosts": {"github.com": ""}}))
|
||||||
|
monkeypatch.setattr("ifcurl.auth.config_path", lambda: config)
|
||||||
|
assert get_token_for_host("github.com") is None
|
||||||
|
|
||||||
|
def test_multiple_hosts(self, tmp_path, monkeypatch):
|
||||||
|
config = tmp_path / "tokens.json"
|
||||||
|
config.write_text(json.dumps({
|
||||||
|
"hosts": {
|
||||||
|
"github.com": "ghp_token",
|
||||||
|
"gitlab.example.com": "glpat_token",
|
||||||
|
}
|
||||||
|
}))
|
||||||
|
monkeypatch.setattr("ifcurl.auth.config_path", lambda: config)
|
||||||
|
assert get_token_for_host("github.com") == "ghp_token"
|
||||||
|
assert get_token_for_host("gitlab.example.com") == "glpat_token"
|
||||||
|
|
||||||
|
|
||||||
|
class TestInjectToken:
|
||||||
|
def test_basic_https_url(self):
|
||||||
|
result = inject_token("https://github.com/org/repo", "mytoken")
|
||||||
|
assert result == "https://mytoken@github.com/org/repo"
|
||||||
|
|
||||||
|
def test_preserves_path(self):
|
||||||
|
result = inject_token("https://example.com/org/repo", "tok")
|
||||||
|
assert result.endswith("/org/repo")
|
||||||
|
|
||||||
|
def test_preserves_custom_port(self):
|
||||||
|
result = inject_token("https://example.com:8080/org/repo", "tok")
|
||||||
|
assert "8080" in result
|
||||||
|
assert result == "https://tok@example.com:8080/org/repo"
|
||||||
|
|
||||||
|
def test_token_appears_before_host(self):
|
||||||
|
result = inject_token("https://github.com/org/repo", "ghp_xyz")
|
||||||
|
assert "ghp_xyz@github.com" in result
|
||||||
|
|
||||||
|
def test_scheme_preserved(self):
|
||||||
|
result = inject_token("https://github.com/org/repo", "tok")
|
||||||
|
assert result.startswith("https://")
|
||||||
91
tests/test_git.py
Normal file
91
tests/test_git.py
Normal file
|
|
@ -0,0 +1,91 @@
|
||||||
|
"""Tests for ifcurl.git — git repository access."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from ifcurl.git import fetch_ifc, fetch_ifc_bytes
|
||||||
|
from ifcurl.url import IfcUrl
|
||||||
|
|
||||||
|
|
||||||
|
def _local_url(repo_path: str, ref: str = "HEAD", path: str = "model.ifc") -> IfcUrl:
|
||||||
|
"""Build an ifc:// URL pointing at a local repo fixture."""
|
||||||
|
return IfcUrl.parse(f"ifc://{repo_path}@{ref}?path={path}")
|
||||||
|
|
||||||
|
|
||||||
|
class TestFetchIfc:
|
||||||
|
def test_returns_hexsha_and_bytes(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"])
|
||||||
|
hexsha, data = fetch_ifc(url)
|
||||||
|
assert len(hexsha) == 40
|
||||||
|
assert all(c in "0123456789abcdef" for c in hexsha)
|
||||||
|
assert isinstance(data, bytes)
|
||||||
|
assert len(data) > 0
|
||||||
|
|
||||||
|
def test_hexsha_matches_commit(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"])
|
||||||
|
hexsha, _ = fetch_ifc(url)
|
||||||
|
assert hexsha == local_ifc_repo["hexsha"]
|
||||||
|
|
||||||
|
def test_bytes_match_committed_file(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"])
|
||||||
|
_, data = fetch_ifc(url)
|
||||||
|
assert data == local_ifc_repo["bytes"]
|
||||||
|
|
||||||
|
def test_explicit_commit_hash_ref(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"], ref=local_ifc_repo["hexsha"])
|
||||||
|
hexsha, data = fetch_ifc(url)
|
||||||
|
assert hexsha == local_ifc_repo["hexsha"]
|
||||||
|
assert data == local_ifc_repo["bytes"]
|
||||||
|
|
||||||
|
def test_tag_ref(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"], ref="tags/v1.0")
|
||||||
|
hexsha, data = fetch_ifc(url)
|
||||||
|
assert hexsha == local_ifc_repo["hexsha"]
|
||||||
|
|
||||||
|
def test_bad_ref_raises(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"], ref="heads/nonexistent")
|
||||||
|
with pytest.raises(ValueError, match="not found"):
|
||||||
|
fetch_ifc(url)
|
||||||
|
|
||||||
|
def test_bad_path_raises(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"], path="missing.ifc")
|
||||||
|
with pytest.raises(ValueError, match="not found"):
|
||||||
|
fetch_ifc(url)
|
||||||
|
|
||||||
|
def test_directory_path_raises(self, local_ifc_repo):
|
||||||
|
# Committing a subdir would be complex; test that asking for a
|
||||||
|
# non-blob raises — here we ask for the root tree entry.
|
||||||
|
url = _local_url(local_ifc_repo["path"], path=".")
|
||||||
|
with pytest.raises((ValueError, KeyError)):
|
||||||
|
fetch_ifc(url)
|
||||||
|
|
||||||
|
def test_no_path_raises(self, local_ifc_repo):
|
||||||
|
url = IfcUrl.parse(f"ifc://{local_ifc_repo['path']}@HEAD")
|
||||||
|
with pytest.raises(ValueError, match="'path' parameter"):
|
||||||
|
fetch_ifc(url)
|
||||||
|
|
||||||
|
|
||||||
|
class TestFetchIfcBytes:
|
||||||
|
def test_returns_bytes(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"])
|
||||||
|
data = fetch_ifc_bytes(url)
|
||||||
|
assert data == local_ifc_repo["bytes"]
|
||||||
|
|
||||||
|
def test_backwards_compat_with_fetch_ifc(self, local_ifc_repo):
|
||||||
|
url = _local_url(local_ifc_repo["path"])
|
||||||
|
_, expected = fetch_ifc(url)
|
||||||
|
assert fetch_ifc_bytes(url) == expected
|
||||||
|
|
||||||
|
|
||||||
|
class TestInvalidRepo:
|
||||||
|
def test_nonexistent_path_raises(self, tmp_path):
|
||||||
|
url = IfcUrl.parse(f"ifc://{tmp_path / 'no_such_repo'}@HEAD?path=m.ifc")
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
fetch_ifc(url)
|
||||||
|
|
||||||
|
def test_non_git_directory_raises(self, tmp_path):
|
||||||
|
(tmp_path / "model.ifc").write_text("not a repo")
|
||||||
|
url = IfcUrl.parse(f"ifc://{tmp_path}@HEAD?path=model.ifc")
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
fetch_ifc(url)
|
||||||
185
tests/test_render.py
Normal file
185
tests/test_render.py
Normal file
|
|
@ -0,0 +1,185 @@
|
||||||
|
"""Tests for ifcurl.render — PNG rendering with camera, clips, and visibility."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from ifcurl.render import render
|
||||||
|
|
||||||
|
try:
|
||||||
|
import pyvista # noqa: F401
|
||||||
|
HAS_PYVISTA = True
|
||||||
|
except ImportError:
|
||||||
|
HAS_PYVISTA = False
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.skipif(not HAS_PYVISTA, reason="pyvista not installed")
|
||||||
|
|
||||||
|
PNG_MAGIC = b"\x89PNG"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Basic rendering
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestRenderBasic:
|
||||||
|
def test_returns_png_bytes(self, model_with_geometry):
|
||||||
|
result = render(model_with_geometry)
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_no_geometry_raises(self):
|
||||||
|
import ifcopenshell.api.aggregate
|
||||||
|
import ifcopenshell.api.owner.settings
|
||||||
|
import ifcopenshell.api.project
|
||||||
|
import ifcopenshell.api.root
|
||||||
|
import ifcopenshell.api.spatial
|
||||||
|
import ifcopenshell.api.unit
|
||||||
|
|
||||||
|
f = ifcopenshell.api.project.create_file()
|
||||||
|
ifcopenshell.api.owner.settings.get_user = lambda ifc: None
|
||||||
|
ifcopenshell.api.owner.settings.get_application = lambda ifc: None
|
||||||
|
project = ifcopenshell.api.root.create_entity(f, ifc_class="IfcProject", name="P")
|
||||||
|
ifcopenshell.api.unit.assign_unit(f)
|
||||||
|
site = ifcopenshell.api.root.create_entity(f, ifc_class="IfcSite")
|
||||||
|
building = ifcopenshell.api.root.create_entity(f, ifc_class="IfcBuilding")
|
||||||
|
storey = ifcopenshell.api.root.create_entity(f, ifc_class="IfcBuildingStorey")
|
||||||
|
ifcopenshell.api.aggregate.assign_object(f, products=[site], relating_object=project)
|
||||||
|
ifcopenshell.api.aggregate.assign_object(f, products=[building], relating_object=site)
|
||||||
|
ifcopenshell.api.aggregate.assign_object(f, products=[storey], relating_object=building)
|
||||||
|
wall = ifcopenshell.api.root.create_entity(f, ifc_class="IfcWall", name="W")
|
||||||
|
ifcopenshell.api.spatial.assign_container(f, products=[wall], relating_structure=storey)
|
||||||
|
with pytest.raises(ValueError, match="No renderable geometry"):
|
||||||
|
render(f)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Selector
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestRenderSelector:
|
||||||
|
def test_selector_returns_png(self, model_with_geometry):
|
||||||
|
result = render(model_with_geometry, selector="IfcWall")
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_selector_no_match_raises(self, model_with_geometry):
|
||||||
|
with pytest.raises(ValueError, match="matched no elements"):
|
||||||
|
render(model_with_geometry, selector="IfcDoor")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Camera
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestRenderCamera:
|
||||||
|
def test_explicit_perspective_camera(self, model_with_geometry):
|
||||||
|
# Camera above and looking down
|
||||||
|
result = render(
|
||||||
|
model_with_geometry,
|
||||||
|
camera=(0.0, 0.0, 20.0, 0.0, 0.0, -1.0, 0.0, 1.0, 0.0),
|
||||||
|
fov=60.0,
|
||||||
|
)
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_explicit_orthographic_camera(self, model_with_geometry):
|
||||||
|
result = render(
|
||||||
|
model_with_geometry,
|
||||||
|
camera=(0.0, 0.0, 20.0, 0.0, 0.0, -1.0, 0.0, 1.0, 0.0),
|
||||||
|
scale=10.0,
|
||||||
|
)
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_no_camera_auto_fits(self, model_with_geometry):
|
||||||
|
# No camera → reset_camera() called, should still produce valid PNG
|
||||||
|
result = render(model_with_geometry)
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_perspective_and_orthographic_differ(self, model_with_geometry):
|
||||||
|
cam = (0.0, 0.0, 20.0, 0.0, 0.0, -1.0, 0.0, 1.0, 0.0)
|
||||||
|
persp = render(model_with_geometry, camera=cam, fov=60.0)
|
||||||
|
ortho = render(model_with_geometry, camera=cam, scale=10.0)
|
||||||
|
assert persp != ortho
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Clipping planes
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestRenderClips:
|
||||||
|
def test_clip_plane_returns_png(self, model_with_geometry):
|
||||||
|
# Clip at z=1.5, keeping below (normal points down -Z)
|
||||||
|
result = render(
|
||||||
|
model_with_geometry,
|
||||||
|
clips=[(0.0, 0.0, 1.5, 0.0, 0.0, -1.0)],
|
||||||
|
)
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_multiple_clip_planes(self, model_with_geometry):
|
||||||
|
result = render(
|
||||||
|
model_with_geometry,
|
||||||
|
clips=[
|
||||||
|
(0.0, 0.0, 2.0, 0.0, 0.0, -1.0),
|
||||||
|
(0.0, 0.0, 0.5, 0.0, 0.0, 1.0),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_clip_produces_different_image(self, model_with_geometry):
|
||||||
|
no_clip = render(model_with_geometry)
|
||||||
|
clipped = render(model_with_geometry, clips=[(0.0, 0.0, 1.0, 0.0, 0.0, -1.0)])
|
||||||
|
assert no_clip != clipped
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Visibility modes
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestRenderVisibility:
|
||||||
|
def test_highlight_mode(self, model_with_geometry):
|
||||||
|
wall = model_with_geometry.by_type("IfcWall")[0]
|
||||||
|
result = render(model_with_geometry, element_ids=[wall.id()], visibility="highlight")
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_ghost_mode(self, model_with_geometry):
|
||||||
|
wall = model_with_geometry.by_type("IfcWall")[0]
|
||||||
|
result = render(model_with_geometry, element_ids=[wall.id()], visibility="ghost")
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_isolate_mode(self, model_with_geometry):
|
||||||
|
wall = model_with_geometry.by_type("IfcWall")[0]
|
||||||
|
result = render(model_with_geometry, element_ids=[wall.id()], visibility="isolate")
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
def test_visibility_modes_produce_different_images(self, model_with_geometry):
|
||||||
|
wall = model_with_geometry.by_type("IfcWall")[0]
|
||||||
|
highlight = render(model_with_geometry, element_ids=[wall.id()], visibility="highlight")
|
||||||
|
ghost = render(model_with_geometry, element_ids=[wall.id()], visibility="ghost")
|
||||||
|
isolate = render(model_with_geometry, element_ids=[wall.id()], visibility="isolate")
|
||||||
|
# All three are valid PNG but should differ in content
|
||||||
|
assert highlight[:4] == ghost[:4] == isolate[:4] == PNG_MAGIC
|
||||||
|
assert len({highlight, ghost, isolate}) > 1
|
||||||
|
|
||||||
|
def test_selector_with_visibility(self, model_with_geometry):
|
||||||
|
result = render(model_with_geometry, selector="IfcWall", visibility="ghost")
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Combined parameters
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestRenderCombined:
|
||||||
|
def test_selector_camera_clip(self, model_with_geometry):
|
||||||
|
result = render(
|
||||||
|
model_with_geometry,
|
||||||
|
selector="IfcWall",
|
||||||
|
camera=(0.0, 0.0, 20.0, 0.0, 0.0, -1.0, 0.0, 1.0, 0.0),
|
||||||
|
fov=45.0,
|
||||||
|
clips=[(0.0, 0.0, 2.0, 0.0, 0.0, -1.0)],
|
||||||
|
visibility="ghost",
|
||||||
|
)
|
||||||
|
assert result[:4] == PNG_MAGIC
|
||||||
269
tests/test_service.py
Normal file
269
tests/test_service.py
Normal file
|
|
@ -0,0 +1,269 @@
|
||||||
|
"""Tests for ifcurl.service — POST /preview endpoint and caching."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from ifcurl.service import _t2_cache, _t3_cache, app
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
# Fake values used across tests
|
||||||
|
FAKE_HEXSHA = "a" * 40
|
||||||
|
FAKE_PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64
|
||||||
|
|
||||||
|
MUTABLE_URL = "ifc://example.com/org/repo@heads/main?path=model.ifc"
|
||||||
|
IMMUTABLE_URL = f"ifc://example.com/org/repo@{FAKE_HEXSHA}?path=model.ifc"
|
||||||
|
# Mutable ref so tier-4 is never checked, keeping tier-3 tests hermetic
|
||||||
|
SELECTOR_URL = "ifc://example.com/org/repo@heads/main?path=model.ifc&selector=IfcWall"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def clear_caches():
|
||||||
|
"""Reset in-memory caches and tier-4 filesystem cache between tests."""
|
||||||
|
_t2_cache.clear()
|
||||||
|
_t3_cache.clear()
|
||||||
|
yield
|
||||||
|
_t2_cache.clear()
|
||||||
|
_t3_cache.clear()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def tmp_t4_cache(tmp_path, monkeypatch):
|
||||||
|
"""Redirect tier-4 PNG cache to a temp directory."""
|
||||||
|
monkeypatch.setattr("ifcurl.service.user_cache_dir", lambda *a, **kw: str(tmp_path))
|
||||||
|
return tmp_path
|
||||||
|
|
||||||
|
|
||||||
|
def _mock_fetch(ifc_bytes: bytes):
|
||||||
|
return lambda ifc_url, token=None: (FAKE_HEXSHA, ifc_bytes)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Input validation
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestValidation:
|
||||||
|
def test_invalid_url_scheme_returns_400(self):
|
||||||
|
r = client.post("/preview", json={"url": "https://example.com/model.ifc"})
|
||||||
|
assert r.status_code == 400
|
||||||
|
|
||||||
|
def test_missing_url_field_returns_422(self):
|
||||||
|
r = client.post("/preview", json={})
|
||||||
|
assert r.status_code == 422
|
||||||
|
|
||||||
|
def test_url_without_path_param_returns_400(self):
|
||||||
|
r = client.post("/preview", json={"url": "ifc://example.com/org/repo@heads/main"})
|
||||||
|
assert r.status_code == 400
|
||||||
|
assert "path" in r.json()["detail"]
|
||||||
|
|
||||||
|
def test_malformed_url_returns_400(self):
|
||||||
|
r = client.post("/preview", json={"url": "ifc://example.com/repo_no_ref?path=m.ifc"})
|
||||||
|
assert r.status_code == 400
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Successful render
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestPreviewRender:
|
||||||
|
def test_returns_png_content_type(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
r = client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
assert r.status_code == 200
|
||||||
|
assert r.headers["content-type"] == "image/png"
|
||||||
|
assert r.content == FAKE_PNG
|
||||||
|
|
||||||
|
def test_fetch_error_returns_404(self):
|
||||||
|
with patch("ifcurl.service.fetch_ifc", side_effect=ValueError("Ref not found")):
|
||||||
|
r = client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
assert r.status_code == 404
|
||||||
|
|
||||||
|
def test_render_error_returns_422(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", side_effect=ValueError("No geometry")):
|
||||||
|
r = client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
assert r.status_code == 422
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 2: byte cache
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestTier2Cache:
|
||||||
|
def test_bytes_cached_after_first_request(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)) as mock_fetch, \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
assert (FAKE_HEXSHA, "model.ifc") in _t2_cache
|
||||||
|
|
||||||
|
def test_second_request_uses_cached_bytes(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
call_count = 0
|
||||||
|
|
||||||
|
def counting_fetch(ifc_url, token=None):
|
||||||
|
nonlocal call_count
|
||||||
|
call_count += 1
|
||||||
|
return FAKE_HEXSHA, ifc_bytes
|
||||||
|
|
||||||
|
with patch("ifcurl.service.fetch_ifc", counting_fetch), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
# fetch_ifc still called both times (for the commit hexsha),
|
||||||
|
# but _t2_cache hit means ifc_bytes comes from cache on 2nd call
|
||||||
|
assert call_count == 2
|
||||||
|
assert (FAKE_HEXSHA, "model.ifc") in _t2_cache
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 3: GUID/selector cache
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestTier3Cache:
|
||||||
|
def test_guids_cached_after_selector_request(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": SELECTOR_URL})
|
||||||
|
assert (FAKE_HEXSHA, "model.ifc", "IfcWall") in _t3_cache
|
||||||
|
|
||||||
|
def test_cached_guids_are_frozenset(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": SELECTOR_URL})
|
||||||
|
cached = _t3_cache[(FAKE_HEXSHA, "model.ifc", "IfcWall")]
|
||||||
|
assert isinstance(cached, frozenset)
|
||||||
|
|
||||||
|
def test_second_selector_request_skips_filter_elements(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
filter_call_count = 0
|
||||||
|
real_filter = __import__("ifcopenshell.util.selector", fromlist=["filter_elements"]).filter_elements
|
||||||
|
|
||||||
|
def counting_filter(model, selector):
|
||||||
|
nonlocal filter_call_count
|
||||||
|
filter_call_count += 1
|
||||||
|
return real_filter(model, selector)
|
||||||
|
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG), \
|
||||||
|
patch("ifcopenshell.util.selector.filter_elements", counting_filter):
|
||||||
|
client.post("/preview", json={"url": SELECTOR_URL})
|
||||||
|
client.post("/preview", json={"url": SELECTOR_URL})
|
||||||
|
|
||||||
|
assert filter_call_count == 1 # only called on first request
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 4: PNG filesystem cache
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestTier4Cache:
|
||||||
|
def test_png_cached_for_immutable_ref(self, tmp_t4_cache, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
render_call_count = 0
|
||||||
|
|
||||||
|
def counting_render(*args, **kwargs):
|
||||||
|
nonlocal render_call_count
|
||||||
|
render_call_count += 1
|
||||||
|
return FAKE_PNG
|
||||||
|
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", counting_render):
|
||||||
|
r1 = client.post("/preview", json={"url": IMMUTABLE_URL})
|
||||||
|
r2 = client.post("/preview", json={"url": IMMUTABLE_URL})
|
||||||
|
|
||||||
|
assert r1.content == FAKE_PNG
|
||||||
|
assert r2.content == FAKE_PNG
|
||||||
|
assert render_call_count == 1 # render only called once
|
||||||
|
|
||||||
|
def test_png_not_cached_for_mutable_ref(self, tmp_t4_cache, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
render_call_count = 0
|
||||||
|
|
||||||
|
def counting_render(*args, **kwargs):
|
||||||
|
nonlocal render_call_count
|
||||||
|
render_call_count += 1
|
||||||
|
return FAKE_PNG
|
||||||
|
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", counting_render):
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
|
||||||
|
assert render_call_count == 2 # no tier-4 cache for mutable ref
|
||||||
|
|
||||||
|
def test_tier4_cache_file_exists_on_disk(self, tmp_t4_cache, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": IMMUTABLE_URL})
|
||||||
|
png_files = list(tmp_t4_cache.rglob("*.png"))
|
||||||
|
assert len(png_files) == 1
|
||||||
|
assert png_files[0].read_bytes() == FAKE_PNG
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Authentication
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestAuthentication:
|
||||||
|
def test_request_token_passed_to_fetch_ifc(self, model_with_geometry):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
received_token = []
|
||||||
|
|
||||||
|
def capturing_fetch(ifc_url, token=None):
|
||||||
|
received_token.append(token)
|
||||||
|
return FAKE_HEXSHA, ifc_bytes
|
||||||
|
|
||||||
|
with patch("ifcurl.service.fetch_ifc", capturing_fetch), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL, "token": "mytoken123"})
|
||||||
|
|
||||||
|
assert received_token == ["mytoken123"]
|
||||||
|
|
||||||
|
def test_config_token_used_when_no_request_token(self, model_with_geometry, monkeypatch):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
received_token = []
|
||||||
|
|
||||||
|
def capturing_fetch(ifc_url, token=None):
|
||||||
|
received_token.append(token)
|
||||||
|
return FAKE_HEXSHA, ifc_bytes
|
||||||
|
|
||||||
|
monkeypatch.setattr("ifcurl.service.get_token_for_host", lambda host: "config_token")
|
||||||
|
with patch("ifcurl.service.fetch_ifc", capturing_fetch), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL})
|
||||||
|
|
||||||
|
assert received_token == ["config_token"]
|
||||||
|
|
||||||
|
def test_request_token_overrides_config_token(self, model_with_geometry, monkeypatch):
|
||||||
|
ifc_bytes = model_with_geometry.to_string().encode()
|
||||||
|
received_token = []
|
||||||
|
|
||||||
|
def capturing_fetch(ifc_url, token=None):
|
||||||
|
received_token.append(token)
|
||||||
|
return FAKE_HEXSHA, ifc_bytes
|
||||||
|
|
||||||
|
monkeypatch.setattr("ifcurl.service.get_token_for_host", lambda host: "config_token")
|
||||||
|
with patch("ifcurl.service.fetch_ifc", capturing_fetch), \
|
||||||
|
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
|
||||||
|
client.post("/preview", json={"url": MUTABLE_URL, "token": "request_token"})
|
||||||
|
|
||||||
|
assert received_token == ["request_token"]
|
||||||
Loading…
Add table
Reference in a new issue