File size: 6,768 Bytes
b516e00 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | """
Comprehensive Cross-Module Compatibility & Logical / Illogical Stress Test Suite.
Verifies seamless inter-operation across ALL library branches:
- blend2d (Core Engine, Context, Path, Image, Matrix)
- naramotion.motion (Timeline, Easing, Physics Springs, Particles, Scene, PathMorph)
- naramotion.shapgeo (GeoPandas, Shapefiles, GeoJSON, Projections, GIS layers)
- naramotion.video (FFmpeg C-API Video Encoding & Decoding)
- naramotion.svg (Vector SVG Exporter)
"""
import os
import math
import pytest
import numpy as np
from shapely.geometry import Point, LineString, Polygon, MultiPolygon
import geopandas as gpd
import naramotion
from naramotion import (
Image, Context, Format, Path, Matrix2D, FillRule, Blend2DError,
VideoWriter, VideoReader,
export_svg,
GeoCanvas, GeoTransform, from_shapely, render_gdf, read_geojson,
to_web_mercator, reproject_gdf,
Timeline, Easing, Spring, trim_path, PathFollower, PathMorph,
ParticleSystem, Camera2D, Node, Layer, Scene
)
# ============================================================================
# 1. Full Cross-Module End-to-End Pipeline
# ============================================================================
def test_full_cross_module_pipeline(tmp_path):
"""
Test entire workflow:
1. Create GIS polygons in Shapely (GeoPandas).
2. Reproject via C++ projection engine (WGS84 -> Web Mercator).
3. Animate camera and vector trim reveals via Timeline (naramotion.motion).
4. Render video frames into native H.264 video via naramotion.video.
5. Export keyframe to vector SVG via naramotion.svg.
"""
# 1. GIS Data Setup
poly = Polygon([(-74.0, 40.7), (-73.9, 40.7), (-73.9, 40.8), (-74.0, 40.8)])
gdf = gpd.GeoDataFrame({"name": ["Manhattan"], "geometry": [poly]}, crs="EPSG:4326")
# 2. C++ Projection
gdf_m = reproject_gdf(gdf, target_crs="EPSG:3857")
gt = GeoTransform(gdf_m.total_bounds, canvas_width=1280, canvas_height=720)
# 3. Motion & Animation Setup
path_geo = from_shapely(gdf_m.geometry.iloc[0], gt)
timeline = Timeline(duration=0.5, fps=30)
ps = ParticleSystem()
ps.preset_sparks(640, 360, count=40)
video_file = str(tmp_path / "cross_pipeline.mp4")
svg_file = str(tmp_path / "frame.svg")
def render_frame(ctx: Context, t: float, frame_idx: int):
ctx.fill_all("#090D16")
# Line trim animation
revealed_path = trim_path(path_geo, 0.0, Easing.cubic_out(t / 0.5))
ctx.stroke_width = 3.0
ctx.stroke_path(revealed_path, style="#38BDF8")
# Particles
ps.update(1.0 / 30.0)
ps.render(ctx)
# 4. Render Video via C++ Engine
timeline.render_video(video_file, width=1280, height=720, fps=30, draw_fn=render_frame)
assert os.path.exists(video_file)
assert os.path.getsize(video_file) > 1000
# 5. Export SVG Vector Frame
export_svg(path_geo, svg_file, width=1280, height=720)
assert os.path.exists(svg_file)
assert os.path.getsize(svg_file) > 100
# ============================================================================
# 2. Logical Edge Cases
# ============================================================================
def test_edge_case_zero_duration_and_empty_paths():
# 1. Zero duration timeline
tl = Timeline(duration=0.0, fps=60)
tl.seek(0.0)
assert tl.current_time == 0.0
# 2. Empty path trimming
empty_p = Path()
trimmed = trim_path(empty_p, 0.2, 0.8)
assert trimmed.is_empty()
# 3. Degenerate single point path
p1 = Path()
p1.move_to(50, 50)
trimmed1 = trim_path(p1, 0.0, 1.0)
assert isinstance(trimmed1, Path)
# 4. Out of bounds timeline seeking
tl2 = Timeline(duration=2.0)
tl2.seek(-10.0)
assert tl2.current_time == 0.0
tl2.seek(100.0)
assert tl2.current_time == 2.0
def test_edge_case_extreme_geometries():
# Giant coordinates (10^9) and microscopic coordinates (10^-9)
giant_poly = Polygon([(0, 0), (1e9, 0), (1e9, 1e9), (0, 1e9)])
micro_poly = Polygon([(0, 0), (1e-9, 0), (1e-9, 1e-9), (0, 1e-9)])
p_giant = from_shapely(giant_poly)
p_micro = from_shapely(micro_poly)
assert isinstance(p_giant, Path)
assert isinstance(p_micro, Path)
img = Image(100, 100, Format.PRGB32)
with Context(img) as ctx:
ctx.fill_path(p_giant, style="red")
ctx.fill_path(p_micro, style="blue")
# ============================================================================
# 3. Illogical Inputs & Robustness Hardening
# ============================================================================
def test_illogical_easing_inputs():
# Negative and extreme progress values
for t in [-1000.0, -1.0, 2.0, 500.0]:
val_elastic = Easing.elastic_out(t)
val_bounce = Easing.bounce_in_out(t)
assert not math.isnan(val_elastic)
assert not math.isnan(val_bounce)
def test_illogical_nan_and_inf_resilience():
# Feeding NaN / Inf coordinates to projection
coords = np.array([
[float('nan'), float('nan')],
[float('inf'), float('-inf')],
[0.0, 0.0]
])
from naramotion.shapgeo import Transformer
t = Transformer.from_crs("EPSG:4326", "EPSG:3857")
# Must not crash process or segfault
res = t.transform_array(coords)
assert res.shape == (3, 2)
assert math.isclose(res[2, 0], 0.0, abs_tol=1e-3)
def test_deep_scene_hierarchy_stress():
# 100 deeply nested scene graph nodes
root = Node()
cur = root
for i in range(100):
next_node = Node(x=1.0, y=1.0)
cur.add_child(next_node)
cur = next_node
mat = cur.get_world_matrix()
assert math.isclose(mat.m20, 100.0, abs_tol=1e-3)
assert math.isclose(mat.m21, 100.0, abs_tol=1e-3)
def test_concurrency_and_memory_stress():
"""Stress test multithreaded rendering with simultaneous particle simulations and GIS conversions."""
import concurrent.futures
def worker_task(idx):
img = Image(400, 400, Format.PRGB32)
ps = ParticleSystem(max_particles=500)
ps.preset_explosion(200, 200, count=100)
with Context(img, thread_count=2) as ctx:
for _ in range(10):
ps.update(0.016)
ctx.fill_all("black")
ps.render(ctx)
return img.width
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
futures = [executor.submit(worker_task, i) for i in range(8)]
results = [f.result() for f in futures]
assert all(r == 400 for r in results)
|