Method test_export_dynamodb_json

tests/test_import_export.py:109–139

Export table to DYNAMODB_JSON format.

(self, dynamodb_client, populated_table)

        return name

    def test_export_dynamodb_json(self, dynamodb_client, populated_table):
        """Export table to DYNAMODB_JSON format."""
        table_name = populated_table
        desc = dynamodb_client.describe_table(TableName=table_name)
        table_arn = desc["Table"]["TableArn"]

        with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
            export_path = f.name

        try:
            resp = extenddb_request("ExportTableToPointInTime", {
                "TableArn": table_arn,
                "FilePath": export_path,
                "ExportFormat": "DYNAMODB_JSON",
            })
            export_desc = resp["ExportDescription"]
            assert export_desc["ExportStatus"] == "COMPLETED"
            assert export_desc["ItemCount"] == 5
            assert export_desc["ExportFormat"] == "DYNAMODB_JSON"

            # Verify file contents.
            with open(export_path) as f:
                lines = [line.strip() for line in f if line.strip()]
            assert len(lines) == 5

            for line in lines:
                obj = json.loads(line)
                assert "Item" in obj
                assert "pk" in obj["Item"]
        finally:
            os.unlink(export_path)

    def test_export_empty_table(self, dynamodb_client, unique_table_name, cleanup_table):
        """Export an empty table produces empty file."""