GTC-2753 Add WDPA AUM test

src/test/resources/wdpa-aum-output/part-00000-702eaaae-1d14-4aed-97f1-0574a1e2af88-c000.csv

@@ -0,0 +1,2 @@
+wdpa_protected_area__id	wdpa_protected_area__name	wdpa_protected_area__iso	wdpa_protected_area__status	umd_tree_cover_loss__year	umd_tree_cover_density_2000__threshold	tsc_tree_cover_loss_drivers__driver	is__umd_regional_primary_forest_2001	is__birdlife_alliance_for_zero_extinction_sites	gfw_planted_forests__type	is__gmw_global_mangrove_extent_1996	is__gmw_global_mangrove_extent_2020	is__wwf_tiger_conservation_landscapes	is__landmark_indigenous_and_community_lands	is__birdlife_key_biodiversity_areas	is__gfw_mining_concessions	is__gfw_peatlands	is__gfw_oil_palm	is__idn_forest_moratorium	is__gfw_wood_fiber	is__gfw_resource_rights	is__gfw_managed_forests	is__umd_tree_cover_gain	is__ifl_intact_forest_landscapes_2000	wri_tropical_tree_cover__decile	umd_global_land_cover__ipcc_class	umd_tree_cover_extent_2000__ha	umd_tree_cover_extent_2010__ha	area__ha	umd_tree_cover_gain__ha	whrc_aboveground_biomass_stock_2000__Mg	umd_tree_cover_loss__ha	whrc_aboveground_biomass_loss__Mg	whrc_aboveground_co2_emissions__Mg	whrc_aboveground_co2_stock_2000__Mg	gfw_full_extent_aboveground_gross_removals__Mg_CO2	gfw_full_extent_belowground_gross_removals__Mg_CO2	gfw_full_extent_net_flux__Mg_CO2e	gfw_full_extent_gross_emissions_CO2_only__Mg_CO2	gfw_full_extent_gross_emissions_non_CO2__Mg_CO2e	gfw_full_extent_gross_emissions__Mg_CO2e	gfw_soil_carbon_stocks_2000__Mg_C	gfw_full_extent_gross_removals__Mg_CO2	umd_tree_cover_loss_from_fires__ha	wri_tropical_tree_cover_extent__ha	gfw_aboveground_carbon_stocks_2000__Mg_C	gfw_belowground_carbon_stocks_2000__Mg_C	umd_tree_cover_density__threshold	tsc_tree_cover_loss_drivers__type	is__birdlife_alliance_for_zero_extinction_site	gfw_plantation__type	is__gmw_mangroves_1996	is__gmw_mangroves_2020	is__gfw_tiger_landscape	is__landmark_land_right	is__gfw_land_right	is__birdlife_key_biodiversity_area	is__gfw_mining	is__peatland	is__gfw_resource_right	is__gfw_managed_forest	is__umd_tree_cover_gain_2000-2012	umd_tree_cover_gain_2000-2012__ha	gfw_net_flux_co2e__Mg	gfw_gross_emissions_co2e_all_gases__Mg	gfw_gross_cumulative_aboveground_belowground_co2_removals__Mg	wdpa_protected_areas__iucn_cat	wdpa_protected_area__iucn_cat
+33177	Mulanje	MWI	Designated	2005	30	Shifting agriculture	false	true		false	false	false	false	true	false	false	false	false	false	false	false	false	false	-1		14.952659305429693	6.884108541322698	15.02667785951498	0.0	3158.2730592625508	14.952659305429693	3158.2730592625508	5442.757238795796	5442.757238795796	169.4592505162455	60.61315612472494	8042.287858127695	8272.360473078466	0.0	8272.360473078466	1120.8571413424913	230.07240664097048	0.0	0.0	1484.3883333693373	531.1805003608428	30	Shifting agriculture	true		false	false	false	false	false	true	false	false	false	false	false	0.0	8042.287858127695	8272.360473078466	230.07240664097048	Other Category	Other Category

src/test/scala/org/globalforestwatch/summarystats/annualupdate_minimal/AnnualUpdateMinimalSpec.scala

@@ -12,6 +12,8 @@ class AnnualUpdateMinimalSpec extends TestEnvironment with DataFrameComparer {
   def idn1_5GadmInputTsvPath = getClass.getResource("/idn1_5Gadm.tsv").toString()
   def idn1_5GadmExpectedOutputPath = getClass.getResource("/idn1_5Gadm-aum-output").toString()
   def wdpaInputTsvPath = getClass.getResource("/mulanje.tsv").toString()
+  def wdpaOutputPath: String = "output/mulanje"
+  def wdpaExpectedOutputPath = getClass.getResource("/wdpa-aum-output").toString()
 
 
   val csvOptions: Map[String, String] = Map(
@@ -69,35 +71,68 @@ class AnnualUpdateMinimalSpec extends TestEnvironment with DataFrameComparer {
   it("matches recorded output for GADM") {
     val gadmDF: DataFrame = AnnualUpdateMinimal(idn1_5GadmInputTsvPath, "gadm")
 
-    // Export and save results 
+    // Transform results to match expected output
     import spark.implicits._
-    val exportDF = gadmDF      
+    val unpackedDF = gadmDF      
       .transform(
         AnnualUpdateMinimalDF.unpackValues(
           List($"id.iso" as "iso", $"id.adm1" as "adm1", $"id.adm2" as "adm2")
         )
       )
-    val adm2DF: DataFrame = exportDF.transform(
+    val exportDF: DataFrame = unpackedDF.transform(
       AnnualUpdateMinimalDF.aggSummary(List("iso", "adm1", "adm2"))
     ) 
-    val top20Rows = adm2DF.limit(1)
-    top20Rows
+    val firstRow = exportDF.limit(1)
+
+    // Export and save results
+    firstRow
       .write
       .options(csvOptions)
       .csv("output/idntest")
 
     // Read expected results and compare
     val expectedDF = readAumResult(idn1_5GadmExpectedOutputPath).limit(1)
-    val top20RowsDF = readAumResult("output/idntest")
-    top20Rows.show()
+    val firstRowDF = readAumResult("output/idntest")
+    firstRowDF.show()
     expectedDF.show()
 
-    assertSmallDataFrameEquality(top20RowsDF, expectedDF, ignoreNullable = true)
+    assertSmallDataFrameEquality(firstRowDF, expectedDF, ignoreNullable = true)
 
   }
 
   it("matches recorded output for WDPA") {
-    //val wdpaDF: DataFrame = AnnualUpdateMinimal(wdpaInputTsvPath, "wdpa")
+    val wdpaDF: DataFrame = AnnualUpdateMinimal(wdpaInputTsvPath, "wdpa")
+
+    // Transform results to match expected output
+    import spark.implicits._
+    val exportDF = wdpaDF      
+      .transform(
+        AnnualUpdateMinimalDF.unpackValues(
+          List(
+            $"id.wdpaId" as "wdpa_protected_area__id",
+            $"id.name" as "wdpa_protected_area__name",
+            $"id.iucnCat" as "wdpa_protected_area__iucn_cat",
+            $"id.iso" as "wdpa_protected_area__iso",
+            $"id.status" as "wdpa_protected_area__status"
+          )
+        )
+      )
+
+    val firstRow = exportDF.limit(1)
+    // Export and save results
+    firstRow
+      .write
+      .options(csvOptions)
+      .csv("output/mulanje")
+
+    // Read expected results and compare
+    val expectedDF = readAumResult(idn1_5GadmExpectedOutputPath).limit(1)
+    val firstRowDF = readAumResult("output/mulanje")
+    firstRowDF.show()
+    expectedDF.show()
+
+    assertSmallDataFrameEquality(firstRowDF, expectedDF, ignoreNullable = true)
+
 
   }