add tests for ReconIntermediatePersist

Author: m-abulazm

src/databricks/labs/lakebridge/reconcile/recon_capture.py

@@ -54,8 +54,9 @@ def write_and_read_df_with_volumes(
 class ReconIntermediatePersist(AbstractReconIntermediatePersist):
     def __init__(self, spark: SparkSession, metadata_config: ReconcileMetadataConfig):
         self._spark = spark
+        self._metadata_config = metadata_config
         self._format = "delta" if self._is_databricks else "parquet"
-        self._base_dir = self._get_uc_volume_path(metadata_config) if self._is_databricks else tempfile.gettempdir()
+        self._base_dir = self._get_uc_volume_path if self._is_databricks else tempfile.gettempdir()
 
     @cached_property
     def _is_databricks(self) -> bool:
@@ -65,11 +66,14 @@ def _is_databricks(self) -> bool:
     def base_dir(self) -> Path:
         return Path(self._base_dir)
 
-    @classmethod
-    def _get_uc_volume_path(cls, metadata_config: ReconcileMetadataConfig):
-        catalog = metadata_config.catalog
-        schema = metadata_config.schema
-        return f"/Volumes/{catalog}/{schema}/{metadata_config.volume}"
+    @property
+    def _get_uc_volume_path(self):
+        return (
+            f"/Volumes/"
+            f"{self._metadata_config.catalog}/"
+            f"{self._metadata_config.schema}/"
+            f"{self._metadata_config.volume}"
+        )
 
     def _write_df_to_volumes(self, df: DataFrame, path: str) -> None:
         df.write.format(self._format).save(path)

tests/integration/reconcile/test_recon_capture.py

@@ -12,6 +12,7 @@
 from databricks.labs.lakebridge.reconcile.recon_capture import (
     ReconCapture,
     generate_final_reconcile_output,
+    ReconIntermediatePersist,
 )
 from databricks.labs.lakebridge.reconcile.recon_output_config import (
     DataReconcileOutput,
@@ -990,3 +991,55 @@ def test_apply_threshold_for_only_threshold_mismatch_with_true_absolute(
     remorph_recon_metrics_df = spark.sql(f"select * from {recon_metadata.schema}.metrics")
     row = remorph_recon_metrics_df.collect()[0]
     assert row.run_metrics.status is True
+
+
+class ReconIntermediatePersistUnderTest(ReconIntermediatePersist):
+    @property
+    def is_databricks(self) -> bool:
+        return self._is_databricks
+
+    @property
+    def format(self):
+        return self._format
+
+
+def test_is_databricks_false(mock_spark):
+    conf = ReconcileMetadataConfig()
+    persist = ReconIntermediatePersistUnderTest(mock_spark, conf)
+
+    assert persist.is_databricks is False
+
+
+def test_is_databricks_true(spark):
+    conf = ReconcileMetadataConfig()
+    persist = ReconIntermediatePersistUnderTest(spark, conf)
+
+    assert persist.is_databricks is True
+
+
+def test_dir_uses_tempfile(mock_spark):
+    conf = ReconcileMetadataConfig()
+    persist = ReconIntermediatePersistUnderTest(mock_spark, conf)
+
+    assert str(persist.base_dir).startswith("/tmp/")
+
+
+def test_format_uses_uc(spark):
+    conf = ReconcileMetadataConfig()
+    persist = ReconIntermediatePersistUnderTest(spark, conf)
+
+    assert str(persist.base_dir).startswith("/Volumes/")
+
+
+def test_format_uses_parquet(mock_spark):
+    conf = ReconcileMetadataConfig()
+    persist = ReconIntermediatePersistUnderTest(mock_spark, conf)
+
+    assert persist.format == "parquet"
+
+
+def test_format_uses_delta(spark):
+    conf = ReconcileMetadataConfig()
+    persist = ReconIntermediatePersistUnderTest(spark, conf)
+
+    assert persist.format == "delta"