Minor improvements to Restore Python API implementation and tests

* Add args check
* Add more tests (negative tests)

Existing UTs

Closes #968

GitOrigin-RevId: dee05a116a1a685425a09b34ebb643d86e70486b

Author: vkorukanti

python/delta/tables.py

@@ -553,6 +553,7 @@ def restoreToVersion(self, version: int) -> DataFrame:
         :rtype: pyspark.sql.DataFrame
         """
 
+        DeltaTable._verify_type_int(version, "version")
         return DataFrame(
             self._jdt.restoreToVersion(version),
             getattr(self._spark, "_wrapped", self._spark)  # type: ignore[attr-defined]
@@ -574,11 +575,22 @@ def restoreToTimestamp(self, timestamp: str) -> DataFrame:
         :rtype: pyspark.sql.DataFrame
         """
 
+        DeltaTable._verify_type_str(timestamp, "timestamp")
         return DataFrame(
             self._jdt.restoreToTimestamp(timestamp),
             getattr(self._spark, "_wrapped", self._spark)  # type: ignore[attr-defined]
         )
 
+    @staticmethod  # type: ignore[arg-type]
+    def _verify_type_str(variable: str, name: str) -> None:
+        if not isinstance(variable, str) or variable is None:
+            raise ValueError("%s needs to be a string but got '%s'." % (name, type(variable)))
+
+    @staticmethod  # type: ignore[arg-type]
+    def _verify_type_int(variable: int, name: str) -> None:
+        if not isinstance(variable, int) or variable is None:
+            raise ValueError("%s needs to be an int but got '%s'." % (name, type(variable)))
+
     @staticmethod
     def _dict_to_jmap(
         sparkSession: SparkSession,

python/delta/tests/test_deltatable.py

@@ -16,7 +16,7 @@
 
 import unittest
 import os
-from typing import List, Set, Dict, Optional, Any, Union, Tuple
+from typing import List, Set, Dict, Optional, Any, Callable, Union, Tuple
 
 from pyspark.sql import DataFrame, Row
 from pyspark.sql.column import _to_seq  # type: ignore[attr-defined]
@@ -827,6 +827,29 @@ def test_restore_to_timestamp(self) -> None:
         restored = DeltaTable.forPath(self.spark, self.tempFile).toDF()
         self.__checkAnswer(restored, [Row(key='a', value=1), Row(key='b', value=2)])
 
+        # we cannot test the actual working of restore to timestamp here but we can make sure
+        # that the api is being called at least
+        def runRestore() -> None:
+            DeltaTable.forPath(self.spark, self.tempFile).restoreToTimestamp('05/04/1999')
+        self.__intercept(runRestore, "The provided timestamp ('05/04/1999') "
+                                     "cannot be converted to a valid timestamp")
+
+    def test_restore_invalid_inputs(self) -> None:
+        df = self.spark.createDataFrame([('a', 1), ('b', 2), ('c', 3)], ["key", "value"])
+        df.write.format("delta").save(self.tempFile)
+
+        dt = DeltaTable.forPath(self.spark, self.tempFile)
+
+        def runRestoreToTimestamp() -> None:
+            dt.restoreToTimestamp(12342323232)  # type: ignore[arg-type]
+        self.__intercept(runRestoreToTimestamp,
+                         "timestamp needs to be a string but got '<class 'int'>'")
+
+        def runRestoreToVersion() -> None:
+            dt.restoreToVersion("0")  # type: ignore[arg-type]
+        self.__intercept(runRestoreToVersion,
+                         "version needs to be an int but got '<class 'str'>'")
+
     def __checkAnswer(self, df: DataFrame,
                       expectedAnswer: List[Any],
                       schema: Union[StructType, List[str]] = ["key", "value"]) -> None:
@@ -870,6 +893,15 @@ def __createFile(self, fileName: str, content: Any) -> None:
     def __checkFileExists(self, fileName: str) -> bool:
         return os.path.exists(os.path.join(self.tempFile, fileName))
 
+    def __intercept(self, func: Callable[[], None], exceptionMsg: str) -> None:
+        seenTheRightException = False
+        try:
+            func()
+        except Exception as e:
+            if exceptionMsg in str(e):
+                seenTheRightException = True
+        assert seenTheRightException, ("Did not catch expected Exception:" + exceptionMsg)
+
 
 if __name__ == "__main__":
     try: