cubefs

1
// Copyright 2022 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package access
16

17
import (
18
	"bytes"
19
	"crypto/rand"
20
	"hash/crc32"
21
	"testing"
22

23
	"github.com/stretchr/testify/require"
24

25
	"github.com/cubefs/cubefs/blobstore/api/access"
26
)
27

28
func TestAccessStreamPutAtBase(t *testing.T) {
29
	ctx := ctxWithName("TestAccessStreamPutAtBase")
30
	sumChecker := func(size int, data []byte) {
31
		dataShards.clean()
32

33
		hasherMap := access.HasherMap{
34
			access.HashAlgCRC32: access.HashAlgCRC32.ToHasher(),
35
		}
36
		hashSumMap := make(access.HashSumMap, len(hasherMap))
37
		err := streamer.PutAt(ctx(), bytes.NewReader(data), clusterID, 1, 10000, int64(size), hasherMap)
38
		require.Nil(t, err)
39
		for alg, hasher := range hasherMap {
40
			hashSumMap[alg] = hasher.Sum(nil)
41
		}
42
		require.Equal(t, hashSumMap.GetSumVal(access.HashAlgCRC32), crc32.ChecksumIEEE(data[:size]))
43
	}
44
	// 0
45
	{
46
		size := 0
47
		err := streamer.PutAt(ctx(), newReader(size), clusterID, 1, 10000, int64(size), nil)
48
		require.NotNil(t, err)
49
	}
50
	// 1 byte
51
	{
52
		size := 1
53
		data := make([]byte, size)
54
		rand.Read(data)
55
		sumChecker(size, data)
56

57
		shardSize := getBufSizes(len(data)).ShardSize
58
		zeroShard := make([]byte, shardSize)
59
		shard1 := make([]byte, shardSize)
60
		copy(shard1, data)
61

62
		require.Equal(t, shard1, dataShards.get(1001, 10000))
63
		require.Equal(t, zeroShard, dataShards.get(1002, 10000))
64
		require.Equal(t, zeroShard, dataShards.get(1004, 10000))
65
		require.Equal(t, 0, len(dataShards.get(1005, 10000)))
66
		require.NotEqual(t, ([]byte)(nil), dataShards.get(1007, 10000))
67
	}
68
	// 4M per shard
69
	{
70
		size := (1 << 22) * 6
71
		data := make([]byte, size)
72
		rand.Read(data)
73
		sumChecker(size, data)
74

75
		require.Equal(t, 1<<22, len(dataShards.get(1001, 10000)))
76
	}
77
	// 4M + 1B, put 1B
78
	{
79
		size := (1 << 22) + 1
80
		data := make([]byte, size)
81
		rand.Read(data)
82
		sumChecker(1, data)
83

84
		shardSize := getBufSizes(1).ShardSize
85
		require.Equal(t, shardSize, len(dataShards.get(1001, 10000)))
86
	}
87
	// 4M, error
88
	{
89
		dataShards.clean()
90
		size := 1 << 22
91
		err := streamer.PutAt(ctx(), newReader(size), clusterID, 1, 10000, int64(size)+1, nil)
92
		require.NotNil(t, err)
93
		require.Equal(t, 0, len(dataShards.get(1001, 10000)))
94
	}
95

96
	dataShards.clean()
97
}
98