Increase coverage for dynamodb _write (#1893)

Author: LeonLuttenberger

tests/conftest.py

@@ -1,3 +1,4 @@
+import os
 from datetime import datetime
 from importlib import reload
 from types import ModuleType
@@ -378,6 +379,18 @@ def glue_data_quality_role(cloudformation_outputs):
     return cloudformation_outputs["GlueDataQualityRole"]
 
 
+@pytest.fixture(scope="function")
+def local_filename() -> Iterator[str]:
+    filename = os.path.join(".", f"{get_time_str_with_random_suffix()}.data")
+
+    yield filename
+
+    try:
+        os.remove(filename)
+    except OSError:
+        pass
+
+
 @pytest.fixture(scope="function", name="wr")
 def awswrangler_import() -> Iterator[ModuleType]:
     import awswrangler

tests/test_moto.py

@@ -1,5 +1,6 @@
 import logging
 import os
+from typing import Any
 from unittest import mock
 from unittest.mock import ANY
 
@@ -521,6 +522,31 @@ def test_dynamodb_basic_usage(moto_dynamodb):
     assert table.item_count == 0
 
 
+@pytest.mark.parametrize("format", ["csv", "json"])
+def test_dynamodb_put_from_file(moto_dynamodb: Any, local_filename: str, format: str) -> None:
+    table_name = "table"
+    df = pd.DataFrame({"key": [1, 2], "value": ["foo", "boo"]})
+
+    if format == "csv":
+        df.to_csv(local_filename, index=False)
+        wr.dynamodb.put_csv(
+            path=local_filename,
+            table_name=table_name,
+        )
+    elif format == "json":
+        df.to_json(local_filename, orient="records")
+        wr.dynamodb.put_json(
+            path=local_filename,
+            table_name=table_name,
+        )
+    else:
+        raise RuntimeError(f"Unknown format {format}")
+
+    df2 = wr.dynamodb.read_partiql_query(query="SELECT * FROM table")
+
+    assert df.shape == df2.shape
+
+
 def test_dynamodb_partiql(moto_dynamodb):
     table_name = "table"
     items = [{"key": 1}, {"key": 2, "my_value": "Hello"}]