feat: Add unit tests for GCP support (#162)

## Summary

- Adding unit tests for #147 
- https://app.asana.com/0/1208949807589885/1208960391734329/f

## Checklist
- [x] Added Unit Tests
- [ ] Covered by existing CI
- [ ] Integration tested
- [ ] Documentation update
<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Added a new test class to validate GCS format partitioning
functionality.

- **Refactor**
	- Updated package structure for test files.
	- Removed specific imports in test files.

- **Chores**
	- Added an import for a BigQuery table in the format handling.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

<!-- av pr metadata
This information is embedded by the av CLI when creating PRs to track
the status of stacks when using Aviator. Please do not delete or edit
this section of the PR.
```
{"parent":"main","parentHead":"","trunk":"main"}
```
-->

---------

Co-authored-by: Thomas Chow <[email protected]>

Author: tchow-zlai

cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/BigQueryFormat.scala

@@ -11,7 +11,8 @@ import com.google.cloud.bigquery.Table
 import com.google.cloud.bigquery.connector.common.BigQueryUtil
 import com.google.cloud.spark.bigquery.repackaged.com.google.cloud.bigquery.TableId
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.functions.{col, to_date}
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.functions.to_date
 
 import scala.collection.JavaConverters._
 

cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/test/BigQueryCatalogTest.scala renamed to cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/BigQueryCatalogTest.scala

@@ -1,7 +1,5 @@
-package ai.chronon.integrations.cloud_gcp.test
+package ai.chronon.integrations.cloud_gcp
 
-import ai.chronon.integrations.cloud_gcp.BQuery
-import ai.chronon.integrations.cloud_gcp.GcpFormatProvider
 import ai.chronon.spark.SparkSessionBuilder
 import ai.chronon.spark.TableUtils
 import com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS

cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/test/DataprocSubmitterTest.scala renamed to cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala

@@ -1,7 +1,5 @@
-package ai.chronon.integrations.cloud_gcp.test
+package ai.chronon.integrations.cloud_gcp
 
-import ai.chronon.integrations.cloud_gcp.DataprocSubmitter
-import ai.chronon.integrations.cloud_gcp.SubmitterConf
 import com.google.api.gax.rpc.UnaryCallable
 import com.google.cloud.dataproc.v1._
 import com.google.cloud.dataproc.v1.stub.JobControllerStub

cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/GCSFormatTest.scala

@@ -0,0 +1,92 @@
+package ai.chronon.integrations.cloud_gcp
+
+import ai.chronon.spark.SparkSessionBuilder
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.StructType
+import org.junit.Assert.assertEquals
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.nio.file.Files
+
+class GCSFormatTest extends AnyFunSuite {
+
+  lazy val spark: SparkSession = SparkSessionBuilder.build(
+    "BigQuerySparkTest",
+    local = true
+  )
+
+  test("partitions method should return correctly parsed partitions as maps") {
+
+    val testData = List(
+      ("20241223", "b", "c"),
+      ("20241224", "e", "f"),
+      ("20241225", "h", "i")
+    )
+
+    val dir = Files.createTempDirectory("spark-test-output").toFile
+    dir.deleteOnExit()
+
+    val df = spark.createDataFrame(testData).toDF("ds", "first", "second")
+    df.write.partitionBy("ds").format("parquet").mode(SaveMode.Overwrite).save(dir.getAbsolutePath)
+    val gcsFormat = GCS(project = "test-project", sourceUri = dir.getAbsolutePath, fileFormat = "parquet")
+    val partitions = gcsFormat.partitions("unused_table")(spark)
+
+    assertEquals(Set(Map("ds" -> "20241223"), Map("ds" -> "20241224"), Map("ds" -> "20241225")), partitions.toSet)
+
+  }
+
+  test("partitions method should handle empty partitions gracefully") {
+
+    val testData = List(
+      ("20241223", "b", "c"),
+      ("20241224", "e", "f"),
+      ("20241225", "h", "i")
+    )
+
+    val dir = Files.createTempDirectory("spark-test-output").toFile
+    dir.deleteOnExit()
+
+    val df = spark.createDataFrame(testData).toDF("ds", "first", "second")
+    df.write.format("parquet").mode(SaveMode.Overwrite).save(dir.getAbsolutePath)
+    val gcsFormat = GCS(project = "test-project", sourceUri = dir.getAbsolutePath, fileFormat = "parquet")
+    val partitions = gcsFormat.partitions("unused_table")(spark)
+
+    assertEquals(Set.empty, partitions.toSet)
+
+  }
+
+  test("partitions method should handle date types") {
+    val testData = List(
+      Row("2024-12-23", "b", "c"),
+      Row("2024-12-24", "e", "f"),
+      Row("2024-12-25", "h", "i")
+    )
+
+    val dir = Files.createTempDirectory("spark-test-output").toFile
+    dir.deleteOnExit()
+
+    val schema = StructType(
+      Seq(
+        StructField("ds", StringType, nullable = true),
+        StructField("first", StringType, nullable = true),
+        StructField("second", StringType, nullable = true)
+      ))
+
+    val df =
+      spark
+        .createDataFrame(spark.sparkContext.parallelize(testData), schema)
+        .toDF("ds", "first", "second")
+        .select(to_date(col("ds"), "yyyy-MM-dd").as("ds"), col("first"), col("second"))
+    df.write.format("parquet").partitionBy("ds").mode(SaveMode.Overwrite).save(dir.getAbsolutePath)
+    val gcsFormat = GCS(project = "test-project", sourceUri = dir.getAbsolutePath, fileFormat = "parquet")
+    val partitions = gcsFormat.partitions("unused_table")(spark)
+
+    assertEquals(Set(Map("ds" -> "2024-12-23"), Map("ds" -> "2024-12-24"), Map("ds" -> "2024-12-25")), partitions.toSet)
+
+  }
+}