r.watershed: Add comprehensive tests for map reuse functionality

raster/r.watershed/testsuite/test_r_watershed_reuse.py

@@ -0,0 +1,216 @@
+"""
+Name:      r_watershed_reuse_test
+Purpose:   Unit tests for r.watershed's ability to reuse existing accumulation
+           and drainage maps instead of recalculating them.
+
+Author:    Sumit Chintanwar
+Copyright: (C) 2026 by Sumit Chintanwar and the GRASS Development Team
+Licence:   This program is free software under the GNU General Public
+           License (>=v2). Read the file COPYING that comes with GRASS
+           for details.
+"""
+
+import unittest
+from grass.gunittest.case import TestCase
+from grass.gunittest.main import test
+
+
+class TestWatershedReuse(TestCase):
+    """Test r.watershed input map reuse functionality"""
+
+    accumulation = "reuse_accum"
+    drainage = "reuse_drain"
+    elevation = "reuse_elevation"
+
+    @classmethod
+    def setUpClass(cls):
+        cls.use_temp_region()
+
+        cls.runModule(
+            "r.mapcalc",
+            expression=f"{cls.elevation} = row() + col()",
+        )
+
+        cls.runModule("g.region", raster=cls.elevation, res=10)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.runModule("g.remove", flags="f", type="raster", name=cls.elevation)
+        cls.del_temp_region()
+
+    def tearDown(self):
+        self.runModule(
+            "g.remove",
+            flags="f",
+            type="raster",
+            pattern="reuse_*",
+            exclude=self.elevation,
+        )
+
+    @unittest.expectedFailure
+    def test_reuse_workflow(self):
+        """Test complete reuse workflow: generate, reuse for accumulation and RUSLE.
+        Expected to fail until accumulation_input and drainage_input parameters are implemented.
+        """
+        # Generate flow maps
+        self.assertModule(
+            "r.watershed",
+            elevation=self.elevation,
+            accumulation=self.accumulation,
+            drainage=self.drainage,
+            threshold=1000,
+        )
+
+        self.assertRasterExists(self.accumulation)
+        self.assertRasterExists(self.drainage)
+
+        self.assertRasterMinMax(
+            self.drainage,
+            refmin=-8,
+            refmax=8,
+            msg="Drainage direction must be between -8 and 8",
+        )
+
+        accum_out = "reuse_accum_out"
+        self.assertModule(
+            "r.watershed",
+            elevation=self.elevation,
+            accumulation_input=self.accumulation,
+            drainage_input=self.drainage,
+            accumulation=accum_out,
+        )
+
+        self.assertRastersNoDifference(
+            actual=accum_out,
+            reference=self.accumulation,
+            precision=0.01,
+            msg="Reused accumulation should match original",
+        )
+
+        ls_factor = "reuse_ls"
+        self.assertModule(
+            "r.watershed",
+            elevation=self.elevation,
+            accumulation_input=self.accumulation,
+            drainage_input=self.drainage,
+            threshold=1000,
+            length_slope=ls_factor,
+            max_slope_length=100,
+        )
+
+        self.assertRasterExists(ls_factor)
+
+    @unittest.expectedFailure
+    def test_input_validation(self):
+        """Test input validation: both inputs required, depression incompatible.
+        Expected to fail until accumulation_input and drainage_input parameters are implemented.
+        """
+        self.assertModule(
+            "r.watershed",
+            elevation=self.elevation,
+            accumulation=self.accumulation,
+            drainage=self.drainage,
+        )
+
+        # Missing drainage_input should fail
+        self.assertModuleFail(
+            "r.watershed",
+            elevation=self.elevation,
+            accumulation_input=self.accumulation,
+            accumulation="reuse_out",
+        )
+
+        # Missing accumulation_input should fail
+        self.assertModuleFail(
+            "r.watershed",
+            elevation=self.elevation,
+            drainage_input=self.drainage,
+            accumulation="reuse_out",
+        )
+
+        # Depression incompatible with input maps
+        self.runModule("r.mapcalc", expression="reuse_dep = 1")
+        self.assertModuleFail(
+            "r.watershed",
+            elevation=self.elevation,
+            accumulation_input=self.accumulation,
+            drainage_input=self.drainage,
+            depression="reuse_dep",
+            accumulation="reuse_out",
+        )
+
+    @unittest.skip(
+        "Basin delineation with input maps not yet supported - TODO: recalculate flow"
+    )
+    def test_basin_limitation(self):
+        """TODO: Add basin delineation test once supported"""
+
+
+@unittest.skip("DEACTIVATED FOR CI")
+class TestWatershedReuseLarge(TestCase):
+    """Performance test on full extent - SKIP FOR CI
+    Remove @unittest.skip to run.
+    Expected to fail until accumulation_input and drainage_input parameters are implemented.
+
+    """
+
+    elevation = "reuse_large_elevation"
+
+    @classmethod
+    def setUpClass(cls):
+        """Set up larger region for performance testing"""
+        cls.use_temp_region()
+
+        cls.runModule("g.region", n=5000, s=0, e=5000, w=0, res=10)
+        cls.runModule(
+            "r.mapcalc",
+            expression=f"{cls.elevation} = row() + col() + rand(0, 100)",
+        )
+        cls.runModule("g.region", raster=cls.elevation)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.runModule("g.remove", flags="f", type="raster", name=cls.elevation)
+        cls.del_temp_region()
+
+    def tearDown(self):
+        self.runModule(
+            "g.remove",
+            flags="f",
+            type="raster",
+            pattern="reuse_large_*",
+            exclude=self.elevation,
+        )
+
+    @unittest.skip("DEACTIVATED FOR CI")
+    def test_large_area_iterations(self):
+        """Test parameter iterations without recalculating flow
+        Expected to fail until accumulation_input and drainage_input parameters are implemented.
+
+        """
+        accum, drain = "reuse_large_accum", "reuse_large_drain"
+
+        self.assertModule(
+            "r.watershed",
+            elevation=self.elevation,
+            accumulation=accum,
+            drainage=drain,
+        )
+
+        for length in [50, 100, 200, 500, 1000]:
+            ls_name = f"reuse_large_ls_{length}"
+            self.assertModule(
+                "r.watershed",
+                elevation=self.elevation,
+                accumulation_input=accum,
+                drainage_input=drain,
+                threshold=1000,
+                length_slope=ls_name,
+                max_slope_length=length,
+                overwrite=True,
+            )
+            self.assertRasterExists(ls_name)
+
+
+if __name__ == "__main__":
+    test()