Method test_import_dynamodb_json

tests/test_import_export.py:173–213

Import items from DYNAMODB_JSON file.

(self, dynamodb_client, unique_table_name, cleanup_table)

    """Tests for ImportTable."""

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

        items = [
            {"Item": {"pk": {"S": f"imp-{i}"}, "val": {"N": str(i)}}}
            for i in range(3)
        ]
        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
            for item in items:
                f.write(json.dumps(item) + "\n")
            source_path = f.name

        try:
            resp = extenddb_request("ImportTable", {
                "FileSource": {"Path": source_path},
                "InputFormat": "DYNAMODB_JSON",
                "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"] == 3
            assert desc["ErrorCount"] == 0

            for i in range(3):
                item = dynamodb_client.get_item(
                    TableName=name, Key={"pk": {"S": f"imp-{i}"}}
                )
                assert item["Item"]["val"]["N"] == str(i)
        finally:
            os.unlink(source_path)

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