Method test_import_csv

tests/test_import_export.py:215–250

Import items from CSV file.

(self, dynamodb_client, unique_table_name, cleanup_table)

            os.unlink(source_path)

    def test_import_csv(self, dynamodb_client, unique_table_name, cleanup_table):
        """Import items from CSV file."""
        name = unique_table_name
        cleanup_table(name)

        csv_content = "pk,name,age\ncsv-1,Alice,30\ncsv-2,Bob,25\n"
        with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
            f.write(csv_content)
            source_path = f.name

        try:
            resp = extenddb_request("ImportTable", {
                "FileSource": {"Path": source_path},
                "InputFormat": "CSV",
                "TableCreationParameters": {
                    "TableName": name,
                    "AttributeDefinitions": [
                        {"AttributeName": "pk", "AttributeType": "S"},
                    ],
                    "KeySchema": [
                        {"AttributeName": "pk", "KeyType": "HASH"},
                    ],
                    "BillingMode": "PAY_PER_REQUEST",
                },
            })
            desc = resp["ImportTableDescription"]
            assert desc["ImportStatus"] == "COMPLETED"
            assert desc["ImportedItemCount"] == 2

            item = dynamodb_client.get_item(
                TableName=name, Key={"pk": {"S": "csv-1"}}
            )
            assert item["Item"]["name"]["S"] == "Alice"
            assert item["Item"]["age"]["S"] == "30"
        finally:
            os.unlink(source_path)

    def test_import_nonexistent_source(self, dynamodb_client, unique_table_name, cleanup_table):
        """Import from nonexistent path returns error."""