KSvend Claude Happy commited on
Commit
7403c8c
·
1 Parent(s): 1ee8d52

test: add end-to-end smoke test for NDVI openEO pipeline

Browse files

Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>

Files changed (1) hide show
  1. tests/test_ndvi_e2e.py +124 -0
tests/test_ndvi_e2e.py ADDED
@@ -0,0 +1,124 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """End-to-end test: NDVI indicator → raster map → chart → report section.
2
+
3
+ Uses mocked openEO (no CDSE credentials needed) with synthetic GeoTIFFs.
4
+ """
5
+ from __future__ import annotations
6
+
7
+ import os
8
+ import tempfile
9
+ from unittest.mock import MagicMock, patch
10
+ from datetime import date
11
+
12
+ import numpy as np
13
+ import rasterio
14
+ from rasterio.transform import from_bounds
15
+ import pytest
16
+
17
+ from app.models import AOI, TimeRange, StatusLevel
18
+ from app.outputs.charts import render_timeseries_chart
19
+ from app.outputs.maps import render_raster_map
20
+
21
+
22
+ BBOX = [32.45, 15.65, 32.65, 15.8]
23
+ WIDTH, HEIGHT = 15, 12
24
+
25
+
26
+ def _write_ndvi_tif(path: str):
27
+ rng = np.random.default_rng(42)
28
+ data = np.zeros((12, HEIGHT, WIDTH), dtype=np.float32)
29
+ for m in range(12):
30
+ data[m] = 0.3 + 0.2 * np.sin(np.pi * (m - 3) / 6) + rng.normal(0, 0.03, (HEIGHT, WIDTH))
31
+ with rasterio.open(
32
+ path, "w", driver="GTiff", height=HEIGHT, width=WIDTH, count=12,
33
+ dtype="float32", crs="EPSG:4326",
34
+ transform=from_bounds(*BBOX, WIDTH, HEIGHT), nodata=-9999.0,
35
+ ) as dst:
36
+ for i in range(12):
37
+ dst.write(data[i], i + 1)
38
+
39
+
40
+ def _write_rgb_tif(path: str):
41
+ rng = np.random.default_rng(43)
42
+ data = rng.integers(500, 1500, (3, HEIGHT, WIDTH), dtype=np.uint16)
43
+ with rasterio.open(
44
+ path, "w", driver="GTiff", height=HEIGHT, width=WIDTH, count=3,
45
+ dtype="uint16", crs="EPSG:4326",
46
+ transform=from_bounds(*BBOX, WIDTH, HEIGHT), nodata=0,
47
+ ) as dst:
48
+ for i in range(3):
49
+ dst.write(data[i], i + 1)
50
+
51
+
52
+ @pytest.mark.asyncio
53
+ async def test_ndvi_full_pipeline():
54
+ """Full pipeline: process NDVI → render raster map → render chart."""
55
+ from app.indicators.ndvi import NdviIndicator
56
+
57
+ aoi = AOI(name="Khartoum Test", bbox=BBOX)
58
+ time_range = TimeRange(start=date(2025, 3, 1), end=date(2026, 3, 1))
59
+
60
+ with tempfile.TemporaryDirectory() as tmpdir:
61
+ ndvi_path = os.path.join(tmpdir, "ndvi.tif")
62
+ rgb_path = os.path.join(tmpdir, "rgb.tif")
63
+ _write_ndvi_tif(ndvi_path)
64
+ _write_rgb_tif(rgb_path)
65
+
66
+ mock_cube = MagicMock()
67
+
68
+ def fake_download(path, **kwargs):
69
+ import shutil
70
+ if "ndvi" in path:
71
+ shutil.copy(ndvi_path, path)
72
+ else:
73
+ shutil.copy(rgb_path, path)
74
+
75
+ mock_cube.download = MagicMock(side_effect=fake_download)
76
+
77
+ with patch("app.indicators.ndvi.get_connection") as mock_conn, \
78
+ patch("app.indicators.ndvi.build_ndvi_graph", return_value=mock_cube), \
79
+ patch("app.indicators.ndvi.build_true_color_graph", return_value=mock_cube):
80
+
81
+ indicator = NdviIndicator()
82
+ result = await indicator.process(aoi, time_range)
83
+
84
+ # Verify result quality
85
+ assert result.indicator_id == "ndvi"
86
+ assert result.data_source == "satellite"
87
+ assert "NDVI" in result.methodology
88
+ assert "pixel" in result.methodology.lower()
89
+ assert len(result.chart_data["dates"]) >= 6
90
+ assert all(isinstance(v, float) for v in result.chart_data["values"])
91
+
92
+ # Render the raster map
93
+ map_out = os.path.join(tmpdir, "ndvi_map.png")
94
+ raster_path = indicator._indicator_raster_path
95
+ tc_path = indicator._true_color_path
96
+ peak = indicator._ndvi_peak_band
97
+
98
+ render_raster_map(
99
+ true_color_path=tc_path,
100
+ indicator_path=raster_path,
101
+ indicator_band=peak,
102
+ aoi=aoi,
103
+ status=result.status,
104
+ output_path=map_out,
105
+ cmap="RdYlGn",
106
+ vmin=-0.2,
107
+ vmax=0.9,
108
+ label="NDVI",
109
+ )
110
+ assert os.path.exists(map_out)
111
+ assert os.path.getsize(map_out) > 10000 # Real map should be >10KB
112
+
113
+ # Render the chart
114
+ chart_out = os.path.join(tmpdir, "ndvi_chart.png")
115
+ render_timeseries_chart(
116
+ chart_data=result.chart_data,
117
+ indicator_name="Vegetation (NDVI)",
118
+ status=result.status,
119
+ trend=result.trend,
120
+ output_path=chart_out,
121
+ y_label="NDVI",
122
+ )
123
+ assert os.path.exists(chart_out)
124
+ assert os.path.getsize(chart_out) > 5000