diff --git a/raster/r.watershed/testsuite/test_r_watershed_reuse.py b/raster/r.watershed/testsuite/test_r_watershed_reuse.py new file mode 100644 index 00000000000..3dbc4ae02a1 --- /dev/null +++ b/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()