4566704
455360bff0
- 改动：init/check 接受可选 targetDir（相对 CloudDir，SanitizeDir 校验），merge 物理落盘 CloudDir/<targetDir>/ - 秒传跨目录 copy-on-hit：库把旧物理文件复制进目标目录，OnInstantHit 携带新路径；源缺失回退正常上传 - 断点续传按目录隔离：同 MD5 在途任务目录不同则另起新任务 - TaskInfo 新增 TargetDir（宿主须持久化）；Repository 12 方法接口零变更，向后兼容 v1.0.x - 新增 dir.go（SanitizeDir/路径拼装反解/copyFile）与 service_test.go（10 个用例：双模式秒传/续传隔离/merge 回归） - 原因：上传位置原本固定在 CloudDir 根，宿主无法实现网盘式选目录上传
419 lines
12 KiB
Go
419 lines
12 KiB
Go
package fileupload
|
||
|
||
import (
|
||
"crypto/md5"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"strings"
|
||
"testing"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
)
|
||
|
||
// ── 内存版 Repository（测试桩） ──────────────────────────────────────
|
||
|
||
type memRepo struct {
|
||
nextID int64
|
||
tasks map[int64]*TaskInfo
|
||
byUID map[string]int64
|
||
chunks map[string]map[int32]struct{}
|
||
}
|
||
|
||
func newMemRepo() *memRepo {
|
||
return &memRepo{
|
||
tasks: map[int64]*TaskInfo{},
|
||
byUID: map[string]int64{},
|
||
chunks: map[string]map[int32]struct{}{},
|
||
}
|
||
}
|
||
|
||
func (r *memRepo) CreateTask(t *TaskInfo) error {
|
||
r.nextID++
|
||
cp := *t
|
||
cp.Id = r.nextID
|
||
r.tasks[cp.Id] = &cp
|
||
r.byUID[cp.UploadId] = cp.Id
|
||
return nil
|
||
}
|
||
|
||
func (r *memRepo) GetTaskByUploadID(uploadID string) (*TaskInfo, error) {
|
||
id, ok := r.byUID[uploadID]
|
||
if !ok {
|
||
return nil, nil
|
||
}
|
||
cp := *r.tasks[id]
|
||
return &cp, nil
|
||
}
|
||
|
||
func (r *memRepo) FindDoneByMD5(md5 string) (*TaskInfo, error) {
|
||
var best *TaskInfo
|
||
for _, t := range r.tasks {
|
||
if t.FileMd5 == md5 && t.Status == StatusDone {
|
||
if best == nil || t.Id > best.Id {
|
||
best = t
|
||
}
|
||
}
|
||
}
|
||
if best == nil {
|
||
return nil, nil
|
||
}
|
||
cp := *best
|
||
return &cp, nil
|
||
}
|
||
|
||
func (r *memRepo) FindUploadingByMD5(md5 string) (*TaskInfo, error) {
|
||
var best *TaskInfo
|
||
for _, t := range r.tasks {
|
||
if t.FileMd5 == md5 && t.Status == StatusUploading {
|
||
if best == nil || t.Id > best.Id {
|
||
best = t
|
||
}
|
||
}
|
||
}
|
||
if best == nil {
|
||
return nil, nil
|
||
}
|
||
cp := *best
|
||
return &cp, nil
|
||
}
|
||
|
||
func (r *memRepo) UpdateTaskMerged(id int64, storagePath string) error {
|
||
r.tasks[id].Status = StatusDone
|
||
r.tasks[id].StoragePath = storagePath
|
||
return nil
|
||
}
|
||
|
||
func (r *memRepo) SetTaskStatus(id int64, status string) error {
|
||
r.tasks[id].Status = status
|
||
return nil
|
||
}
|
||
|
||
func (r *memRepo) CountChunks(uploadID string) (int, error) {
|
||
return len(r.chunks[uploadID]), nil
|
||
}
|
||
|
||
func (r *memRepo) ChunkExists(uploadID string, index int32) (bool, error) {
|
||
_, ok := r.chunks[uploadID][index]
|
||
return ok, nil
|
||
}
|
||
|
||
func (r *memRepo) CreateChunk(c *ChunkInfo) error {
|
||
if r.chunks[c.UploadId] == nil {
|
||
r.chunks[c.UploadId] = map[int32]struct{}{}
|
||
}
|
||
r.chunks[c.UploadId][c.ChunkIndex] = struct{}{}
|
||
return nil
|
||
}
|
||
|
||
func (r *memRepo) ListChunkIndexes(uploadID string) ([]int32, error) {
|
||
var out []int32
|
||
for i := range r.chunks[uploadID] {
|
||
out = append(out, i)
|
||
}
|
||
sort.Slice(out, func(a, b int) bool { return out[a] < out[b] })
|
||
return out, nil
|
||
}
|
||
|
||
func (r *memRepo) DeleteChunks(uploadID string) error {
|
||
delete(r.chunks, uploadID)
|
||
return nil
|
||
}
|
||
|
||
func (r *memRepo) DeleteTaskByUploadID(uploadID string) error {
|
||
id, ok := r.byUID[uploadID]
|
||
if !ok {
|
||
return nil
|
||
}
|
||
delete(r.tasks, id)
|
||
delete(r.byUID, uploadID)
|
||
return nil
|
||
}
|
||
|
||
// ── 测试工具 ─────────────────────────────────────────────────────────
|
||
|
||
func newTestSvc(t *testing.T) (*service, *memRepo, string) {
|
||
t.Helper()
|
||
repo := newMemRepo()
|
||
cloud := t.TempDir()
|
||
svc := newService(Options{
|
||
ChunkRoot: t.TempDir(),
|
||
CloudDir: cloud,
|
||
WebPathPrefix: "cloud",
|
||
URLPrefix: "/uploads",
|
||
Envelope: func(c *gin.Context, resp any, err error) {},
|
||
}, repo)
|
||
return svc, repo, cloud
|
||
}
|
||
|
||
func mustMD5(b []byte) string { return fmt.Sprintf("%x", md5.Sum(b)) }
|
||
|
||
func seedDoneTask(t *testing.T, svc *service, repo *memRepo, md5Str, targetDir string) *TaskInfo {
|
||
t.Helper()
|
||
task := TaskInfo{
|
||
UploadId: newUploadID(),
|
||
FileName: "old.bin",
|
||
FileMd5: md5Str,
|
||
Status: StatusDone,
|
||
TargetDir: targetDir,
|
||
StoragePath: buildStoragePath("cloud", targetDir, "old.bin"),
|
||
}
|
||
if err := repo.CreateTask(&task); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// 物理文件落盘
|
||
phys := svc.physicalPath(task.StoragePath)
|
||
if err := os.MkdirAll(filepath.Dir(phys), 0o766); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := os.WriteFile(phys, []byte("OLD-CONTENT"), 0o644); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return &task
|
||
}
|
||
|
||
// ── SanitizeDir ─────────────────────────────────────────────────────
|
||
|
||
func TestSanitizeDir(t *testing.T) {
|
||
ok := map[string]string{
|
||
"": "", // 根目录（固定模式）
|
||
"a": "a",
|
||
"a/": "a", // 尾随 / 归一
|
||
"a/b": "a/b",
|
||
`a\b`: "a/b",
|
||
"/abs/": "abs", // 前导 / 剥离，按相对处理
|
||
" docs / reports ": "docs / reports",
|
||
"云盘/资料": "云盘/资料",
|
||
"a\x01b": "ab", // 控制字符剔除
|
||
}
|
||
for in, want := range ok {
|
||
got, err := SanitizeDir(in)
|
||
if err != nil || got != want {
|
||
t.Errorf("SanitizeDir(%q) = %q, %v; want %q", in, got, err, want)
|
||
}
|
||
}
|
||
bad := []string{
|
||
"..", "a/../b", "a/./b", "a//b",
|
||
strings.Repeat("a/", 20) + "b", // 超深
|
||
strings.Repeat("a", MaxDirLength+1), // 超长
|
||
"a:b", "a*b", `a"b`, "a<b", "a>b", "a|b", // Windows 非法字符
|
||
}
|
||
for _, in := range bad {
|
||
if got, err := SanitizeDir(in); err == nil {
|
||
t.Errorf("SanitizeDir(%q) = %q, want error", in, got)
|
||
}
|
||
}
|
||
}
|
||
|
||
// ── init：秒传双模式 ──────────────────────────────────────────────────
|
||
|
||
func initReq(name, md5Str string, targetDir string) *InitReq {
|
||
return &InitReq{
|
||
FileName: name,
|
||
FileSize: 10,
|
||
FileMD5: md5Str,
|
||
ChunkSize: 10,
|
||
TotalChunk: 1,
|
||
TargetDir: targetDir,
|
||
}
|
||
}
|
||
|
||
func TestInitInstantHitFixedMode(t *testing.T) {
|
||
svc, repo, cloud := newTestSvc(t)
|
||
sum := mustMD5([]byte("OLD-CONTENT"))
|
||
seedDoneTask(t, svc, repo, sum, "")
|
||
|
||
resp, err := svc.init(initReq("old.bin", sum, ""), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if resp.(map[string]any)["skipUpload"] != true {
|
||
t.Fatalf("固定模式应秒传: %v", resp)
|
||
}
|
||
if got := resp.(map[string]any)["fileUrl"]; got != "/uploads/cloud/old.bin" {
|
||
t.Fatalf("fileUrl = %v, 与 v1.0.x 行为不一致", got)
|
||
}
|
||
// 不应产生复制：cloud 根下仍只有 old.bin
|
||
entries, _ := os.ReadDir(cloud)
|
||
if len(entries) != 1 {
|
||
t.Fatalf("固定模式秒传不应复制文件, cloud 根条目数 = %d", len(entries))
|
||
}
|
||
}
|
||
|
||
func TestInitInstantHitCrossDirCopy(t *testing.T) {
|
||
svc, repo, cloud := newTestSvc(t)
|
||
sum := mustMD5([]byte("OLD-CONTENT"))
|
||
seedDoneTask(t, svc, repo, sum, "")
|
||
|
||
var hit TaskInfo
|
||
svc.opts.OnInstantHit = func(ti TaskInfo) error { hit = ti; return nil }
|
||
|
||
resp, err := svc.init(initReq("old.bin", sum, "docs/reports"), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got := resp.(map[string]any)["fileUrl"]; got != "/uploads/cloud/docs/reports/old.bin" {
|
||
t.Fatalf("fileUrl = %v", got)
|
||
}
|
||
if hit.StoragePath != "cloud/docs/reports/old.bin" || hit.TargetDir != "docs/reports" {
|
||
t.Fatalf("OnInstantHit 携带路径错误: %+v", hit)
|
||
}
|
||
// 物理复制存在且内容一致
|
||
data, err := os.ReadFile(filepath.Join(cloud, "docs", "reports", "old.bin"))
|
||
if err != nil || string(data) != "OLD-CONTENT" {
|
||
t.Fatalf("目标目录物理副本缺失或不一致: %v, %q", err, data)
|
||
}
|
||
}
|
||
|
||
func TestInitInstantHitFallbackWhenSourceMissing(t *testing.T) {
|
||
svc, repo, _ := newTestSvc(t)
|
||
sum := mustMD5([]byte("OLD-CONTENT"))
|
||
done := seedDoneTask(t, svc, repo, sum, "")
|
||
if err := os.Remove(svc.physicalPath(done.StoragePath)); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
resp, err := svc.init(initReq("old.bin", sum, "docs"), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
m := resp.(map[string]any)
|
||
if m["skipUpload"] != false || m["uploadId"] == "" {
|
||
t.Fatalf("旧物理文件缺失应回退正常上传: %v", m)
|
||
}
|
||
}
|
||
|
||
// ── init：断点续传目录隔离 ────────────────────────────────────────────
|
||
|
||
func TestInitResumeDirIsolation(t *testing.T) {
|
||
svc, repo, _ := newTestSvc(t)
|
||
sum := mustMD5([]byte("OLD-CONTENT"))
|
||
|
||
first, err := svc.init(initReq("old.bin", sum, "a"), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
oldUID := first.(map[string]any)["uploadId"].(string)
|
||
|
||
same, err := svc.init(initReq("old.bin", sum, "a"), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got := same.(map[string]any)["uploadId"]; got != oldUID {
|
||
t.Fatalf("同目录续传应复用任务: %v != %v", got, oldUID)
|
||
}
|
||
|
||
diff, err := svc.init(initReq("old.bin", sum, "b"), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
newUID := diff.(map[string]any)["uploadId"].(string)
|
||
if newUID == oldUID {
|
||
t.Fatal("目录不同应另起新任务")
|
||
}
|
||
if task, _ := repo.GetTaskByUploadID(newUID); task.TargetDir != "b" {
|
||
t.Fatalf("新任务 TargetDir 应为 b: %+v", task)
|
||
}
|
||
|
||
fixed, err := svc.init(initReq("old.bin", sum, ""), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got := fixed.(map[string]any)["uploadId"]; got == oldUID {
|
||
t.Fatal("固定模式与目录 a 的在途任务也应隔离")
|
||
}
|
||
}
|
||
|
||
// ── merge：落盘目标目录 ──────────────────────────────────────────────
|
||
|
||
func mergeTask(t *testing.T, svc *service, repo *memRepo, targetDir, content string) string {
|
||
t.Helper()
|
||
sum := mustMD5([]byte(content))
|
||
resp, err := svc.init(initReq("new.bin", sum, targetDir), "op", 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
uid := resp.(map[string]any)["uploadId"].(string)
|
||
|
||
dir := svc.chunkDir(uid)
|
||
if err := os.MkdirAll(dir, 0o766); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := os.WriteFile(filepath.Join(dir, "0"), []byte(content), 0o644); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := repo.CreateChunk(&ChunkInfo{UploadId: uid, ChunkIndex: 0}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
out, err := svc.merge(uid)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got := out.(map[string]any)["fileUrl"]; got != "/uploads/"+repo.tasks[repo.byUID[uid]].StoragePath {
|
||
t.Fatalf("fileUrl 与 storagePath 不对应: %v vs %s", got, repo.tasks[repo.byUID[uid]].StoragePath)
|
||
}
|
||
return repo.tasks[repo.byUID[uid]].StoragePath
|
||
}
|
||
|
||
func TestMergeIntoTargetDir(t *testing.T) {
|
||
svc, repo, cloud := newTestSvc(t)
|
||
sp := mergeTask(t, svc, repo, "docs/reports", "HELLO")
|
||
|
||
if sp != "cloud/docs/reports/"+filepath.Base(sp) {
|
||
t.Fatalf("storagePath 形态错误: %s", sp)
|
||
}
|
||
data, err := os.ReadFile(filepath.Join(cloud, "docs", "reports", filepath.Base(sp)))
|
||
if err != nil || string(data) != "HELLO" {
|
||
t.Fatalf("目标目录物理文件缺失或不一致: %v", err)
|
||
}
|
||
}
|
||
|
||
func TestMergeFixedModeRegression(t *testing.T) {
|
||
svc, repo, cloud := newTestSvc(t)
|
||
sp := mergeTask(t, svc, repo, "", "HELLO")
|
||
|
||
// 与 v1.0.x 逐字节一致：cloud/<nanots>_<name>，物理在 cloud 根
|
||
if !strings.HasPrefix(sp, "cloud/") || strings.Contains(strings.TrimPrefix(sp, "cloud/"), "/") {
|
||
t.Fatalf("固定模式 storagePath 应平铺在 cloud 根: %s", sp)
|
||
}
|
||
if _, err := os.Stat(filepath.Join(cloud, filepath.Base(sp))); err != nil {
|
||
t.Fatalf("固定模式物理文件应在 cloud 根: %v", err)
|
||
}
|
||
}
|
||
|
||
// ── check：目录过滤 ──────────────────────────────────────────────────
|
||
|
||
func TestCheckUploadingDirFilter(t *testing.T) {
|
||
svc, repo, _ := newTestSvc(t)
|
||
sum := mustMD5([]byte("OLD-CONTENT"))
|
||
if _, err := svc.init(initReq("old.bin", sum, "a"), "op", 1); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
uploading, err := svc.check(sum, "a")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if uploading.(map[string]any)["status"] != StatusUploading {
|
||
t.Fatalf("同目录 check 应报 uploading: %v", uploading)
|
||
}
|
||
|
||
noneResp, err := svc.check(sum, "b")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if noneResp.(map[string]any)["status"] != "none" {
|
||
t.Fatalf("异目录 check 应报 none: %v", noneResp)
|
||
}
|
||
|
||
// done 命中不受目录影响（init 会 copy-on-hit）
|
||
seedDoneTask(t, svc, repo, mustMD5([]byte("OTHER")), "")
|
||
doneResp, err := svc.check(mustMD5([]byte("OTHER")), "zzz")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if doneResp.(map[string]any)["status"] != StatusDone {
|
||
t.Fatalf("done 命中应报 done: %v", doneResp)
|
||
}
|
||
}
|