NaraMotion / tests /test_cross_module_integration.py
ArifiIslam's picture
Release NaraMotion v0.26.0: Structure of Arrays (SoA), Path Arena Allocator, Uniform Grid Indexing, Dual-Buffer Rendering & Centroid Spring Pop-in
b516e00 verified
Raw History Blame Contribute Delete
6.77 kB
"""
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)