updated testing workflow
This commit is contained in:
Andrea Pozzetti
@@ -0,0 +1,8 @@
|
||||
name: Test Installation
|
||||
on: [push, pull_request]
|
||||
|
||||
jobs:
|
||||
test:
|
||||
uses: PozzettiAndrea/comfy-test/.github/workflows/test-matrix.yml@main
|
||||
with:
|
||||
config-file: "comfy-test.toml"
|
||||
@@ -1,59 +0,0 @@
|
||||
name: Tests
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main, dev]
|
||||
pull_request:
|
||||
branches: [main, dev]
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ${{ matrix.os }}
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
os: [ubuntu-latest, macos-latest, windows-latest]
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python 3.10
|
||||
uses: actions/setup-python@v4
|
||||
with:
|
||||
python-version: "3.10"
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install -r requirements.txt
|
||||
pip install -r tests/requirements-test.txt
|
||||
|
||||
- name: Run unit tests
|
||||
run: |
|
||||
pytest tests/ -m "unit" -v --tb=short --ignore=tests/browser_tests
|
||||
|
||||
- name: Run optional dependency tests
|
||||
continue-on-error: true
|
||||
run: |
|
||||
pytest tests/ -m "optional" -v --tb=short --ignore=tests/browser_tests
|
||||
|
||||
- name: Upload test outputs
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: test-outputs-${{ matrix.os }}
|
||||
path: tests/outputs/
|
||||
retention-days: 7
|
||||
|
||||
- name: Generate coverage report
|
||||
if: matrix.os == 'ubuntu-latest'
|
||||
run: |
|
||||
pip install pytest-cov
|
||||
pytest tests/ -m "unit" --cov=nodes --cov-report=xml --cov-report=html --ignore=tests/browser_tests
|
||||
|
||||
- name: Upload coverage
|
||||
if: matrix.os == 'ubuntu-latest'
|
||||
uses: codecov/codecov-action@v4
|
||||
with:
|
||||
file: ./coverage.xml
|
||||
fail_ci_if_error: false
|
||||
@@ -0,0 +1,19 @@
|
||||
[test]
|
||||
name = "ComfyUI-GeometryPack"
|
||||
python_version = "3.11"
|
||||
cpu_only = true
|
||||
timeout = 300
|
||||
levels = ["syntax", "install", "registration", "instantiation", "validation"]
|
||||
|
||||
[test.workflows]
|
||||
files = [
|
||||
"workflows/mesh_info.json",
|
||||
"workflows/transform.json",
|
||||
"workflows/boolean_ops.json",
|
||||
]
|
||||
timeout = 120
|
||||
|
||||
[test.platforms]
|
||||
linux = true
|
||||
windows = true
|
||||
windows_portable = true
|
||||
@@ -1,25 +0,0 @@
|
||||
# Test outputs - never commit these!
|
||||
outputs/
|
||||
|
||||
# Pytest cache
|
||||
__pycache__/
|
||||
.pytest_cache/
|
||||
*.pyc
|
||||
*.pyo
|
||||
|
||||
# Coverage reports
|
||||
.coverage
|
||||
htmlcov/
|
||||
coverage.xml
|
||||
*.cover
|
||||
|
||||
# IDEs
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
*.swo
|
||||
*~
|
||||
|
||||
# OS files
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
-191
@@ -1,191 +0,0 @@
|
||||
# ComfyUI-GeometryPack Test Suite
|
||||
|
||||
Comprehensive test suite for all GeometryPack nodes with visual output verification.
|
||||
|
||||
## Example Test Results
|
||||
|
||||
### Stanford Bunny Processing
|
||||
|
||||
|
||||
### Remeshing Operations
|
||||
|
||||
|
||||
Each test generates its own output folder with before/after meshes and rendered images for visual verification.
|
||||
|
||||
---
|
||||
|
||||
## Directory Structure
|
||||
|
||||
```
|
||||
tests/
|
||||
├── test_scripts/ # All test files and fixtures
|
||||
│ ├── conftest.py # Pytest fixtures and helpers
|
||||
│ ├── test_*.py # Individual test modules
|
||||
│ └── .gitignore
|
||||
├── outputs/ # Generated test outputs
|
||||
│ ├── meshes/ # Saved mesh files (.obj, .ply, .stl)
|
||||
│ └── renders/ # PNG preview images
|
||||
├── assets/ # Test input files (bunny.stl, etc.)
|
||||
└── README.md # This file
|
||||
```
|
||||
|
||||
## Quick Start
|
||||
|
||||
```bash
|
||||
# Install test dependencies
|
||||
pip install -r requirements-dev.txt
|
||||
|
||||
# Run all unit tests
|
||||
pytest -m "unit" -v
|
||||
|
||||
# Run specific test file
|
||||
pytest test_scripts/test_primitives.py -v
|
||||
|
||||
# Run with output images
|
||||
pytest -v
|
||||
```
|
||||
|
||||
## Test Organization
|
||||
|
||||
### Unit Tests (Fast, No Optional Dependencies)
|
||||
|
||||
**test_io.py** - I/O operations
|
||||
- `test_load_mesh_stl` - Load STL file (Stanford Bunny)
|
||||
- `test_save_mesh_obj` - Save mesh to OBJ format
|
||||
- `test_save_mesh_ply` - Save mesh to PLY format
|
||||
- `test_save_load_cycle` - Round-trip save/load for all formats (OBJ, PLY, STL, OFF)
|
||||
|
||||
**test_primitives.py** - Primitive generation
|
||||
- `test_create_primitive_shapes` - Create cube, plane primitives
|
||||
- `test_create_sphere` - Create sphere and verify radius
|
||||
- `test_create_primitive_with_subdivisions` - Test subdivision levels
|
||||
- `test_create_primitive_sizes` - Test different primitive sizes
|
||||
|
||||
**test_analysis.py** - Mesh analysis
|
||||
- `test_mesh_info_cube` - Get mesh statistics for cube
|
||||
- `test_mesh_info_bunny` - Get mesh statistics for Stanford Bunny
|
||||
- `test_mark_boundary_edges_closed_mesh` - Boundary detection on closed mesh
|
||||
- `test_mark_boundary_edges_open_mesh` - Boundary detection on open mesh
|
||||
- `test_mesh_info_volume_area` - Verify volume/area calculations
|
||||
|
||||
**test_transforms.py** - Transformations
|
||||
- `test_center_mesh` - Center mesh at origin
|
||||
- `test_center_mesh_preserves_shape` - Verify shape preservation
|
||||
|
||||
**test_conversion.py** - Format conversions
|
||||
- `test_strip_mesh_adjacency` - Convert mesh to point cloud (no sampling)
|
||||
- `test_strip_mesh_with_normals` - Point cloud with normals
|
||||
- `test_mesh_to_point_cloud_sampling` - Sample point clouds (100, 500, 1000 points)
|
||||
- `test_mesh_to_point_cloud_methods` - Test uniform/even/face_weighted sampling
|
||||
- `test_point_cloud_with_normals` - Generate point clouds with normals
|
||||
|
||||
**test_repair.py** - Mesh repair
|
||||
- `test_fix_normals` - Fix inconsistent normals
|
||||
- `test_check_normals` - Check normal consistency
|
||||
- `test_fill_holes` - Fill holes in open mesh
|
||||
- `test_compute_normals_faceted` - Compute faceted normals
|
||||
- `test_compute_normals_smooth` - Compute smooth vertex normals
|
||||
- `test_visualize_normal_field` - Generate normal scalar fields
|
||||
|
||||
**test_remeshing.py** - Remeshing operations
|
||||
- `test_mesh_decimation` - Reduce triangle count
|
||||
- `test_mesh_subdivision` - Subdivide mesh (1, 2, 3 iterations)
|
||||
- `test_laplacian_smoothing` - Laplacian smoothing
|
||||
|
||||
**test_visualization.py** - Visualization exports
|
||||
- `test_preview_mesh_threejs` - Three.js preview (GLB export)
|
||||
- `test_preview_mesh_vtk` - VTK.js preview (STL export)
|
||||
- `test_preview_mesh_vtk_filters` - VTK.js with filters
|
||||
- `test_preview_mesh_vtk_fields` - VTK.js with scalar fields
|
||||
|
||||
### Optional Dependency Tests
|
||||
|
||||
**test_remeshing.py** - Advanced remeshing
|
||||
- `test_pymeshlab_remesh` - PyMeshLab isotropic remeshing (requires pymeshlab)
|
||||
- `test_cgal_remesh` - CGAL isotropic remeshing (requires cgal)
|
||||
- `test_instant_meshes_remesh` - InstantMeshes remeshing (requires PyNanoInstantMeshes)
|
||||
|
||||
**test_uv.py** - UV unwrapping
|
||||
- `test_xatlas_unwrap` - xAtlas UV unwrapping (requires xatlas)
|
||||
- `test_libigl_lscm` - libigl LSCM unwrapping (requires igl)
|
||||
- `test_libigl_harmonic` - libigl harmonic unwrapping (requires igl)
|
||||
- `test_libigl_arap` - libigl ARAP unwrapping (requires igl)
|
||||
|
||||
**test_distance.py** - Distance metrics
|
||||
- `test_hausdorff_distance_identical` - Hausdorff distance (identical meshes)
|
||||
- `test_hausdorff_distance_different` - Hausdorff distance (different meshes)
|
||||
- `test_chamfer_distance_identical` - Chamfer distance (identical meshes)
|
||||
- `test_chamfer_distance_different` - Chamfer distance (different meshes)
|
||||
- `test_compute_sdf` - Signed distance field generation (32, 64 resolution)
|
||||
|
||||
### Blender Tests (Slow)
|
||||
|
||||
**test_remeshing.py** - Blender remeshing
|
||||
- `test_blender_voxel_remesh` - Blender voxel remeshing
|
||||
- `test_blender_quadriflow_remesh` - Blender Quadriflow remeshing
|
||||
|
||||
**test_uv.py** - Blender UV unwrapping
|
||||
- `test_blender_uv_unwrap` - Blender Smart UV Project
|
||||
- `test_blender_cube_projection` - Cube projection UV
|
||||
- `test_blender_cylinder_projection` - Cylinder projection UV
|
||||
- `test_blender_sphere_projection` - Sphere projection UV
|
||||
|
||||
### Integration Tests
|
||||
|
||||
**test_integration.py** - Multi-node pipelines
|
||||
- `test_pipeline_create_analyze_save` - Create → Analyze → Save
|
||||
- `test_pipeline_load_remesh_save` - Load → Remesh → Save
|
||||
- `test_pipeline_create_subdivide_smooth` - Create → Subdivide → Smooth
|
||||
- `test_pipeline_repair_analyze` - Repair → Analyze
|
||||
- `test_pipeline_mesh_to_pointcloud` - Mesh → Point Cloud conversion
|
||||
- `test_pipeline_boundary_detection_visualization` - Boundary → Visualization
|
||||
- `test_pipeline_full_processing` - Full processing pipeline
|
||||
|
||||
## Running Specific Test Categories
|
||||
|
||||
```bash
|
||||
# Unit tests only (fastest)
|
||||
pytest tests/ -m "unit"
|
||||
|
||||
# Optional dependency tests
|
||||
pytest tests/ -m "optional"
|
||||
|
||||
# Blender tests (requires Blender installed)
|
||||
pytest tests/ -m "blender"
|
||||
|
||||
# Slow tests
|
||||
pytest tests/ -m "slow"
|
||||
|
||||
# Skip slow tests
|
||||
pytest tests/ -m "not slow"
|
||||
```
|
||||
|
||||
## Test Outputs
|
||||
|
||||
All tests generate visual outputs for manual verification:
|
||||
|
||||
- **Meshes**: `tests/outputs/meshes/` - OBJ, PLY, STL files
|
||||
- **Renders**: `tests/outputs/renders/` - PNG preview images
|
||||
|
||||
## Coverage
|
||||
|
||||
Run tests with coverage report:
|
||||
|
||||
```bash
|
||||
pytest tests/ --cov=nodes --cov-report=html
|
||||
open htmlcov/index.html
|
||||
```
|
||||
|
||||
## CI/CD
|
||||
|
||||
GitHub Actions runs tests automatically on:
|
||||
- Ubuntu, macOS, Windows
|
||||
- Python 3.10
|
||||
- Uploads test outputs as artifacts
|
||||
|
||||
## Test Statistics
|
||||
|
||||
- **Total test files**: 10
|
||||
- **Total test cases**: 60+
|
||||
- **Node coverage**: 36/36 nodes tested
|
||||
- **All tests run on CPU** - No GPU required
|
||||
@@ -1,246 +0,0 @@
|
||||
# Browser Tests for ComfyUI GeometryPack
|
||||
|
||||
Browser automation tests for VTK viewer UI using Selenium and pytest.
|
||||
|
||||
## Overview
|
||||
|
||||
These tests verify the VTK mesh viewer functionality in a real browser environment, including:
|
||||
- File format detection (STL, OBJ, VTP)
|
||||
- Viewer initialization and loading
|
||||
- Scalar field visualization
|
||||
- Widget-viewer integration
|
||||
|
||||
## Requirements
|
||||
|
||||
- **Selenium 4.15+** (already installed in your environment)
|
||||
- **ComfyUI running** on `http://localhost:8188` (or custom URL)
|
||||
- **Chrome/Safari/Firefox** browser installed
|
||||
|
||||
## Quick Start
|
||||
|
||||
### 1. Start ComfyUI
|
||||
|
||||
```bash
|
||||
# Make sure ComfyUI is running
|
||||
python main.py --port 8188
|
||||
```
|
||||
|
||||
### 2. Run All Browser Tests
|
||||
|
||||
```bash
|
||||
# Run all browser tests
|
||||
pytest tests/browser_tests/ -v
|
||||
|
||||
# Run with visible browser (not headless)
|
||||
pytest tests/browser_tests/ -v --headed
|
||||
|
||||
# Run specific test file
|
||||
pytest tests/browser_tests/test_vtk_viewer.py -v
|
||||
|
||||
# Run specific test
|
||||
pytest tests/browser_tests/test_vtk_viewer.py::TestVTKViewer::test_vtp_format_supported -v
|
||||
```
|
||||
|
||||
### 3. View Results
|
||||
|
||||
- **Screenshots on failure**: Saved to `test-screenshots/`
|
||||
- **Console logs on failure**: Saved to `test-screenshots/*.log`
|
||||
|
||||
## Test Files
|
||||
|
||||
### `test_vtk_viewer.py`
|
||||
Tests for the basic VTK viewer (`viewer_vtk.html`):
|
||||
- ✅ VTP format support (the bug we fixed)
|
||||
- ✅ STL/OBJ format still work
|
||||
- ✅ File format detection with query strings
|
||||
- ✅ Viewer controls present
|
||||
- ✅ Unsupported formats rejected
|
||||
|
||||
### `test_vtk_filters.py`
|
||||
Tests for the advanced filters viewer (`viewer_vtk_filters.html`):
|
||||
- ✅ VTP file loading
|
||||
- ✅ Scalar field selector exists
|
||||
- ✅ VTP prioritized over STL
|
||||
- ✅ postMessage communication
|
||||
- ✅ Multiple format support
|
||||
|
||||
### `test_vtk_fields.py`
|
||||
Tests for PreviewMeshVTKFields widget integration:
|
||||
- ✅ Widget loads correct viewer (filters, not basic)
|
||||
- ✅ Scalar field selector present
|
||||
- ✅ VTP files accepted
|
||||
- ✅ postMessage integration
|
||||
- ✅ Comparison between viewers
|
||||
|
||||
## Command Line Options
|
||||
|
||||
```bash
|
||||
# Browser selection
|
||||
pytest tests/browser_tests/ --browser=chrome # Default
|
||||
pytest tests/browser_tests/ --browser=safari
|
||||
pytest tests/browser_tests/ --browser=firefox
|
||||
|
||||
# Headed mode (see the browser)
|
||||
pytest tests/browser_tests/ --headed
|
||||
|
||||
# Custom ComfyUI URL
|
||||
pytest tests/browser_tests/ --comfyui-url=http://localhost:8000
|
||||
|
||||
# Run only browser tests
|
||||
pytest -m browser -v
|
||||
|
||||
# Skip browser tests
|
||||
pytest -m "not browser" -v
|
||||
```
|
||||
|
||||
## Writing New Tests
|
||||
|
||||
### Basic Test Template
|
||||
|
||||
```python
|
||||
import pytest
|
||||
|
||||
@pytest.mark.browser
|
||||
class TestMyFeature:
|
||||
def test_something(self, viewer_page):
|
||||
# Load the viewer
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Send a mesh load message
|
||||
viewer_page.send_load_mesh_message("test.vtp")
|
||||
|
||||
# Check console logs
|
||||
logs = viewer_page.get_console_logs()
|
||||
assert "VTP" in "\n".join(logs)
|
||||
|
||||
# Check elements
|
||||
assert viewer_page.check_element_exists("#myElement")
|
||||
```
|
||||
|
||||
## Fixtures
|
||||
|
||||
### `browser`
|
||||
Selenium WebDriver instance (Chrome/Safari/Firefox)
|
||||
- Auto-captures screenshots on failure
|
||||
- Auto-captures console logs on failure
|
||||
- Configurable headless/headed mode
|
||||
|
||||
### `viewer_page`
|
||||
Helper class for VTK viewer interactions:
|
||||
- `load_viewer(name)` - Load a specific viewer HTML
|
||||
- `send_load_mesh_message(filename)` - Simulate ComfyUI postMessage
|
||||
- `get_console_logs()` - Get browser console output
|
||||
- `check_element_exists(selector)` - Check if element present
|
||||
- `wait_for_text(text)` - Wait for text to appear
|
||||
|
||||
## CI/CD Integration
|
||||
|
||||
Add to `.github/workflows/test.yml`:
|
||||
|
||||
```yaml
|
||||
- name: Install Chrome
|
||||
uses: browser-actions/setup-chrome@latest
|
||||
|
||||
- name: Start ComfyUI
|
||||
run: |
|
||||
python main.py --port 8188 &
|
||||
sleep 10 # Wait for server to start
|
||||
|
||||
- name: Run browser tests
|
||||
run: |
|
||||
pytest tests/browser_tests/ -v --headed=false
|
||||
|
||||
- name: Upload screenshots on failure
|
||||
if: failure()
|
||||
uses: actions/upload-artifact@v3
|
||||
with:
|
||||
name: test-screenshots
|
||||
path: test-screenshots/
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Browser driver not found
|
||||
```bash
|
||||
# Install webdriver-manager for automatic driver management
|
||||
pip install webdriver-manager
|
||||
|
||||
# Or manually download ChromeDriver
|
||||
# https://chromedriver.chromium.org/downloads
|
||||
```
|
||||
|
||||
### ComfyUI not running
|
||||
```bash
|
||||
# Make sure ComfyUI is started first
|
||||
python main.py --port 8188
|
||||
|
||||
# Or use custom URL
|
||||
pytest tests/browser_tests/ --comfyui-url=http://your-server:8188
|
||||
```
|
||||
|
||||
### Tests failing with timeout
|
||||
- Increase timeout in `conftest.py`
|
||||
- Check if ComfyUI is accessible
|
||||
- Run with `--headed` to see what's happening
|
||||
|
||||
### Can't see what's happening
|
||||
```bash
|
||||
# Run with visible browser
|
||||
pytest tests/browser_tests/ --headed -v
|
||||
|
||||
# Add more sleeps to slow down
|
||||
# Edit test files and increase time.sleep() values
|
||||
```
|
||||
|
||||
## What These Tests Verify
|
||||
|
||||
These browser tests were created to verify the fix for:
|
||||
|
||||
**Issue**: PreviewMeshVTKFields node couldn't display VTP files or scalar fields
|
||||
|
||||
**Root Cause**:
|
||||
1. `viewer_vtk.html` didn't support VTP format
|
||||
2. `mesh_preview_vtk_fields.js` loaded wrong viewer (basic instead of filters)
|
||||
|
||||
**Fix Applied**:
|
||||
1. Added VTP support to `viewer_vtk.html` (imported `vtkXMLPolyDataReader`)
|
||||
2. Changed `mesh_preview_vtk_fields.js` to load `viewer_vtk_filters.html`
|
||||
|
||||
**Tests Verify**:
|
||||
- ✅ VTP files now load in basic viewer
|
||||
- ✅ Filters viewer has scalar field selector
|
||||
- ✅ Fields widget uses correct viewer
|
||||
- ✅ No regression in STL/OBJ support
|
||||
|
||||
## Examples
|
||||
|
||||
### Check if VTP fix works
|
||||
```bash
|
||||
pytest tests/browser_tests/test_vtk_viewer.py::TestVTKViewer::test_vtp_format_supported -v --headed
|
||||
```
|
||||
|
||||
### Verify filters viewer has scalar selector
|
||||
```bash
|
||||
pytest tests/browser_tests/test_vtk_filters.py::TestVTKFiltersViewer::test_scalar_field_selector_exists -v --headed
|
||||
```
|
||||
|
||||
### Compare basic vs filters viewer
|
||||
```bash
|
||||
pytest tests/browser_tests/test_vtk_fields.py::TestVTKFieldsWidget::test_comparison_basic_vs_filters_viewer -v --headed
|
||||
```
|
||||
|
||||
## Notes
|
||||
|
||||
- Tests use real browser (Chrome by default)
|
||||
- Tests can run headless or with visible browser
|
||||
- Screenshots saved automatically on failure
|
||||
- Console logs captured for debugging
|
||||
- Works with existing pytest infrastructure
|
||||
- Integrates with pytest markers (`-m browser`)
|
||||
|
||||
## Related Files
|
||||
|
||||
- `web/viewer_vtk.html` - Basic VTK viewer
|
||||
- `web/viewer_vtk_filters.html` - Advanced viewer with scalar fields
|
||||
- `web/js/mesh_preview_vtk_fields.js` - Fields widget (loads filters viewer)
|
||||
- `web/js/mesh_preview_vtk_filters.js` - Filters widget
|
||||
@@ -1,198 +0,0 @@
|
||||
"""
|
||||
Browser test fixtures for ComfyUI GeometryPack VTK viewers.
|
||||
|
||||
This module provides pytest fixtures for browser automation testing using Selenium.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import time
|
||||
import os
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.chrome.service import Service
|
||||
from selenium.webdriver.chrome.options import Options
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
from selenium.common.exceptions import NoSuchElementException
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def comfyui_url():
|
||||
"""Base URL for ComfyUI server. Override with --comfyui-url option."""
|
||||
return os.environ.get("COMFYUI_URL", "http://localhost:8188")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def browser(request, comfyui_url):
|
||||
"""
|
||||
Create a Selenium WebDriver instance for browser testing.
|
||||
|
||||
Supports Chrome, Safari, and Firefox. Chrome is the default.
|
||||
Use --headed flag to run with visible browser (default is headless).
|
||||
"""
|
||||
# Get options from command line
|
||||
headed = request.config.getoption("--headed", default=False)
|
||||
browser_name = request.config.getoption("--browser", default="chrome").lower()
|
||||
|
||||
# Set up Chrome (most common)
|
||||
if browser_name == "chrome":
|
||||
chrome_options = Options()
|
||||
if not headed:
|
||||
chrome_options.add_argument("--headless=new")
|
||||
chrome_options.add_argument("--no-sandbox")
|
||||
chrome_options.add_argument("--disable-dev-shm-usage")
|
||||
chrome_options.add_argument("--disable-gpu")
|
||||
chrome_options.add_argument("--window-size=1920,1080")
|
||||
|
||||
# Enable browser logging
|
||||
chrome_options.set_capability("goog:loggingPrefs", {"browser": "ALL"})
|
||||
|
||||
driver = webdriver.Chrome(options=chrome_options)
|
||||
|
||||
elif browser_name == "safari":
|
||||
# Safari on Mac
|
||||
driver = webdriver.Safari()
|
||||
driver.set_window_size(1920, 1080)
|
||||
|
||||
elif browser_name == "firefox":
|
||||
firefox_options = webdriver.FirefoxOptions()
|
||||
if not headed:
|
||||
firefox_options.add_argument("--headless")
|
||||
firefox_options.add_argument("--width=1920")
|
||||
firefox_options.add_argument("--height=1080")
|
||||
driver = webdriver.Firefox(options=firefox_options)
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unsupported browser: {browser_name}")
|
||||
|
||||
# Set implicit wait
|
||||
driver.implicitly_wait(10)
|
||||
|
||||
# Store base URL
|
||||
driver.comfyui_url = comfyui_url
|
||||
|
||||
yield driver
|
||||
|
||||
# Teardown: capture screenshot on failure
|
||||
if request.node.rep_call.failed:
|
||||
screenshot_dir = "test-screenshots"
|
||||
os.makedirs(screenshot_dir, exist_ok=True)
|
||||
screenshot_path = os.path.join(
|
||||
screenshot_dir,
|
||||
f"{request.node.name}_{int(time.time())}.png"
|
||||
)
|
||||
driver.save_screenshot(screenshot_path)
|
||||
print(f"\n📸 Screenshot saved: {screenshot_path}")
|
||||
|
||||
# Capture browser console logs
|
||||
try:
|
||||
logs = driver.get_log("browser")
|
||||
log_path = os.path.join(
|
||||
screenshot_dir,
|
||||
f"{request.node.name}_{int(time.time())}_console.log"
|
||||
)
|
||||
with open(log_path, "w") as f:
|
||||
for log in logs:
|
||||
f.write(f"[{log['level']}] {log['message']}\n")
|
||||
print(f"📝 Console logs saved: {log_path}")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not capture console logs: {e}")
|
||||
|
||||
driver.quit()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def viewer_page(browser, comfyui_url):
|
||||
"""
|
||||
Load a VTK viewer page and return helper methods for interacting with it.
|
||||
"""
|
||||
class ViewerPage:
|
||||
def __init__(self, driver, base_url):
|
||||
self.driver = driver
|
||||
self.base_url = base_url
|
||||
|
||||
def load_viewer(self, viewer_name="viewer_vtk.html"):
|
||||
"""Load a specific VTK viewer HTML file."""
|
||||
url = f"{self.base_url}/extensions/ComfyUI-GeometryPack/{viewer_name}"
|
||||
print(f"🌐 Loading viewer: {url}")
|
||||
self.driver.get(url)
|
||||
time.sleep(2) # Wait for vtk.js to initialize
|
||||
|
||||
def send_load_mesh_message(self, filename, file_type="output", subfolder=""):
|
||||
"""Send a LOAD_MESH postMessage to the viewer (simulating ComfyUI)."""
|
||||
filepath = f"/view?filename={filename}&type={file_type}&subfolder={subfolder}"
|
||||
script = f"""
|
||||
window.postMessage({{
|
||||
type: 'LOAD_MESH',
|
||||
filepath: '{filepath}',
|
||||
timestamp: Date.now()
|
||||
}}, '*');
|
||||
"""
|
||||
print(f"📤 Sending LOAD_MESH message: {filepath}")
|
||||
self.driver.execute_script(script)
|
||||
time.sleep(2) # Wait for mesh to load
|
||||
|
||||
def get_console_logs(self):
|
||||
"""Get browser console logs."""
|
||||
try:
|
||||
logs = self.driver.get_log("browser")
|
||||
return [f"[{log['level']}] {log['message']}" for log in logs]
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not get console logs: {e}")
|
||||
return []
|
||||
|
||||
def wait_for_text(self, text, timeout=10):
|
||||
"""Wait for specific text to appear on the page."""
|
||||
WebDriverWait(self.driver, timeout).until(
|
||||
EC.text_to_be_present_in_element((By.TAG_NAME, "body"), text)
|
||||
)
|
||||
|
||||
def check_element_exists(self, selector, by=By.CSS_SELECTOR):
|
||||
"""Check if an element exists on the page."""
|
||||
try:
|
||||
self.driver.find_element(by, selector)
|
||||
return True
|
||||
except NoSuchElementException:
|
||||
return False
|
||||
|
||||
def get_element_text(self, selector, by=By.CSS_SELECTOR):
|
||||
"""Get text content of an element."""
|
||||
try:
|
||||
element = self.driver.find_element(by, selector)
|
||||
return element.text
|
||||
except NoSuchElementException:
|
||||
return None
|
||||
|
||||
return ViewerPage(browser, comfyui_url)
|
||||
|
||||
|
||||
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
|
||||
def pytest_runtest_makereport(item, call):
|
||||
"""
|
||||
Hook to store test result in the request fixture for screenshot capture.
|
||||
"""
|
||||
outcome = yield
|
||||
rep = outcome.get_result()
|
||||
setattr(item, f"rep_{rep.when}", rep)
|
||||
|
||||
|
||||
def pytest_addoption(parser):
|
||||
"""Add custom command line options."""
|
||||
parser.addoption(
|
||||
"--headed",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help="Run browser tests with visible browser window (not headless)"
|
||||
)
|
||||
parser.addoption(
|
||||
"--browser",
|
||||
action="store",
|
||||
default="chrome",
|
||||
help="Browser to use for tests (chrome, safari, firefox)"
|
||||
)
|
||||
parser.addoption(
|
||||
"--comfyui-url",
|
||||
action="store",
|
||||
default="http://localhost:8188",
|
||||
help="Base URL for ComfyUI server"
|
||||
)
|
||||
@@ -1,176 +0,0 @@
|
||||
"""
|
||||
Browser tests for basic VTK viewer (viewer_vtk.html).
|
||||
|
||||
Tests file format detection, VTP support, and basic mesh loading.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import time
|
||||
|
||||
|
||||
@pytest.mark.browser
|
||||
class TestVTKViewer:
|
||||
"""Test suite for viewer_vtk.html - basic VTK viewer."""
|
||||
|
||||
def test_viewer_loads(self, viewer_page):
|
||||
"""Test that the basic VTK viewer HTML page loads successfully."""
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Check that the page loaded
|
||||
assert "GeomPack" in viewer_page.driver.title or \
|
||||
viewer_page.check_element_exists("#container"), \
|
||||
"VTK viewer page did not load correctly"
|
||||
|
||||
# Check for loading indicator
|
||||
assert viewer_page.check_element_exists("#loading"), \
|
||||
"Loading indicator not found"
|
||||
|
||||
print("✅ Viewer loaded successfully")
|
||||
|
||||
def test_vtp_format_supported(self, viewer_page):
|
||||
"""Test that VTP file format is now supported (the bug we fixed)."""
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Send a VTP file load message
|
||||
viewer_page.send_load_mesh_message("test_mesh.vtp")
|
||||
|
||||
# Wait a bit for processing
|
||||
time.sleep(2)
|
||||
|
||||
# Get console logs
|
||||
logs = viewer_page.get_console_logs()
|
||||
log_text = "\n".join(logs)
|
||||
|
||||
# Check that VTP is detected
|
||||
assert "VTP" in log_text or ".vtp" in log_text, \
|
||||
"VTP file format not detected in logs"
|
||||
|
||||
# Check that there's NO "Unsupported file format" error
|
||||
assert "Unsupported file format" not in log_text, \
|
||||
"VTP file format still showing as unsupported!"
|
||||
|
||||
# Check that VTP reader is mentioned (if loaded successfully)
|
||||
if "VTP (XML PolyData)" in log_text or "VTP" in log_text:
|
||||
print("✅ VTP format is recognized")
|
||||
else:
|
||||
print("⚠️ VTP format detection unclear from logs")
|
||||
|
||||
print(f"\n📝 Console logs:\n{log_text}")
|
||||
|
||||
def test_stl_format_still_works(self, viewer_page):
|
||||
"""Test that STL files still work after our VTP changes."""
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Send an STL file load message
|
||||
viewer_page.send_load_mesh_message("test_mesh.stl")
|
||||
time.sleep(2)
|
||||
|
||||
# Get console logs
|
||||
logs = viewer_page.get_console_logs()
|
||||
log_text = "\n".join(logs)
|
||||
|
||||
# Check that STL is detected
|
||||
assert "STL" in log_text or ".stl" in log_text, \
|
||||
"STL file format not detected"
|
||||
|
||||
# Check that there's no unsupported format error
|
||||
assert "Unsupported file format" not in log_text, \
|
||||
"STL files broken after VTP changes"
|
||||
|
||||
print("✅ STL format still works")
|
||||
|
||||
def test_obj_format_still_works(self, viewer_page):
|
||||
"""Test that OBJ files still work after our VTP changes."""
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Send an OBJ file load message
|
||||
viewer_page.send_load_mesh_message("test_mesh.obj")
|
||||
time.sleep(2)
|
||||
|
||||
# Get console logs
|
||||
logs = viewer_page.get_console_logs()
|
||||
log_text = "\n".join(logs)
|
||||
|
||||
# Check that OBJ is detected
|
||||
assert "OBJ" in log_text or ".obj" in log_text, \
|
||||
"OBJ file format not detected"
|
||||
|
||||
# Check that there's no unsupported format error
|
||||
assert "Unsupported file format" not in log_text, \
|
||||
"OBJ files broken after VTP changes"
|
||||
|
||||
print("✅ OBJ format still works")
|
||||
|
||||
def test_unsupported_format_error(self, viewer_page):
|
||||
"""Test that truly unsupported formats still show error."""
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Send an unsupported file format
|
||||
viewer_page.send_load_mesh_message("test_mesh.fbx")
|
||||
time.sleep(2)
|
||||
|
||||
# Get console logs
|
||||
logs = viewer_page.get_console_logs()
|
||||
log_text = "\n".join(logs)
|
||||
|
||||
# Should show unsupported format error for FBX
|
||||
assert "Unsupported file format" in log_text or "supports STL, OBJ, and VTP" in log_text, \
|
||||
"Unsupported format (FBX) did not trigger error message"
|
||||
|
||||
print("✅ Unsupported formats correctly rejected")
|
||||
|
||||
def test_file_detection_in_query_string(self, viewer_page):
|
||||
"""
|
||||
Test that file format detection works with query string URLs.
|
||||
|
||||
This was the original bug - the URL has query parameters like:
|
||||
/view?filename=mesh.vtp&type=output&subfolder=
|
||||
"""
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Send a realistic ComfyUI-style URL with query params
|
||||
script = """
|
||||
window.postMessage({
|
||||
type: 'LOAD_MESH',
|
||||
filepath: '/view?filename=preview_vtk_fields_abc123.vtp&type=output&subfolder=',
|
||||
timestamp: Date.now()
|
||||
}, '*');
|
||||
"""
|
||||
viewer_page.driver.execute_script(script)
|
||||
time.sleep(2)
|
||||
|
||||
# Get console logs
|
||||
logs = viewer_page.get_console_logs()
|
||||
log_text = "\n".join(logs)
|
||||
|
||||
# The .vtp extension should be detected even with query params
|
||||
assert ".vtp" in log_text, \
|
||||
"VTP extension not detected in query string URL"
|
||||
|
||||
# Should not show unsupported format error
|
||||
assert "Unsupported file format" not in log_text, \
|
||||
"VTP file in query string URL not recognized"
|
||||
|
||||
print("✅ File format detection works with query string URLs")
|
||||
|
||||
def test_controls_present(self, viewer_page):
|
||||
"""Test that viewer controls are present."""
|
||||
viewer_page.load_viewer("viewer_vtk.html")
|
||||
|
||||
# Check for control elements
|
||||
assert viewer_page.check_element_exists("#controls"), \
|
||||
"Controls panel not found"
|
||||
|
||||
assert viewer_page.check_element_exists("#showEdges"), \
|
||||
"Show edges checkbox not found"
|
||||
|
||||
assert viewer_page.check_element_exists("#representation"), \
|
||||
"Representation selector not found"
|
||||
|
||||
assert viewer_page.check_element_exists("#resetCamera"), \
|
||||
"Reset camera button not found"
|
||||
|
||||
assert viewer_page.check_element_exists("#screenshot"), \
|
||||
"Screenshot button not found"
|
||||
|
||||
print("✅ All viewer controls present")
|
||||
@@ -1,18 +0,0 @@
|
||||
# Include base requirements
|
||||
-r ../requirements.txt
|
||||
|
||||
# Test frameworks
|
||||
pytest>=8.0.0
|
||||
pytest-cov>=4.1.0
|
||||
pytest-mock>=3.12.0
|
||||
|
||||
# Browser automation (for @pytest.mark.browser tests)
|
||||
selenium>=4.0.0
|
||||
|
||||
# Workflow execution (for @pytest.mark.workflow tests)
|
||||
websocket-client>=1.6.0
|
||||
|
||||
# Note: CGAL and Blender need special installation for full test coverage
|
||||
# CGAL: pip install cgal (may need system dependencies)
|
||||
# Blender: Automatically downloaded by install.py
|
||||
# Browser tests also require Chrome/Firefox/Safari to be installed
|
||||
@@ -1,12 +0,0 @@
|
||||
# Test outputs (generated during test runs)
|
||||
outputs/
|
||||
|
||||
# Pytest cache
|
||||
__pycache__/
|
||||
.pytest_cache/
|
||||
*.pyc
|
||||
|
||||
# Coverage reports
|
||||
.coverage
|
||||
htmlcov/
|
||||
coverage.xml
|
||||
@@ -1 +0,0 @@
|
||||
"""Test scripts for ComfyUI-GeometryPack."""
|
||||
@@ -1,192 +0,0 @@
|
||||
"""Pytest configuration and fixtures for ComfyUI-GeometryPack tests."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
import numpy as np
|
||||
import trimesh
|
||||
|
||||
# Add parent directory to path so we can import nodes
|
||||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
# Set up paths
|
||||
TEST_DIR = Path(__file__).parent.parent # Go up to tests/ directory
|
||||
ASSETS_DIR = TEST_DIR / "assets"
|
||||
OUTPUT_DIR = TEST_DIR / "outputs"
|
||||
MESHES_DIR = OUTPUT_DIR / "meshes"
|
||||
RENDERS_DIR = OUTPUT_DIR / "renders"
|
||||
|
||||
# Mock ComfyUI modules before importing nodes
|
||||
sys.modules["folder_paths"] = MagicMock()
|
||||
sys.modules["folder_paths"].get_input_directory = MagicMock(
|
||||
return_value=str(ASSETS_DIR)
|
||||
)
|
||||
sys.modules["folder_paths"].get_output_directory = MagicMock(
|
||||
return_value=str(MESHES_DIR)
|
||||
)
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
"""Register custom markers."""
|
||||
config.addinivalue_line("markers", "unit: Unit tests (fast, no optional dependencies)")
|
||||
config.addinivalue_line("markers", "optional: Tests requiring optional dependencies")
|
||||
config.addinivalue_line("markers", "blender: Tests requiring Blender")
|
||||
config.addinivalue_line("markers", "slow: Slow tests (heavy processing)")
|
||||
config.addinivalue_line("markers", "browser: Browser automation tests (require Selenium + browser + ComfyUI running)")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def setup_output_dirs():
|
||||
"""Create output directories for test artifacts."""
|
||||
MESHES_DIR.mkdir(parents=True, exist_ok=True)
|
||||
RENDERS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_assets_dir():
|
||||
"""Return path to test assets directory."""
|
||||
return ASSETS_DIR
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_output_dir(request):
|
||||
"""Return path to test-specific output directory."""
|
||||
# Create a directory for this specific test
|
||||
test_name = request.node.name
|
||||
test_dir = OUTPUT_DIR / test_name
|
||||
test_dir.mkdir(parents=True, exist_ok=True)
|
||||
return test_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def meshes_output_dir(test_output_dir):
|
||||
"""Return path to test-specific meshes directory.
|
||||
|
||||
For I/O tests, this returns the mocked ComfyUI output directory
|
||||
where SaveMesh actually saves files.
|
||||
"""
|
||||
# Return the mocked output directory that SaveMesh uses
|
||||
# This is where files are actually saved via folder_paths.get_output_directory()
|
||||
return Path(sys.modules["folder_paths"].get_output_directory())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def renders_output_dir(test_output_dir):
|
||||
"""Return path to test-specific renders directory."""
|
||||
render_dir = test_output_dir / "renders"
|
||||
render_dir.mkdir(exist_ok=True)
|
||||
return render_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bunny_mesh():
|
||||
"""Load Stanford Bunny mesh."""
|
||||
bunny_path = ASSETS_DIR / "Stanford_Bunny.stl"
|
||||
if bunny_path.exists():
|
||||
return trimesh.load(str(bunny_path))
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cube_mesh():
|
||||
"""Create a simple cube mesh."""
|
||||
return trimesh.creation.box(extents=[1.0, 1.0, 1.0])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sphere_mesh():
|
||||
"""Create a simple sphere mesh."""
|
||||
return trimesh.creation.icosphere(subdivisions=2, radius=1.0)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def open_mesh():
|
||||
"""Create an open mesh (cylinder without caps) for UV unwrapping tests."""
|
||||
# Create a cylinder and remove the caps to make it open
|
||||
cylinder = trimesh.creation.cylinder(radius=1.0, height=2.0, sections=16)
|
||||
|
||||
# Identify and remove cap faces (faces at top/bottom with normals pointing up/down)
|
||||
face_normals = cylinder.face_normals
|
||||
# Cap faces have normals nearly vertical (pointing up or down)
|
||||
is_cap = np.abs(face_normals[:, 2]) > 0.9 # Z component dominates
|
||||
|
||||
# Keep only side faces
|
||||
side_faces = cylinder.faces[~is_cap]
|
||||
|
||||
# Create open mesh with only side faces and clean it up
|
||||
open_cylinder = trimesh.Trimesh(vertices=cylinder.vertices, faces=side_faces, process=True)
|
||||
|
||||
# Remove unreferenced vertices
|
||||
open_cylinder.remove_unreferenced_vertices()
|
||||
|
||||
return open_cylinder
|
||||
|
||||
|
||||
def render_mesh_to_png(mesh, output_path, resolution=(800, 800)):
|
||||
"""Render mesh to PNG file for visual verification."""
|
||||
try:
|
||||
import matplotlib.pyplot as plt
|
||||
from mpl_toolkits.mplot3d import Axes3D
|
||||
from mpl_toolkits.mplot3d.art3d import Poly3DCollection
|
||||
|
||||
fig = plt.figure(figsize=(8, 8))
|
||||
ax = fig.add_subplot(111, projection='3d')
|
||||
|
||||
# Create mesh collection
|
||||
collection = Poly3DCollection(
|
||||
mesh.vertices[mesh.faces],
|
||||
alpha=0.7,
|
||||
facecolor='cyan',
|
||||
edgecolor='black',
|
||||
linewidths=0.1
|
||||
)
|
||||
ax.add_collection3d(collection)
|
||||
|
||||
# Set axis limits
|
||||
scale = mesh.vertices.flatten()
|
||||
ax.auto_scale_xyz(scale, scale, scale)
|
||||
|
||||
# Labels
|
||||
ax.set_xlabel('X')
|
||||
ax.set_ylabel('Y')
|
||||
ax.set_zlabel('Z')
|
||||
|
||||
plt.savefig(output_path, dpi=100, bbox_inches='tight')
|
||||
plt.close()
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Warning: Could not render mesh: {e}")
|
||||
return False
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def render_helper(renders_output_dir):
|
||||
"""Helper fixture for rendering meshes."""
|
||||
def _render(mesh, name, **kwargs):
|
||||
output_path = renders_output_dir / f"{name}.png"
|
||||
render_mesh_to_png(mesh, output_path, **kwargs)
|
||||
return output_path
|
||||
return _render
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def save_mesh_helper(meshes_output_dir):
|
||||
"""Helper fixture for saving meshes."""
|
||||
def _save(mesh, name, format="obj"):
|
||||
output_path = meshes_output_dir / f"{name}.{format}"
|
||||
mesh.export(str(output_path))
|
||||
return output_path
|
||||
return _save
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def save_and_render(save_mesh_helper, render_helper):
|
||||
"""Combined helper to save mesh and render it."""
|
||||
def _save_and_render(mesh, name, format="obj"):
|
||||
save_mesh_helper(mesh, name, format)
|
||||
render_helper(mesh, name)
|
||||
return True
|
||||
return _save_and_render
|
||||
@@ -1,60 +0,0 @@
|
||||
"""Tests for analysis nodes (MeshInfoNode, MarkBoundaryEdgesNode)."""
|
||||
|
||||
import pytest
|
||||
from nodes.analysis import MeshInfoNode, MarkBoundaryEdgesNode
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mesh_info_cube(cube_mesh):
|
||||
"""Test mesh info on cube."""
|
||||
node = MeshInfoNode()
|
||||
info = node.get_mesh_info(trimesh=cube_mesh)[0]
|
||||
|
||||
assert "vertices" in info.lower()
|
||||
assert "faces" in info.lower()
|
||||
assert "8" in info # cube has 8 vertices
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mesh_info_bunny(bunny_mesh):
|
||||
"""Test mesh info on Stanford Bunny."""
|
||||
if bunny_mesh is None:
|
||||
pytest.skip("Bunny mesh not available")
|
||||
|
||||
node = MeshInfoNode()
|
||||
info = node.get_mesh_info(trimesh=bunny_mesh)[0]
|
||||
|
||||
assert "vertices" in info.lower()
|
||||
assert "faces" in info.lower()
|
||||
assert "volume" in info.lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mark_boundary_edges_closed_mesh(cube_mesh, save_mesh_helper):
|
||||
"""Test boundary detection on closed mesh (should have no boundaries)."""
|
||||
node = MarkBoundaryEdgesNode()
|
||||
mesh_with_field, info = node.mark_boundary(trimesh=cube_mesh)
|
||||
|
||||
assert mesh_with_field is not None
|
||||
assert "boundary" in info.lower()
|
||||
|
||||
save_mesh_helper(mesh_with_field, "cube_with_boundary_field", "ply")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mark_boundary_edges_open_mesh(open_mesh, save_mesh_helper, render_helper):
|
||||
"""Test boundary detection on open mesh (should have boundaries)."""
|
||||
# Skip this test - open_mesh fixture creates invalid mesh
|
||||
pytest.skip("Open mesh fixture needs fixing")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mesh_info_volume_area(sphere_mesh):
|
||||
"""Test mesh info calculates volume and area."""
|
||||
node = MeshInfoNode()
|
||||
info = node.get_mesh_info(trimesh=sphere_mesh)[0]
|
||||
|
||||
assert "volume" in info.lower()
|
||||
assert "area" in info.lower()
|
||||
# Sphere with radius 0.5 should have volume ~0.5
|
||||
assert "0." in info
|
||||
@@ -1,70 +0,0 @@
|
||||
"""Tests for conversion nodes (StripMeshAdjacencyNode, MeshToPointCloudNode)."""
|
||||
|
||||
import pytest
|
||||
from nodes.conversion import StripMeshAdjacencyNode, MeshToPointCloudNode
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_strip_mesh_adjacency(cube_mesh):
|
||||
"""Test converting mesh to point cloud without sampling."""
|
||||
node = StripMeshAdjacencyNode()
|
||||
pc = node.strip_adjacency(trimesh=cube_mesh)[0]
|
||||
|
||||
assert pc is not None
|
||||
assert "points" in pc
|
||||
assert pc["points"].shape[0] == cube_mesh.vertices.shape[0]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_strip_mesh_with_normals(cube_mesh):
|
||||
"""Test converting mesh to point cloud with normals."""
|
||||
node = StripMeshAdjacencyNode()
|
||||
pc = node.strip_adjacency(trimesh=cube_mesh, include_normals="true")[0]
|
||||
|
||||
assert "normals" in pc
|
||||
assert pc["normals"].shape[0] == pc["points"].shape[0]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("sample_count", [100, 500, 1000])
|
||||
def test_mesh_to_point_cloud_sampling(sphere_mesh, sample_count):
|
||||
"""Test sampling point cloud from mesh."""
|
||||
node = MeshToPointCloudNode()
|
||||
pc = node.mesh_to_pointcloud(
|
||||
trimesh=sphere_mesh,
|
||||
sample_count=sample_count,
|
||||
sampling_method="uniform"
|
||||
)[0]
|
||||
|
||||
assert pc["points"].shape[0] == sample_count
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("method", ["uniform", "face_weighted"])
|
||||
def test_mesh_to_point_cloud_methods(sphere_mesh, method):
|
||||
"""Test different sampling methods."""
|
||||
node = MeshToPointCloudNode()
|
||||
pc = node.mesh_to_pointcloud(
|
||||
trimesh=sphere_mesh,
|
||||
sample_count=500,
|
||||
sampling_method=method
|
||||
)[0]
|
||||
|
||||
assert pc is not None
|
||||
# "face_weighted" might return different count
|
||||
assert pc["points"].shape[0] > 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_point_cloud_with_normals(sphere_mesh):
|
||||
"""Test point cloud generation with normals."""
|
||||
node = MeshToPointCloudNode()
|
||||
pc = node.mesh_to_pointcloud(
|
||||
trimesh=sphere_mesh,
|
||||
sample_count=300,
|
||||
sampling_method="uniform",
|
||||
include_normals="true"
|
||||
)[0]
|
||||
|
||||
assert "normals" in pc
|
||||
assert pc["normals"].shape == (300, 3)
|
||||
@@ -1,88 +0,0 @@
|
||||
"""Tests for distance metric nodes."""
|
||||
|
||||
import pytest
|
||||
from nodes.distance import HausdorffDistanceNode, ChamferDistanceNode, ComputeSDFNode
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_hausdorff_distance_identical(sphere_mesh):
|
||||
"""Test Hausdorff distance between identical meshes."""
|
||||
pytest.importorskip("point_cloud_utils")
|
||||
|
||||
node = HausdorffDistanceNode()
|
||||
distance, details = node.compute_distance(
|
||||
trimesh_a=sphere_mesh,
|
||||
trimesh_b=sphere_mesh,
|
||||
sample_count=1000
|
||||
)
|
||||
|
||||
# Relaxed tolerance: sampling introduces variation even for identical meshes
|
||||
assert distance < 0.2 # Should be near zero for identical meshes
|
||||
assert "hausdorff" in details.lower()
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_hausdorff_distance_different(cube_mesh, sphere_mesh):
|
||||
"""Test Hausdorff distance between different meshes."""
|
||||
pytest.importorskip("point_cloud_utils")
|
||||
|
||||
node = HausdorffDistanceNode()
|
||||
distance, details = node.compute_distance(
|
||||
trimesh_a=cube_mesh,
|
||||
trimesh_b=sphere_mesh,
|
||||
sample_count=1000
|
||||
)
|
||||
|
||||
assert distance > 0
|
||||
assert "hausdorff" in details.lower()
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_chamfer_distance_identical(sphere_mesh):
|
||||
"""Test Chamfer distance between identical meshes."""
|
||||
pytest.importorskip("point_cloud_utils")
|
||||
|
||||
node = ChamferDistanceNode()
|
||||
distance, info = node.compute_distance(
|
||||
trimesh_a=sphere_mesh,
|
||||
trimesh_b=sphere_mesh,
|
||||
sample_count=1000
|
||||
)
|
||||
|
||||
# Relaxed tolerance: sampling introduces variation even for identical meshes
|
||||
assert distance < 0.15
|
||||
assert "chamfer" in info.lower()
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_chamfer_distance_different(cube_mesh, sphere_mesh):
|
||||
"""Test Chamfer distance between different meshes."""
|
||||
pytest.importorskip("point_cloud_utils")
|
||||
|
||||
node = ChamferDistanceNode()
|
||||
distance, info = node.compute_distance(
|
||||
trimesh_a=cube_mesh,
|
||||
trimesh_b=sphere_mesh,
|
||||
sample_count=1000
|
||||
)
|
||||
|
||||
assert distance > 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("resolution", [32, 64])
|
||||
def test_compute_sdf(sphere_mesh, resolution):
|
||||
"""Test SDF computation using libigl."""
|
||||
pytest.importorskip("igl")
|
||||
|
||||
node = ComputeSDFNode()
|
||||
sdf_volume, info = node.compute_sdf(
|
||||
trimesh=sphere_mesh,
|
||||
resolution=resolution
|
||||
)
|
||||
|
||||
assert sdf_volume is not None
|
||||
assert "voxels" in sdf_volume
|
||||
assert "resolution" in sdf_volume
|
||||
assert sdf_volume["resolution"] == resolution
|
||||
assert "distance" in info.lower() # Check for "distance" instead of "sdf"
|
||||
@@ -1,154 +0,0 @@
|
||||
"""Integration tests for node pipelines."""
|
||||
|
||||
import pytest
|
||||
from nodes.io import LoadMesh, SaveMesh
|
||||
from nodes.primitives import CreatePrimitive
|
||||
from nodes.analysis import MeshInfoNode, MarkBoundaryEdgesNode
|
||||
from nodes.transforms import CenterMeshNode
|
||||
from nodes.remeshing import MeshDecimationNode, MeshSubdivisionNode
|
||||
from nodes.repair import FixNormalsNode, ComputeNormalsNode
|
||||
from nodes.conversion import MeshToPointCloudNode
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pipeline_create_analyze_save(save_mesh_helper, render_helper):
|
||||
"""Test: Create → Analyze → Save pipeline."""
|
||||
# Create primitive
|
||||
create_node = CreatePrimitive()
|
||||
mesh = create_node.create_primitive(shape="sphere", size=1.0)[0]
|
||||
|
||||
# Analyze
|
||||
info_node = MeshInfoNode()
|
||||
info = info_node.get_mesh_info(trimesh=mesh)[0]
|
||||
assert "vertices" in info.lower()
|
||||
|
||||
# Save
|
||||
save_mesh_helper(mesh, "pipeline_create_analyze", "obj")
|
||||
render_helper(mesh, "pipeline_create_analyze")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pipeline_load_remesh_save(bunny_mesh, save_mesh_helper, render_helper):
|
||||
"""Test: Load → Remesh → Save pipeline."""
|
||||
if bunny_mesh is None:
|
||||
pytest.skip("Bunny mesh not available")
|
||||
|
||||
# Decimate
|
||||
decimate_node = MeshDecimationNode()
|
||||
decimated, info = decimate_node.decimate(
|
||||
trimesh=bunny_mesh,
|
||||
target_face_count=500
|
||||
)
|
||||
|
||||
# Center
|
||||
center_node = CenterMeshNode()
|
||||
centered = center_node.center_mesh(trimesh=decimated)[0]
|
||||
|
||||
# Save
|
||||
save_mesh_helper(centered, "pipeline_load_remesh_save", "obj")
|
||||
render_helper(centered, "pipeline_load_remesh_save")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pipeline_create_subdivide_smooth(cube_mesh, save_mesh_helper, render_helper):
|
||||
"""Test: Create → Subdivide → Smooth pipeline."""
|
||||
from nodes.remeshing import LaplacianSmoothingNode
|
||||
|
||||
# Subdivide
|
||||
subdivide_node = MeshSubdivisionNode()
|
||||
subdivided, _ = subdivide_node.subdivide(
|
||||
trimesh=cube_mesh,
|
||||
iterations=2,
|
||||
method="loop"
|
||||
)
|
||||
|
||||
# Smooth
|
||||
smooth_node = LaplacianSmoothingNode()
|
||||
smoothed = smooth_node.smooth(
|
||||
trimesh=subdivided,
|
||||
iterations=10,
|
||||
lambda_factor=0.5
|
||||
)[0]
|
||||
|
||||
save_mesh_helper(smoothed, "pipeline_subdivide_smooth", "obj")
|
||||
render_helper(smoothed, "pipeline_subdivide_smooth")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pipeline_repair_analyze(sphere_mesh, save_mesh_helper):
|
||||
"""Test: Repair → Analyze pipeline."""
|
||||
# Fix normals
|
||||
fix_node = FixNormalsNode()
|
||||
fixed, _ = fix_node.fix_normals(trimesh=sphere_mesh)
|
||||
|
||||
# Analyze
|
||||
info_node = MeshInfoNode()
|
||||
info = info_node.get_mesh_info(trimesh=fixed)[0]
|
||||
assert "vertices" in info.lower()
|
||||
|
||||
save_mesh_helper(fixed, "pipeline_repair_analyze", "obj")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pipeline_mesh_to_pointcloud(sphere_mesh):
|
||||
"""Test: Create → Convert to Point Cloud pipeline."""
|
||||
# Convert to point cloud
|
||||
pc_node = MeshToPointCloudNode()
|
||||
pc = pc_node.mesh_to_pointcloud(
|
||||
trimesh=sphere_mesh,
|
||||
sample_count=1000,
|
||||
sampling_method="uniform",
|
||||
include_normals="true"
|
||||
)[0]
|
||||
|
||||
assert pc["points"].shape[0] == 1000
|
||||
assert "normals" in pc
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pipeline_boundary_detection_visualization(cube_mesh, save_mesh_helper):
|
||||
"""Test: Boundary Detection → Visualization pipeline."""
|
||||
from nodes.analysis import MarkBoundaryEdgesNode
|
||||
# Mark boundaries
|
||||
boundary_node = MarkBoundaryEdgesNode()
|
||||
mesh_with_field, info = boundary_node.mark_boundary(trimesh=cube_mesh)
|
||||
|
||||
# Compute normals
|
||||
normals_node = ComputeNormalsNode()
|
||||
with_normals = normals_node.compute_normals(
|
||||
trimesh=mesh_with_field,
|
||||
smooth_vertex_normals="true"
|
||||
)[0]
|
||||
|
||||
save_mesh_helper(with_normals, "pipeline_boundary_viz", "ply")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pipeline_full_processing(cube_mesh, save_mesh_helper, render_helper):
|
||||
"""Test: Full processing pipeline."""
|
||||
# Center
|
||||
center_node = CenterMeshNode()
|
||||
centered = center_node.center_mesh(trimesh=cube_mesh)[0]
|
||||
|
||||
# Subdivide
|
||||
subdivide_node = MeshSubdivisionNode()
|
||||
subdivided, _ = subdivide_node.subdivide(
|
||||
trimesh=centered,
|
||||
iterations=2,
|
||||
method="loop"
|
||||
)
|
||||
|
||||
# Compute normals
|
||||
normals_node = ComputeNormalsNode()
|
||||
final = normals_node.compute_normals(
|
||||
trimesh=subdivided,
|
||||
smooth_vertex_normals="true"
|
||||
)[0]
|
||||
|
||||
# Analyze
|
||||
info_node = MeshInfoNode()
|
||||
info = info_node.get_mesh_info(trimesh=final)[0]
|
||||
assert "vertices" in info.lower()
|
||||
|
||||
save_mesh_helper(final, "pipeline_full_processing", "obj")
|
||||
render_helper(final, "pipeline_full_processing")
|
||||
@@ -1,66 +0,0 @@
|
||||
"""Tests for I/O nodes (LoadMesh, SaveMesh)."""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from nodes.io import LoadMesh, SaveMesh
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_load_mesh_stl(bunny_mesh, render_helper, save_mesh_helper):
|
||||
"""Test loading STL file."""
|
||||
if bunny_mesh is None:
|
||||
pytest.skip("Stanford_Bunny.stl not found in assets")
|
||||
|
||||
node = LoadMesh()
|
||||
mesh = node.load_mesh(file_path="Stanford_Bunny.stl")[0]
|
||||
|
||||
assert mesh is not None
|
||||
assert mesh.vertices.shape[0] > 0
|
||||
assert mesh.faces.shape[0] > 0
|
||||
|
||||
save_mesh_helper(mesh, "loaded_bunny", "obj")
|
||||
render_helper(mesh, "loaded_bunny")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_save_mesh_obj(cube_mesh, meshes_output_dir):
|
||||
"""Test saving mesh to OBJ format."""
|
||||
node = SaveMesh()
|
||||
status = node.save_mesh(trimesh=cube_mesh, file_path="test_cube.obj")[0]
|
||||
|
||||
assert "successfully" in status.lower()
|
||||
assert (meshes_output_dir / "test_cube.obj").exists()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_save_mesh_ply(sphere_mesh, meshes_output_dir):
|
||||
"""Test saving mesh to PLY format."""
|
||||
node = SaveMesh()
|
||||
status = node.save_mesh(trimesh=sphere_mesh, file_path="test_sphere.ply")[0]
|
||||
|
||||
assert "successfully" in status.lower()
|
||||
assert (meshes_output_dir / "test_sphere.ply").exists()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("format", ["obj", "ply", "stl", "off"])
|
||||
def test_save_load_cycle(cube_mesh, meshes_output_dir, format):
|
||||
"""Test save/load cycle for different formats."""
|
||||
save_node = SaveMesh()
|
||||
load_node = LoadMesh()
|
||||
|
||||
filename = f"cycle_test.{format}"
|
||||
# Save to output directory
|
||||
save_node.save_mesh(trimesh=cube_mesh, file_path=filename)
|
||||
|
||||
# Load from output directory (files are saved there)
|
||||
from pathlib import Path
|
||||
saved_path = meshes_output_dir / filename
|
||||
if not saved_path.exists():
|
||||
pytest.skip(f"File {filename} was not saved correctly")
|
||||
|
||||
loaded_mesh = load_node.load_mesh(file_path=str(saved_path))[0]
|
||||
|
||||
assert loaded_mesh is not None
|
||||
assert loaded_mesh.vertices.shape[0] > 0
|
||||
assert loaded_mesh.faces.shape[0] > 0
|
||||
@@ -1,64 +0,0 @@
|
||||
"""Tests for primitive generation (CreatePrimitive)."""
|
||||
|
||||
import pytest
|
||||
from nodes.primitives import CreatePrimitive
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("shape,expected_verts", [
|
||||
("cube", 8),
|
||||
("plane", 4),
|
||||
])
|
||||
def test_create_primitive_shapes(shape, expected_verts, render_helper, save_mesh_helper):
|
||||
"""Test creating different primitive shapes."""
|
||||
node = CreatePrimitive()
|
||||
mesh = node.create_primitive(shape=shape, size=1.0)[0]
|
||||
|
||||
assert mesh is not None
|
||||
assert mesh.vertices.shape[0] >= expected_verts
|
||||
assert mesh.faces.shape[0] > 0
|
||||
assert mesh.is_watertight or shape == "plane"
|
||||
|
||||
save_mesh_helper(mesh, f"primitive_{shape}", "obj")
|
||||
render_helper(mesh, f"primitive_{shape}")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_sphere(render_helper, save_mesh_helper):
|
||||
"""Test creating sphere primitive."""
|
||||
node = CreatePrimitive()
|
||||
mesh = node.create_primitive(shape="sphere", size=1.0)[0]
|
||||
|
||||
assert mesh is not None
|
||||
assert mesh.vertices.shape[0] > 10
|
||||
assert mesh.is_watertight
|
||||
# Size parameter is diameter, so radius = size/2
|
||||
assert abs(mesh.bounding_sphere.primitive.radius - 0.5) < 0.1
|
||||
|
||||
save_mesh_helper(mesh, "primitive_sphere", "obj")
|
||||
render_helper(mesh, "primitive_sphere")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_primitive_with_subdivisions(render_helper, save_mesh_helper):
|
||||
"""Test creating primitive with subdivisions."""
|
||||
node = CreatePrimitive()
|
||||
mesh_sub0 = node.create_primitive(shape="sphere", size=1.0, subdivisions=0)[0]
|
||||
mesh_sub2 = node.create_primitive(shape="sphere", size=1.0, subdivisions=2)[0]
|
||||
|
||||
assert mesh_sub2.vertices.shape[0] > mesh_sub0.vertices.shape[0]
|
||||
|
||||
save_mesh_helper(mesh_sub2, "primitive_sphere_subdivided", "obj")
|
||||
render_helper(mesh_sub2, "primitive_sphere_subdivided")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("size", [0.5, 1.0, 2.0])
|
||||
def test_create_primitive_sizes(size):
|
||||
"""Test creating primitives with different sizes."""
|
||||
node = CreatePrimitive()
|
||||
mesh = node.create_primitive(shape="cube", size=size)[0]
|
||||
|
||||
bounds = mesh.bounds
|
||||
extent = bounds[1] - bounds[0]
|
||||
assert all(extent >= size * 0.9)
|
||||
@@ -1,206 +0,0 @@
|
||||
"""Tests for remeshing nodes."""
|
||||
|
||||
import pytest
|
||||
from nodes.remeshing import (
|
||||
PyMeshLabRemeshNode,
|
||||
CGALIsotropicRemeshNode,
|
||||
MeshDecimationNode,
|
||||
MeshSubdivisionNode,
|
||||
LaplacianSmoothingNode,
|
||||
BlenderVoxelRemeshNode,
|
||||
BlenderQuadriflowRemeshNode,
|
||||
InstantMeshesRemeshNode,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.optional
|
||||
def test_pymeshlab_remesh(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test PyMeshLab isotropic remeshing."""
|
||||
pytest.importorskip("pymeshlab")
|
||||
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = PyMeshLabRemeshNode()
|
||||
|
||||
try:
|
||||
remeshed = node.remesh(
|
||||
trimesh=sphere_mesh,
|
||||
target_edge_length=0.1,
|
||||
iterations=10
|
||||
)[0]
|
||||
except ValueError as e:
|
||||
if "PyMeshLab meshing filter not available" in str(e):
|
||||
pytest.skip("PyMeshLab meshing filter not available (missing OpenGL libraries)")
|
||||
raise
|
||||
|
||||
assert remeshed is not None
|
||||
assert remeshed.faces.shape[0] > 0
|
||||
|
||||
# Save remeshed result
|
||||
save_mesh_helper(remeshed, "01_pymeshlab_remeshed", "obj")
|
||||
render_helper(remeshed, "01_pymeshlab_remeshed")
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_cgal_remesh(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test CGAL isotropic remeshing."""
|
||||
pytest.importorskip("cgal")
|
||||
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = CGALIsotropicRemeshNode()
|
||||
remeshed = node.remesh(
|
||||
trimesh=sphere_mesh,
|
||||
target_edge_length=0.1,
|
||||
iterations=5,
|
||||
protect_boundaries="false"
|
||||
)[0]
|
||||
|
||||
assert remeshed is not None
|
||||
assert remeshed.faces.shape[0] > 0
|
||||
|
||||
# Save remeshed result
|
||||
save_mesh_helper(remeshed, "01_cgal_remeshed", "obj")
|
||||
render_helper(remeshed, "01_cgal_remeshed")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mesh_decimation(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test mesh decimation - reduces triangle count while preserving shape."""
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
original_face_count = sphere_mesh.faces.shape[0]
|
||||
target_faces = original_face_count // 4
|
||||
|
||||
node = MeshDecimationNode()
|
||||
decimated, info = node.decimate(
|
||||
trimesh=sphere_mesh,
|
||||
target_face_count=target_faces
|
||||
)
|
||||
|
||||
assert decimated is not None
|
||||
assert decimated.faces.shape[0] < original_face_count
|
||||
assert "face" in info.lower()
|
||||
|
||||
# Save decimated result
|
||||
save_mesh_helper(decimated, "01_decimated", "obj")
|
||||
render_helper(decimated, "01_decimated")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("iterations", [1, 2, 3])
|
||||
def test_mesh_subdivision(cube_mesh, iterations, save_mesh_helper, render_helper):
|
||||
"""Test mesh subdivision - increases mesh resolution."""
|
||||
# Save original (only on first iteration to avoid duplication across params)
|
||||
save_mesh_helper(cube_mesh, "00_original", "obj")
|
||||
render_helper(cube_mesh, "00_original")
|
||||
|
||||
original_face_count = cube_mesh.faces.shape[0]
|
||||
|
||||
node = MeshSubdivisionNode()
|
||||
subdivided, info = node.subdivide(
|
||||
trimesh=cube_mesh,
|
||||
iterations=iterations,
|
||||
method="loop"
|
||||
)
|
||||
|
||||
assert subdivided is not None
|
||||
assert subdivided.faces.shape[0] > original_face_count
|
||||
|
||||
# Save with iteration count in name
|
||||
save_mesh_helper(subdivided, f"01_subdivided_x{iterations}", "obj")
|
||||
render_helper(subdivided, f"01_subdivided_x{iterations}")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_laplacian_smoothing(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test Laplacian smoothing - smooths mesh surface."""
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = LaplacianSmoothingNode()
|
||||
smoothed = node.smooth(
|
||||
trimesh=sphere_mesh,
|
||||
iterations=10,
|
||||
lambda_factor=0.5
|
||||
)[0]
|
||||
|
||||
assert smoothed is not None
|
||||
assert smoothed.vertices.shape == sphere_mesh.vertices.shape
|
||||
|
||||
# Save smoothed result
|
||||
save_mesh_helper(smoothed, "01_smoothed", "obj")
|
||||
render_helper(smoothed, "01_smoothed")
|
||||
|
||||
|
||||
@pytest.mark.blender
|
||||
@pytest.mark.slow
|
||||
def test_blender_voxel_remesh(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test Blender voxel remeshing."""
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = BlenderVoxelRemeshNode()
|
||||
remeshed = node.voxel_remesh(trimesh=sphere_mesh, voxel_size=0.1)[0]
|
||||
|
||||
assert remeshed is not None
|
||||
|
||||
# Save remeshed result
|
||||
save_mesh_helper(remeshed, "01_voxel_remeshed", "obj")
|
||||
render_helper(remeshed, "01_voxel_remeshed")
|
||||
|
||||
|
||||
@pytest.mark.blender
|
||||
@pytest.mark.slow
|
||||
def test_blender_quadriflow_remesh(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test Blender Quadriflow remeshing."""
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = BlenderQuadriflowRemeshNode()
|
||||
remeshed = node.quadriflow_remesh(
|
||||
trimesh=sphere_mesh,
|
||||
target_face_count=500
|
||||
)[0]
|
||||
|
||||
assert remeshed is not None
|
||||
|
||||
# Save remeshed result
|
||||
save_mesh_helper(remeshed, "01_quadriflow_remeshed", "obj")
|
||||
render_helper(remeshed, "01_quadriflow_remeshed")
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
@pytest.mark.slow
|
||||
def test_instant_meshes_remesh(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test InstantMeshes remeshing."""
|
||||
pytest.importorskip("PyNanoInstantMeshes")
|
||||
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = InstantMeshesRemeshNode()
|
||||
remeshed, info = node.instant_remesh(
|
||||
trimesh=sphere_mesh,
|
||||
target_vertex_count=500,
|
||||
deterministic="true",
|
||||
crease_angle=30.0
|
||||
)
|
||||
|
||||
assert remeshed is not None
|
||||
assert "vertices" in info.lower()
|
||||
|
||||
# Save remeshed result
|
||||
save_mesh_helper(remeshed, "01_instant_remeshed", "obj")
|
||||
render_helper(remeshed, "01_instant_remeshed")
|
||||
@@ -1,121 +0,0 @@
|
||||
"""Tests for repair nodes (FixNormalsNode, CheckNormalsNode, FillHolesNode, etc.)."""
|
||||
|
||||
import pytest
|
||||
import numpy as np
|
||||
from nodes.repair import (
|
||||
FixNormalsNode,
|
||||
CheckNormalsNode,
|
||||
FillHolesNode,
|
||||
ComputeNormalsNode,
|
||||
VisualizNormalFieldNode,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_fix_normals(cube_mesh, save_mesh_helper, render_helper):
|
||||
"""Test fixing mesh normals."""
|
||||
# Save original
|
||||
save_mesh_helper(cube_mesh, "00_original", "obj")
|
||||
render_helper(cube_mesh, "00_original")
|
||||
|
||||
node = FixNormalsNode()
|
||||
fixed_mesh, info = node.fix_normals(trimesh=cube_mesh)
|
||||
|
||||
assert fixed_mesh is not None
|
||||
assert "normal" in info.lower()
|
||||
|
||||
# Save fixed result
|
||||
save_mesh_helper(fixed_mesh, "01_fixed_normals", "obj")
|
||||
render_helper(fixed_mesh, "01_fixed_normals")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_check_normals(sphere_mesh):
|
||||
"""Test checking normal consistency."""
|
||||
node = CheckNormalsNode()
|
||||
report = node.check_normals(trimesh=sphere_mesh)[0]
|
||||
|
||||
assert report is not None
|
||||
assert "normal" in report.lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_fill_holes(open_mesh, save_mesh_helper, render_helper):
|
||||
"""Test filling holes in open mesh."""
|
||||
# Save original open mesh
|
||||
save_mesh_helper(open_mesh, "00_original_open", "obj")
|
||||
render_helper(open_mesh, "00_original_open")
|
||||
|
||||
node = FillHolesNode()
|
||||
filled_mesh, info = node.fill_holes(trimesh=open_mesh)
|
||||
|
||||
assert filled_mesh is not None
|
||||
assert "hole" in info.lower()
|
||||
|
||||
# Save filled result
|
||||
save_mesh_helper(filled_mesh, "01_holes_filled", "obj")
|
||||
render_helper(filled_mesh, "01_holes_filled")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_compute_normals_faceted(cube_mesh, save_mesh_helper, render_helper):
|
||||
"""Test computing faceted normals."""
|
||||
# Save original
|
||||
save_mesh_helper(cube_mesh, "00_original", "obj")
|
||||
render_helper(cube_mesh, "00_original")
|
||||
|
||||
node = ComputeNormalsNode()
|
||||
mesh_with_normals = node.compute_normals(
|
||||
trimesh=cube_mesh,
|
||||
smooth_vertex_normals="false"
|
||||
)[0]
|
||||
|
||||
assert mesh_with_normals is not None
|
||||
assert hasattr(mesh_with_normals, 'face_normals')
|
||||
|
||||
# Save with faceted normals
|
||||
save_mesh_helper(mesh_with_normals, "01_faceted_normals", "obj")
|
||||
render_helper(mesh_with_normals, "01_faceted_normals")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_compute_normals_smooth(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test computing smooth vertex normals."""
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = ComputeNormalsNode()
|
||||
mesh_with_normals = node.compute_normals(
|
||||
trimesh=sphere_mesh,
|
||||
smooth_vertex_normals="true"
|
||||
)[0]
|
||||
|
||||
assert mesh_with_normals is not None
|
||||
|
||||
# Save with smooth normals
|
||||
save_mesh_helper(mesh_with_normals, "01_smooth_normals", "obj")
|
||||
render_helper(mesh_with_normals, "01_smooth_normals")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_visualize_normal_field(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test visualizing normal field as scalar fields."""
|
||||
# Save original
|
||||
save_mesh_helper(sphere_mesh, "00_original", "obj")
|
||||
render_helper(sphere_mesh, "00_original")
|
||||
|
||||
node = VisualizNormalFieldNode()
|
||||
mesh_with_fields, info = node.visualize_normals(trimesh=sphere_mesh)
|
||||
|
||||
assert mesh_with_fields is not None
|
||||
assert "normal" in info.lower()
|
||||
|
||||
# Check that scalar fields were added
|
||||
if hasattr(mesh_with_fields, 'vertex_attributes'):
|
||||
attrs = mesh_with_fields.vertex_attributes
|
||||
assert any(k.startswith('normal_') for k in attrs.keys())
|
||||
|
||||
# Save with fields (PLY supports vertex attributes)
|
||||
save_mesh_helper(mesh_with_fields, "01_with_normal_fields", "ply")
|
||||
render_helper(mesh_with_fields, "01_with_normal_fields")
|
||||
@@ -1,34 +0,0 @@
|
||||
"""Tests for transform nodes (CenterMeshNode)."""
|
||||
|
||||
import pytest
|
||||
import numpy as np
|
||||
from nodes.transforms import CenterMeshNode
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_center_mesh(cube_mesh, save_mesh_helper, render_helper):
|
||||
"""Test centering mesh at origin."""
|
||||
# Translate cube away from origin
|
||||
cube_mesh.vertices += [5.0, 3.0, -2.0]
|
||||
|
||||
node = CenterMeshNode()
|
||||
centered = node.center_mesh(trimesh=cube_mesh)[0]
|
||||
|
||||
# Check that centroid is at origin
|
||||
centroid = centered.vertices.mean(axis=0)
|
||||
assert np.allclose(centroid, [0, 0, 0], atol=1e-6)
|
||||
|
||||
save_mesh_helper(centered, "centered_cube", "obj")
|
||||
render_helper(centered, "centered_cube")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_center_mesh_preserves_shape(sphere_mesh):
|
||||
"""Test that centering preserves mesh shape."""
|
||||
original_bounds_extent = sphere_mesh.bounds[1] - sphere_mesh.bounds[0]
|
||||
|
||||
node = CenterMeshNode()
|
||||
centered = node.center_mesh(trimesh=sphere_mesh)[0]
|
||||
|
||||
centered_extent = centered.bounds[1] - centered.bounds[0]
|
||||
assert np.allclose(original_bounds_extent, centered_extent, atol=1e-6)
|
||||
@@ -1,124 +0,0 @@
|
||||
"""Tests for UV unwrapping nodes."""
|
||||
|
||||
import pytest
|
||||
from nodes.uv import (
|
||||
XAtlasUVUnwrapNode,
|
||||
LibiglLSCMNode,
|
||||
LibiglHarmonicNode,
|
||||
LibiglARAPNode,
|
||||
BlenderUVUnwrapNode,
|
||||
BlenderCubeProjectionNode,
|
||||
BlenderCylinderProjectionNode,
|
||||
BlenderSphereProjectionNode,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_xatlas_unwrap(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test xAtlas UV unwrapping."""
|
||||
pytest.importorskip("xatlas")
|
||||
|
||||
node = XAtlasUVUnwrapNode()
|
||||
unwrapped = node.uv_unwrap(mesh=sphere_mesh)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
assert hasattr(unwrapped.visual, 'uv') or 'uv' in unwrapped.metadata
|
||||
|
||||
save_mesh_helper(unwrapped, "sphere_xatlas_uv", "obj")
|
||||
render_helper(unwrapped, "sphere_xatlas_uv")
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_libigl_lscm(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test libigl LSCM UV unwrapping."""
|
||||
pytest.importorskip("igl")
|
||||
|
||||
node = LibiglLSCMNode()
|
||||
unwrapped = node.uv_unwrap(mesh=sphere_mesh)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
|
||||
save_mesh_helper(unwrapped, "sphere_lscm_uv", "obj")
|
||||
render_helper(unwrapped, "sphere_lscm_uv")
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_libigl_harmonic(open_mesh, save_mesh_helper, render_helper):
|
||||
"""Test libigl harmonic UV unwrapping."""
|
||||
pytest.importorskip("igl")
|
||||
|
||||
node = LibiglHarmonicNode()
|
||||
unwrapped = node.uv_unwrap(mesh=open_mesh)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
|
||||
save_mesh_helper(unwrapped, "open_harmonic_uv", "obj")
|
||||
render_helper(unwrapped, "open_harmonic_uv")
|
||||
|
||||
|
||||
@pytest.mark.optional
|
||||
def test_libigl_arap(open_mesh, save_mesh_helper, render_helper):
|
||||
"""Test libigl ARAP UV unwrapping."""
|
||||
pytest.importorskip("igl")
|
||||
|
||||
node = LibiglARAPNode()
|
||||
unwrapped = node.uv_unwrap(trimesh=open_mesh, iterations=10)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
|
||||
save_mesh_helper(unwrapped, "open_arap_uv", "obj")
|
||||
render_helper(unwrapped, "open_arap_uv")
|
||||
|
||||
|
||||
@pytest.mark.blender
|
||||
@pytest.mark.slow
|
||||
def test_blender_uv_unwrap(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test Blender Smart UV Project."""
|
||||
node = BlenderUVUnwrapNode()
|
||||
unwrapped = node.uv_unwrap(
|
||||
trimesh=sphere_mesh,
|
||||
angle_limit=66.0,
|
||||
island_margin=0.02
|
||||
)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
|
||||
save_mesh_helper(unwrapped, "sphere_blender_smart_uv", "obj")
|
||||
render_helper(unwrapped, "sphere_blender_smart_uv")
|
||||
|
||||
|
||||
@pytest.mark.blender
|
||||
@pytest.mark.slow
|
||||
def test_blender_cube_projection(cube_mesh, save_mesh_helper):
|
||||
"""Test Blender cube projection UV."""
|
||||
node = BlenderCubeProjectionNode()
|
||||
unwrapped = node.uv_unwrap(trimesh=cube_mesh, cube_size=1.0)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
|
||||
save_mesh_helper(unwrapped, "cube_projection_uv", "obj")
|
||||
|
||||
|
||||
@pytest.mark.blender
|
||||
@pytest.mark.slow
|
||||
def test_blender_cylinder_projection(sphere_mesh, save_mesh_helper):
|
||||
"""Test Blender cylinder projection UV."""
|
||||
node = BlenderCylinderProjectionNode()
|
||||
unwrapped = node.uv_unwrap(trimesh=sphere_mesh, radius=1.0)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
|
||||
save_mesh_helper(unwrapped, "sphere_cylinder_uv", "obj")
|
||||
|
||||
|
||||
@pytest.mark.blender
|
||||
@pytest.mark.slow
|
||||
def test_blender_sphere_projection(sphere_mesh, save_mesh_helper, render_helper):
|
||||
"""Test Blender sphere projection UV."""
|
||||
node = BlenderSphereProjectionNode()
|
||||
unwrapped = node.uv_unwrap(mesh=sphere_mesh)[0]
|
||||
|
||||
assert unwrapped is not None
|
||||
|
||||
save_mesh_helper(unwrapped, "sphere_sphere_uv", "obj")
|
||||
render_helper(unwrapped, "sphere_sphere_uv")
|
||||
@@ -1,69 +0,0 @@
|
||||
"""Tests for visualization nodes."""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from nodes.visualization import (
|
||||
PreviewMeshNode,
|
||||
PreviewMeshVTKNode,
|
||||
PreviewMeshVTKFiltersNode,
|
||||
PreviewMeshVTKFieldsNode,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_preview_mesh_threejs(sphere_mesh, meshes_output_dir):
|
||||
"""Test Three.js mesh preview (GLB export)."""
|
||||
node = PreviewMeshNode()
|
||||
result = node.preview_mesh(trimesh=sphere_mesh)
|
||||
|
||||
# Check that result contains UI data
|
||||
assert result is not None
|
||||
assert "ui" in result
|
||||
assert "mesh_file" in result["ui"]
|
||||
|
||||
# Verify GLB file was created in the ComfyUI output folder
|
||||
glb_files = list(meshes_output_dir.rglob("*.glb"))
|
||||
assert len(glb_files) > 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_preview_mesh_vtk(sphere_mesh, meshes_output_dir):
|
||||
"""Test VTK.js mesh preview (STL export)."""
|
||||
node = PreviewMeshVTKNode()
|
||||
result = node.preview_mesh_vtk(trimesh=sphere_mesh)
|
||||
|
||||
assert result is not None
|
||||
assert "ui" in result
|
||||
|
||||
# Verify STL file was created in the ComfyUI output folder
|
||||
stl_files = list(meshes_output_dir.rglob("*.stl"))
|
||||
assert len(stl_files) > 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_preview_mesh_vtk_filters(sphere_mesh, test_output_dir):
|
||||
"""Test VTK.js preview with filters."""
|
||||
node = PreviewMeshVTKFiltersNode()
|
||||
result = node.preview_mesh_vtk_filters(trimesh=sphere_mesh)
|
||||
|
||||
assert result is not None
|
||||
assert "ui" in result
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_preview_mesh_vtk_fields(sphere_mesh, meshes_output_dir):
|
||||
"""Test VTK.js preview with scalar fields."""
|
||||
# Add a simple scalar field
|
||||
if not hasattr(sphere_mesh, 'vertex_attributes'):
|
||||
sphere_mesh.vertex_attributes = {}
|
||||
sphere_mesh.vertex_attributes['test_field'] = sphere_mesh.vertices[:, 2]
|
||||
|
||||
node = PreviewMeshVTKFieldsNode()
|
||||
result = node.preview_mesh_vtk_fields(trimesh=sphere_mesh)
|
||||
|
||||
assert result is not None
|
||||
assert "ui" in result
|
||||
|
||||
# Verify VTP file was created (supports fields) in the ComfyUI output folder
|
||||
vtp_files = list(meshes_output_dir.rglob("*.vtp"))
|
||||
assert len(vtp_files) > 0
|
||||
@@ -1,166 +0,0 @@
|
||||
# Workflow-Based Testing
|
||||
|
||||
This directory contains infrastructure for executing and testing ComfyUI workflows via the ComfyUI API.
|
||||
|
||||
## Overview
|
||||
|
||||
Unlike traditional unit tests that instantiate nodes directly, these tests execute actual workflow JSON files through a running ComfyUI server. This provides end-to-end validation of the complete workflow execution path.
|
||||
|
||||
## Architecture
|
||||
|
||||
### Core Components
|
||||
|
||||
**WorkflowExecutor** (`workflow_executor.py`)
|
||||
- Connects to ComfyUI server via websocket
|
||||
- Queues workflow prompts for execution
|
||||
- Waits for completion and retrieves outputs
|
||||
- Usage:
|
||||
```python
|
||||
with WorkflowExecutor() as executor:
|
||||
outputs = executor.execute_workflow_file("workflow.json")
|
||||
```
|
||||
|
||||
**WorkflowConverter** (`workflow_converter.py`)
|
||||
- Converts UI workflow JSON format to API format
|
||||
- Maps widget values to input parameters
|
||||
- Resolves link connections between nodes
|
||||
- Usage:
|
||||
```python
|
||||
from workflow_converter import WorkflowConverter
|
||||
api_prompt = WorkflowConverter.convert_to_api_format(workflow_json)
|
||||
```
|
||||
|
||||
**MeshValidator** (`mesh_validators.py`)
|
||||
- Validates mesh geometry and topology
|
||||
- Checks vertex/face counts, bounding boxes, UV coordinates
|
||||
- Provides assertions for test validation
|
||||
- Usage:
|
||||
```python
|
||||
is_valid, error = MeshValidator.is_valid_mesh(mesh)
|
||||
assert is_valid, error
|
||||
```
|
||||
|
||||
## Running Tests
|
||||
|
||||
### Prerequisites
|
||||
|
||||
1. **Start ComfyUI server:**
|
||||
```bash
|
||||
cd /path/to/ComfyUI
|
||||
python main.py --listen 127.0.0.1 --port 8188
|
||||
```
|
||||
|
||||
2. **Install test dependencies:**
|
||||
```bash
|
||||
pip install pytest websocket-client
|
||||
```
|
||||
|
||||
### Run Workflow Tests
|
||||
|
||||
```bash
|
||||
# Run all workflow tests
|
||||
pytest tests/workflow_tests/ -v -m workflow
|
||||
|
||||
# Run only simple workflow tests
|
||||
pytest tests/workflow_tests/ -v -m workflow_simple
|
||||
|
||||
# Run a specific workflow test
|
||||
pytest tests/workflow_tests/test_simple_workflows.py::test_preview_mesh_workflow -v
|
||||
|
||||
# Run unit tests (no ComfyUI server needed)
|
||||
pytest tests/workflow_tests/ -v -m unit
|
||||
```
|
||||
|
||||
## Test Markers
|
||||
|
||||
Tests use pytest markers for categorization:
|
||||
|
||||
- `@pytest.mark.workflow` - All workflow execution tests
|
||||
- `@pytest.mark.workflow_simple` - Simple workflows (2-3 nodes)
|
||||
- `@pytest.mark.workflow_medium` - Medium workflows (4-5 nodes)
|
||||
- `@pytest.mark.workflow_complex` - Complex workflows (6-7 nodes)
|
||||
- `@pytest.mark.unit` - Unit tests (no server required)
|
||||
|
||||
## Adding New Workflow Tests
|
||||
|
||||
1. **Add workflow JSON to `/workflows/` directory**
|
||||
|
||||
2. **Add to test parameters:**
|
||||
```python
|
||||
SIMPLE_WORKFLOWS = [
|
||||
"preview_mesh.json",
|
||||
"your_new_workflow.json", # Add here
|
||||
]
|
||||
```
|
||||
|
||||
3. **Optionally add specific test:**
|
||||
```python
|
||||
@pytest.mark.workflow
|
||||
def test_your_workflow(workflow_executor):
|
||||
outputs = workflow_executor.execute_workflow_file("your_workflow.json")
|
||||
assert outputs is not None
|
||||
# Add custom validations
|
||||
```
|
||||
|
||||
## Workflow Format
|
||||
|
||||
Workflows can be in two formats:
|
||||
|
||||
**UI Format** (from ComfyUI save):
|
||||
```json
|
||||
{
|
||||
"nodes": [
|
||||
{
|
||||
"id": 1,
|
||||
"type": "GeomPackLoadMesh",
|
||||
"widgets_values": ["bunny.stl"],
|
||||
"inputs": [],
|
||||
"outputs": [{"name": "mesh", "links": [1]}]
|
||||
}
|
||||
],
|
||||
"links": [[1, 1, 0, 2, 0, "TRIMESH"]]
|
||||
}
|
||||
```
|
||||
|
||||
**API Format** (converted automatically):
|
||||
```json
|
||||
{
|
||||
"1": {
|
||||
"class_type": "GeomPackLoadMesh",
|
||||
"inputs": {
|
||||
"filename": "bunny.stl"
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
**"ComfyUI server not available"**
|
||||
- Ensure ComfyUI is running at 127.0.0.1:8188
|
||||
- Check firewall settings
|
||||
- Verify ComfyUI started without errors
|
||||
|
||||
**"Workflow execution failed"**
|
||||
- Check ComfyUI server logs for errors
|
||||
- Verify all required nodes are installed
|
||||
- Check that input files exist in ComfyUI input directory
|
||||
|
||||
**"websocket-client package required"**
|
||||
- Install: `pip install websocket-client`
|
||||
|
||||
## CI/CD Integration
|
||||
|
||||
For GitHub Actions, see `.github/workflows/test-workflows.yml`:
|
||||
- Installs ComfyUI
|
||||
- Starts server in background
|
||||
- Runs workflow tests
|
||||
- Uploads output artifacts
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
- Golden reference comparison (compare outputs to baseline meshes)
|
||||
- Visual diff generation (render and compare images)
|
||||
- Performance benchmarking
|
||||
- Parallel workflow execution
|
||||
- Test result dashboard
|
||||
@@ -1,14 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2025 ComfyUI-GeometryPack Contributors
|
||||
|
||||
"""
|
||||
Workflow-based testing infrastructure for ComfyUI-GeometryPack.
|
||||
|
||||
This package provides tools for executing ComfyUI workflow JSON files
|
||||
via the ComfyUI API and validating their outputs.
|
||||
"""
|
||||
|
||||
from .workflow_executor import WorkflowExecutor
|
||||
from .mesh_validators import MeshValidator
|
||||
|
||||
__all__ = ['WorkflowExecutor', 'MeshValidator']
|
||||
@@ -1,215 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2025 ComfyUI-GeometryPack Contributors
|
||||
|
||||
"""
|
||||
Mesh validation utilities for workflow testing.
|
||||
|
||||
Provides validators to check mesh outputs from workflow execution.
|
||||
"""
|
||||
|
||||
import numpy as np
|
||||
import trimesh
|
||||
|
||||
|
||||
class MeshValidator:
|
||||
"""Validates mesh geometry and topology."""
|
||||
|
||||
@staticmethod
|
||||
def is_valid_mesh(mesh):
|
||||
"""
|
||||
Check if a mesh has valid basic geometry.
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
|
||||
Returns:
|
||||
tuple: (is_valid, error_message)
|
||||
"""
|
||||
if not isinstance(mesh, trimesh.Trimesh):
|
||||
return False, f"Not a Trimesh object: {type(mesh)}"
|
||||
|
||||
if len(mesh.vertices) == 0:
|
||||
return False, "Mesh has no vertices"
|
||||
|
||||
if len(mesh.faces) == 0:
|
||||
return False, "Mesh has no faces"
|
||||
|
||||
if np.any(np.isnan(mesh.vertices)):
|
||||
return False, "Mesh contains NaN vertices"
|
||||
|
||||
if np.any(np.isinf(mesh.vertices)):
|
||||
return False, "Mesh contains infinite vertices"
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def check_vertex_count(mesh, min_vertices=3, max_vertices=None):
|
||||
"""
|
||||
Check if mesh has acceptable vertex count.
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
min_vertices: Minimum vertex count
|
||||
max_vertices: Maximum vertex count (None = no limit)
|
||||
|
||||
Returns:
|
||||
tuple: (is_valid, error_message)
|
||||
"""
|
||||
count = len(mesh.vertices)
|
||||
|
||||
if count < min_vertices:
|
||||
return False, f"Too few vertices: {count} < {min_vertices}"
|
||||
|
||||
if max_vertices and count > max_vertices:
|
||||
return False, f"Too many vertices: {count} > {max_vertices}"
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def check_face_count(mesh, min_faces=1, max_faces=None):
|
||||
"""
|
||||
Check if mesh has acceptable face count.
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
min_faces: Minimum face count
|
||||
max_faces: Maximum face count (None = no limit)
|
||||
|
||||
Returns:
|
||||
tuple: (is_valid, error_message)
|
||||
"""
|
||||
count = len(mesh.faces)
|
||||
|
||||
if count < min_faces:
|
||||
return False, f"Too few faces: {count} < {min_faces}"
|
||||
|
||||
if max_faces and count > max_faces:
|
||||
return False, f"Too many faces: {count} > {max_faces}"
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def check_manifold(mesh):
|
||||
"""
|
||||
Check if mesh is manifold (watertight, no edges shared by >2 faces).
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
|
||||
Returns:
|
||||
tuple: (is_manifold, error_message)
|
||||
"""
|
||||
try:
|
||||
is_manifold = mesh.is_watertight
|
||||
if not is_manifold:
|
||||
return False, "Mesh is not watertight/manifold"
|
||||
return True, None
|
||||
except Exception as e:
|
||||
return False, f"Error checking manifold: {e}"
|
||||
|
||||
@staticmethod
|
||||
def check_bounding_box(mesh, expected_min=None, expected_max=None, tolerance=0.1):
|
||||
"""
|
||||
Check if mesh bounding box is within expected range.
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
expected_min: Expected minimum bounds (x, y, z)
|
||||
expected_max: Expected maximum bounds (x, y, z)
|
||||
tolerance: Tolerance for comparison
|
||||
|
||||
Returns:
|
||||
tuple: (is_valid, error_message)
|
||||
"""
|
||||
bounds = mesh.bounds
|
||||
actual_min = bounds[0]
|
||||
actual_max = bounds[1]
|
||||
|
||||
if expected_min is not None:
|
||||
diff = np.abs(actual_min - np.array(expected_min))
|
||||
if np.any(diff > tolerance):
|
||||
return False, f"Bounds min mismatch: {actual_min} vs {expected_min}"
|
||||
|
||||
if expected_max is not None:
|
||||
diff = np.abs(actual_max - np.array(expected_max))
|
||||
if np.any(diff > tolerance):
|
||||
return False, f"Bounds max mismatch: {actual_max} vs {expected_max}"
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def check_has_uv_coordinates(mesh):
|
||||
"""
|
||||
Check if mesh has UV coordinates.
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
|
||||
Returns:
|
||||
tuple: (has_uvs, error_message)
|
||||
"""
|
||||
if not hasattr(mesh.visual, 'uv'):
|
||||
return False, "Mesh has no UV coordinates"
|
||||
|
||||
if mesh.visual.uv is None or len(mesh.visual.uv) == 0:
|
||||
return False, "Mesh UV coordinates are empty"
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def check_vertex_attributes(mesh, expected_attributes):
|
||||
"""
|
||||
Check if mesh has expected vertex attributes.
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
expected_attributes: List of attribute names
|
||||
|
||||
Returns:
|
||||
tuple: (has_attributes, error_message)
|
||||
"""
|
||||
if not hasattr(mesh, 'vertex_attributes'):
|
||||
return False, "Mesh has no vertex_attributes"
|
||||
|
||||
for attr in expected_attributes:
|
||||
if attr not in mesh.vertex_attributes:
|
||||
return False, f"Missing vertex attribute: {attr}"
|
||||
|
||||
return True, None
|
||||
|
||||
@classmethod
|
||||
def validate_all(cls, mesh, checks=None):
|
||||
"""
|
||||
Run all specified validation checks on a mesh.
|
||||
|
||||
Args:
|
||||
mesh: trimesh.Trimesh object
|
||||
checks: List of check names to run (None = run basic checks)
|
||||
|
||||
Returns:
|
||||
tuple: (all_valid, list of error messages)
|
||||
"""
|
||||
if checks is None:
|
||||
checks = ['is_valid_mesh', 'check_vertex_count', 'check_face_count']
|
||||
|
||||
errors = []
|
||||
|
||||
for check_name in checks:
|
||||
if not hasattr(cls, check_name):
|
||||
errors.append(f"Unknown check: {check_name}")
|
||||
continue
|
||||
|
||||
check_method = getattr(cls, check_name)
|
||||
|
||||
try:
|
||||
# Call with just mesh for basic checks
|
||||
is_valid, error = check_method(mesh)
|
||||
if not is_valid:
|
||||
errors.append(f"{check_name}: {error}")
|
||||
except TypeError:
|
||||
# Check requires additional parameters, skip
|
||||
continue
|
||||
except Exception as e:
|
||||
errors.append(f"{check_name} raised exception: {e}")
|
||||
|
||||
return len(errors) == 0, errors
|
||||
@@ -1,187 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2025 ComfyUI-GeometryPack Contributors
|
||||
|
||||
"""
|
||||
Tests for simple workflows (2-3 nodes).
|
||||
|
||||
These tests execute actual workflow JSON files through the ComfyUI API
|
||||
and validate their outputs.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from .workflow_executor import WorkflowExecutor
|
||||
from .mesh_validators import MeshValidator
|
||||
|
||||
# Path to workflows directory
|
||||
WORKFLOWS_DIR = Path(__file__).parent.parent.parent / "workflows"
|
||||
|
||||
# Simple workflow files (2-3 nodes)
|
||||
SIMPLE_WORKFLOWS = [
|
||||
"preview_mesh.json",
|
||||
"side_by_side_viewer.json",
|
||||
"transform.json",
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def workflow_executor():
|
||||
"""
|
||||
Create workflow executor connected to ComfyUI server.
|
||||
|
||||
Note: This requires a running ComfyUI server at 127.0.0.1:8188
|
||||
"""
|
||||
try:
|
||||
executor = WorkflowExecutor()
|
||||
executor.connect()
|
||||
yield executor
|
||||
executor.disconnect()
|
||||
except Exception as e:
|
||||
pytest.skip(f"ComfyUI server not available: {e}")
|
||||
|
||||
|
||||
@pytest.mark.workflow
|
||||
@pytest.mark.workflow_simple
|
||||
@pytest.mark.parametrize("workflow_file", SIMPLE_WORKFLOWS)
|
||||
def test_simple_workflow_execution(workflow_executor, workflow_file):
|
||||
"""
|
||||
Test that simple workflows execute successfully.
|
||||
|
||||
Args:
|
||||
workflow_executor: WorkflowExecutor fixture
|
||||
workflow_file: Workflow JSON filename
|
||||
"""
|
||||
workflow_path = WORKFLOWS_DIR / workflow_file
|
||||
|
||||
if not workflow_path.exists():
|
||||
pytest.skip(f"Workflow file not found: {workflow_path}")
|
||||
|
||||
# Execute workflow
|
||||
try:
|
||||
outputs = workflow_executor.execute_workflow_file(str(workflow_path), timeout=120)
|
||||
except Exception as e:
|
||||
pytest.fail(f"Workflow execution failed: {e}")
|
||||
|
||||
# Validate outputs exist
|
||||
assert outputs is not None, "Workflow produced no outputs"
|
||||
assert len(outputs) > 0, "Workflow output dictionary is empty"
|
||||
|
||||
|
||||
@pytest.mark.workflow
|
||||
@pytest.mark.workflow_simple
|
||||
def test_preview_mesh_workflow(workflow_executor):
|
||||
"""
|
||||
Test preview_mesh.json workflow specifically.
|
||||
|
||||
Workflow: LoadMesh → PreviewMesh
|
||||
Expected: Mesh loads and preview node executes
|
||||
"""
|
||||
workflow_path = WORKFLOWS_DIR / "preview_mesh.json"
|
||||
|
||||
if not workflow_path.exists():
|
||||
pytest.skip(f"Workflow file not found: {workflow_path}")
|
||||
|
||||
outputs = workflow_executor.execute_workflow_file(str(workflow_path), timeout=60)
|
||||
|
||||
# Check outputs exist
|
||||
assert outputs is not None
|
||||
assert len(outputs) > 0, "No outputs from preview_mesh workflow"
|
||||
|
||||
# Note: Preview nodes typically don't return mesh data in outputs,
|
||||
# they just display. Success here means no errors during execution.
|
||||
|
||||
|
||||
@pytest.mark.workflow
|
||||
@pytest.mark.workflow_simple
|
||||
def test_transform_workflow(workflow_executor):
|
||||
"""
|
||||
Test transform.json workflow specifically.
|
||||
|
||||
Workflow: CreatePrimitive → TransformMesh → PreviewMeshDual
|
||||
Expected: Primitive created, transformed, and previewed
|
||||
"""
|
||||
workflow_path = WORKFLOWS_DIR / "transform.json"
|
||||
|
||||
if not workflow_path.exists():
|
||||
pytest.skip(f"Workflow file not found: {workflow_path}")
|
||||
|
||||
outputs = workflow_executor.execute_workflow_file(str(workflow_path), timeout=60)
|
||||
|
||||
# Check outputs exist
|
||||
assert outputs is not None
|
||||
assert len(outputs) > 0, "No outputs from transform workflow"
|
||||
|
||||
|
||||
@pytest.mark.workflow
|
||||
@pytest.mark.workflow_simple
|
||||
def test_side_by_side_viewer_workflow(workflow_executor):
|
||||
"""
|
||||
Test side_by_side_viewer.json workflow specifically.
|
||||
|
||||
Workflow: CreatePrimitive → PreviewMeshDual
|
||||
Expected: Primitive created and displayed in dual viewer
|
||||
"""
|
||||
workflow_path = WORKFLOWS_DIR / "side_by_side_viewer.json"
|
||||
|
||||
if not workflow_path.exists():
|
||||
pytest.skip(f"Workflow file not found: {workflow_path}")
|
||||
|
||||
outputs = workflow_executor.execute_workflow_file(str(workflow_path), timeout=60)
|
||||
|
||||
# Check outputs exist
|
||||
assert outputs is not None
|
||||
assert len(outputs) > 0, "No outputs from side_by_side_viewer workflow"
|
||||
|
||||
|
||||
# Validation helper tests (can be used for debugging)
|
||||
@pytest.mark.unit
|
||||
def test_mesh_validator_basic():
|
||||
"""Test MeshValidator basic functionality."""
|
||||
import trimesh
|
||||
import numpy as np
|
||||
|
||||
# Create a simple valid mesh
|
||||
vertices = np.array([[0, 0, 0], [1, 0, 0], [0, 1, 0]])
|
||||
faces = np.array([[0, 1, 2]])
|
||||
mesh = trimesh.Trimesh(vertices=vertices, faces=faces)
|
||||
|
||||
# Should pass validation
|
||||
is_valid, error = MeshValidator.is_valid_mesh(mesh)
|
||||
assert is_valid, f"Valid mesh failed validation: {error}"
|
||||
|
||||
# Check vertex count
|
||||
is_valid, error = MeshValidator.check_vertex_count(mesh, min_vertices=3, max_vertices=10)
|
||||
assert is_valid, f"Vertex count check failed: {error}"
|
||||
|
||||
# Check face count
|
||||
is_valid, error = MeshValidator.check_face_count(mesh, min_faces=1, max_faces=10)
|
||||
assert is_valid, f"Face count check failed: {error}"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mesh_validator_invalid():
|
||||
"""Test MeshValidator with invalid meshes."""
|
||||
import trimesh
|
||||
import numpy as np
|
||||
|
||||
# Create a valid mesh first
|
||||
vertices = np.array([[0, 0, 0], [1, 0, 0], [0, 1, 0]])
|
||||
faces = np.array([[0, 1, 2]])
|
||||
mesh = trimesh.Trimesh(vertices=vertices, faces=faces)
|
||||
|
||||
# Now manually inject NaN (trimesh may clean it during construction)
|
||||
mesh.vertices[1, 0] = np.nan
|
||||
|
||||
is_valid, error = MeshValidator.is_valid_mesh(mesh)
|
||||
assert not is_valid, "Mesh with NaN vertices should fail validation"
|
||||
assert "NaN" in error
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_workflow_converter_import():
|
||||
"""Test that WorkflowConverter can be imported."""
|
||||
from .workflow_converter import WorkflowConverter
|
||||
|
||||
assert WorkflowConverter is not None
|
||||
assert hasattr(WorkflowConverter, 'convert_to_api_format')
|
||||
assert hasattr(WorkflowConverter, 'load_workflow')
|
||||
@@ -1,126 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2025 ComfyUI-GeometryPack Contributors
|
||||
|
||||
"""
|
||||
Workflow JSON to API format converter.
|
||||
|
||||
Converts ComfyUI UI workflow JSON format to the API prompt format
|
||||
required by the ComfyUI server.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Dict, Any, List
|
||||
|
||||
|
||||
class WorkflowConverter:
|
||||
"""Converts workflow UI JSON to API format."""
|
||||
|
||||
@staticmethod
|
||||
def convert_to_api_format(workflow_json: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Convert UI workflow JSON to API prompt format.
|
||||
|
||||
Args:
|
||||
workflow_json: Workflow in UI format (has 'nodes' and 'links' arrays)
|
||||
|
||||
Returns:
|
||||
Dict in API format (node IDs as keys)
|
||||
"""
|
||||
api_prompt = {}
|
||||
|
||||
# Build link lookup table: link_id -> (source_node, source_output_slot)
|
||||
link_map = {}
|
||||
for link in workflow_json.get('links', []):
|
||||
# Link format: [link_id, source_node, source_slot, dest_node, dest_slot, type]
|
||||
link_id = link[0]
|
||||
source_node = link[1]
|
||||
source_slot = link[2]
|
||||
link_map[link_id] = (source_node, source_slot)
|
||||
|
||||
# Convert each node
|
||||
for node in workflow_json.get('nodes', []):
|
||||
node_id = str(node['id'])
|
||||
node_type = node['type']
|
||||
|
||||
api_node = {
|
||||
'class_type': node_type,
|
||||
'inputs': {}
|
||||
}
|
||||
|
||||
# Process input connections (from links)
|
||||
for input_def in node.get('inputs', []):
|
||||
input_name = input_def['name']
|
||||
link_id = input_def.get('link')
|
||||
|
||||
if link_id is not None and link_id in link_map:
|
||||
source_node, source_slot = link_map[link_id]
|
||||
api_node['inputs'][input_name] = [str(source_node), source_slot]
|
||||
|
||||
# Process widget values (direct inputs)
|
||||
widget_values = node.get('widgets_values', [])
|
||||
if widget_values:
|
||||
# Map widget values to input names
|
||||
# This requires knowing the node's INPUT_TYPES, but we'll use a heuristic:
|
||||
# Get input names that don't have links (they're widget inputs)
|
||||
linked_inputs = set(inp['name'] for inp in node.get('inputs', []) if inp.get('link') is not None)
|
||||
|
||||
# Try to map widget values to non-linked inputs by order
|
||||
widget_input_names = [inp['name'] for inp in node.get('inputs', []) if inp['name'] not in linked_inputs]
|
||||
|
||||
for i, widget_value in enumerate(widget_values):
|
||||
if i < len(widget_input_names):
|
||||
api_node['inputs'][widget_input_names[i]] = widget_value
|
||||
else:
|
||||
# If we have more widget values than expected inputs,
|
||||
# use generic names (this might not work perfectly)
|
||||
api_node['inputs'][f'widget_{i}'] = widget_value
|
||||
|
||||
api_prompt[node_id] = api_node
|
||||
|
||||
return api_prompt
|
||||
|
||||
@staticmethod
|
||||
def load_workflow(workflow_path: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Load workflow JSON from file.
|
||||
|
||||
Args:
|
||||
workflow_path: Path to workflow JSON file
|
||||
|
||||
Returns:
|
||||
Workflow dictionary
|
||||
"""
|
||||
with open(workflow_path, 'r') as f:
|
||||
return json.load(f)
|
||||
|
||||
@staticmethod
|
||||
def save_api_prompt(api_prompt: Dict[str, Any], output_path: str):
|
||||
"""
|
||||
Save API prompt to JSON file.
|
||||
|
||||
Args:
|
||||
api_prompt: API format prompt
|
||||
output_path: Output file path
|
||||
"""
|
||||
with open(output_path, 'w') as f:
|
||||
json.dump(api_prompt, f, indent=2)
|
||||
|
||||
@classmethod
|
||||
def convert_workflow_file(cls, workflow_path: str, output_path: str = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Load workflow from file, convert to API format, and optionally save.
|
||||
|
||||
Args:
|
||||
workflow_path: Path to workflow JSON file
|
||||
output_path: Optional path to save API format
|
||||
|
||||
Returns:
|
||||
API format prompt
|
||||
"""
|
||||
workflow = cls.load_workflow(workflow_path)
|
||||
api_prompt = cls.convert_to_api_format(workflow)
|
||||
|
||||
if output_path:
|
||||
cls.save_api_prompt(api_prompt, output_path)
|
||||
|
||||
return api_prompt
|
||||
@@ -1,206 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2025 ComfyUI-GeometryPack Contributors
|
||||
|
||||
"""
|
||||
Workflow executor for running ComfyUI workflows via API.
|
||||
|
||||
Executes workflow JSON files through the ComfyUI server API and
|
||||
retrieves results.
|
||||
"""
|
||||
|
||||
import json
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
import uuid
|
||||
import time
|
||||
from typing import Dict, Any, Optional
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import websocket
|
||||
WEBSOCKET_AVAILABLE = True
|
||||
except ImportError:
|
||||
WEBSOCKET_AVAILABLE = False
|
||||
|
||||
|
||||
class WorkflowExecutor:
|
||||
"""Executes ComfyUI workflows via the server API."""
|
||||
|
||||
def __init__(self, server_address="127.0.0.1:8188"):
|
||||
"""
|
||||
Initialize workflow executor.
|
||||
|
||||
Args:
|
||||
server_address: ComfyUI server address (host:port)
|
||||
"""
|
||||
self.server_address = server_address
|
||||
self.client_id = str(uuid.uuid4())
|
||||
self.ws = None
|
||||
|
||||
def connect(self):
|
||||
"""Connect to ComfyUI websocket for real-time updates."""
|
||||
if not WEBSOCKET_AVAILABLE:
|
||||
raise ImportError("websocket-client package required for workflow execution")
|
||||
|
||||
ws_url = f"ws://{self.server_address}/ws?clientId={self.client_id}"
|
||||
self.ws = websocket.WebSocket()
|
||||
self.ws.connect(ws_url)
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from ComfyUI websocket."""
|
||||
if self.ws:
|
||||
self.ws.close()
|
||||
self.ws = None
|
||||
|
||||
def queue_prompt(self, prompt: Dict[str, Any]) -> str:
|
||||
"""
|
||||
Queue a prompt for execution.
|
||||
|
||||
Args:
|
||||
prompt: API format prompt
|
||||
|
||||
Returns:
|
||||
Prompt ID for tracking
|
||||
"""
|
||||
p = {"prompt": prompt, "client_id": self.client_id}
|
||||
data = json.dumps(p).encode('utf-8')
|
||||
|
||||
req = urllib.request.Request(
|
||||
f"http://{self.server_address}/prompt",
|
||||
data=data,
|
||||
headers={'Content-Type': 'application/json'}
|
||||
)
|
||||
|
||||
response = urllib.request.urlopen(req)
|
||||
result = json.loads(response.read())
|
||||
|
||||
if 'prompt_id' not in result:
|
||||
raise RuntimeError(f"Failed to queue prompt: {result}")
|
||||
|
||||
return result['prompt_id']
|
||||
|
||||
def wait_for_completion(self, prompt_id: str, timeout: int = 300) -> bool:
|
||||
"""
|
||||
Wait for prompt execution to complete.
|
||||
|
||||
Args:
|
||||
prompt_id: Prompt ID to wait for
|
||||
timeout: Maximum wait time in seconds
|
||||
|
||||
Returns:
|
||||
True if completed successfully
|
||||
"""
|
||||
if not self.ws:
|
||||
raise RuntimeError("Not connected to websocket")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
while True:
|
||||
if time.time() - start_time > timeout:
|
||||
raise TimeoutError(f"Workflow execution timed out after {timeout}s")
|
||||
|
||||
out = self.ws.recv()
|
||||
|
||||
if isinstance(out, str):
|
||||
message = json.loads(out)
|
||||
|
||||
if message['type'] == 'executing':
|
||||
data = message['data']
|
||||
|
||||
# When node is None and prompt_id matches, execution is done
|
||||
if data['node'] is None and data['prompt_id'] == prompt_id:
|
||||
return True
|
||||
|
||||
elif message['type'] == 'execution_error':
|
||||
error_data = message['data']
|
||||
raise RuntimeError(f"Execution error: {error_data}")
|
||||
|
||||
def get_history(self, prompt_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Get execution history for a prompt.
|
||||
|
||||
Args:
|
||||
prompt_id: Prompt ID
|
||||
|
||||
Returns:
|
||||
History data including outputs
|
||||
"""
|
||||
url = f"http://{self.server_address}/history/{prompt_id}"
|
||||
|
||||
with urllib.request.urlopen(url) as response:
|
||||
history = json.loads(response.read())
|
||||
|
||||
if prompt_id not in history:
|
||||
raise RuntimeError(f"Prompt {prompt_id} not found in history")
|
||||
|
||||
return history[prompt_id]
|
||||
|
||||
def get_outputs(self, prompt_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Get outputs from executed workflow.
|
||||
|
||||
Args:
|
||||
prompt_id: Prompt ID
|
||||
|
||||
Returns:
|
||||
Dictionary of node outputs
|
||||
"""
|
||||
history = self.get_history(prompt_id)
|
||||
return history.get('outputs', {})
|
||||
|
||||
def execute_workflow(self, workflow: Dict[str, Any], timeout: int = 300) -> Dict[str, Any]:
|
||||
"""
|
||||
Execute a workflow and return outputs.
|
||||
|
||||
Args:
|
||||
workflow: Workflow in API format
|
||||
timeout: Maximum execution time in seconds
|
||||
|
||||
Returns:
|
||||
Dictionary of outputs by node ID
|
||||
"""
|
||||
# Queue the workflow
|
||||
prompt_id = self.queue_prompt(workflow)
|
||||
|
||||
# Wait for completion
|
||||
self.wait_for_completion(prompt_id, timeout)
|
||||
|
||||
# Get and return outputs
|
||||
return self.get_outputs(prompt_id)
|
||||
|
||||
def execute_workflow_file(self, workflow_path: str, timeout: int = 300) -> Dict[str, Any]:
|
||||
"""
|
||||
Load and execute a workflow JSON file.
|
||||
|
||||
Args:
|
||||
workflow_path: Path to workflow JSON (UI or API format)
|
||||
timeout: Maximum execution time in seconds
|
||||
|
||||
Returns:
|
||||
Dictionary of outputs by node ID
|
||||
"""
|
||||
from .workflow_converter import WorkflowConverter
|
||||
|
||||
# Load workflow
|
||||
with open(workflow_path, 'r') as f:
|
||||
workflow_json = json.load(f)
|
||||
|
||||
# Check if it's UI format (has 'nodes' array) or API format
|
||||
if 'nodes' in workflow_json:
|
||||
# Convert from UI format to API format
|
||||
workflow = WorkflowConverter.convert_to_api_format(workflow_json)
|
||||
else:
|
||||
# Already in API format
|
||||
workflow = workflow_json
|
||||
|
||||
return self.execute_workflow(workflow, timeout)
|
||||
|
||||
def __enter__(self):
|
||||
"""Context manager entry."""
|
||||
self.connect()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Context manager exit."""
|
||||
self.disconnect()
|
||||
return False
|
||||
Reference in New Issue
Block a user