Skip to content
Merged
Show file tree
Hide file tree
Changes from 13 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 5 additions & 26 deletions models/migrations/base/tests.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@ import (
"context"
"fmt"
"os"
"path"
"path/filepath"
"runtime"
"testing"
Expand All @@ -35,27 +34,7 @@ func PrepareTestEnv(t *testing.T, skip int, syncModels ...any) (*xorm.Engine, fu
ourSkip := 2
ourSkip += skip
deferFn := testlogger.PrintCurrentTest(t, ourSkip)
assert.NoError(t, os.RemoveAll(setting.RepoRootPath))
assert.NoError(t, unittest.CopyDir(path.Join(filepath.Dir(setting.AppPath), "tests/gitea-repositories-meta"), setting.RepoRootPath))
ownerDirs, err := os.ReadDir(setting.RepoRootPath)
if err != nil {
assert.NoError(t, err, "unable to read the new repo root: %v\n", err)
}
for _, ownerDir := range ownerDirs {
if !ownerDir.Type().IsDir() {
continue
}
repoDirs, err := os.ReadDir(filepath.Join(setting.RepoRootPath, ownerDir.Name()))
if err != nil {
assert.NoError(t, err, "unable to read the new repo root: %v\n", err)
}
for _, repoDir := range repoDirs {
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "objects", "pack"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "objects", "info"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "refs", "heads"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "refs", "tag"), 0o755)
}
}
assert.NoError(t, unittest.SyncDirs(filepath.Join(filepath.Dir(setting.AppPath), "tests/gitea-repositories-meta"), setting.RepoRootPath))

if err := deleteDB(); err != nil {
t.Errorf("unable to reset database: %v", err)
Expand Down Expand Up @@ -123,20 +102,20 @@ func MainTest(m *testing.M) {
if runtime.GOOS == "windows" {
giteaBinary += ".exe"
}
setting.AppPath = path.Join(giteaRoot, giteaBinary)
setting.AppPath = filepath.Join(giteaRoot, giteaBinary)
if _, err := os.Stat(setting.AppPath); err != nil {
fmt.Printf("Could not find gitea binary at %s\n", setting.AppPath)
os.Exit(1)
}

giteaConf := os.Getenv("GITEA_CONF")
if giteaConf == "" {
giteaConf = path.Join(filepath.Dir(setting.AppPath), "tests/sqlite.ini")
giteaConf = filepath.Join(filepath.Dir(setting.AppPath), "tests/sqlite.ini")
fmt.Printf("Environment variable $GITEA_CONF not set - defaulting to %s\n", giteaConf)
}

if !path.IsAbs(giteaConf) {
setting.CustomConf = path.Join(giteaRoot, giteaConf)
if !filepath.IsAbs(giteaConf) {
setting.CustomConf = filepath.Join(giteaRoot, giteaConf)
} else {
setting.CustomConf = giteaConf
}
Expand Down
84 changes: 44 additions & 40 deletions models/unittest/fscopy.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,8 @@
package unittest

import (
"errors"
"io"
"os"
"path"
"path/filepath"
"strings"

"code.gitea.io/gitea/modules/util"
Expand All @@ -32,67 +30,73 @@ func Copy(src, dest string) error {
return os.Symlink(target, dest)
}

sr, err := os.Open(src)
return util.CopyFile(src, dest)
}

// Sync synchronizes the two files. This is skipped if both files
// exist and the size, modtime, and mode match.
func Sync(srcPath, destPath string) error {
dest, err := os.Stat(destPath)
if err != nil {
if os.IsNotExist(err) {
return Copy(srcPath, destPath)
}
return err
}
defer sr.Close()

dw, err := os.Create(dest)
src, err := os.Stat(srcPath)
if err != nil {
return err
}
defer dw.Close()

if _, err = io.Copy(dw, sr); err != nil {
return err
if src.Size() == dest.Size() &&
src.ModTime() == dest.ModTime() &&
src.Mode() == dest.Mode() {
return nil
}

// Set back file information.
if err = os.Chtimes(dest, si.ModTime(), si.ModTime()); err != nil {
return err
}
return os.Chmod(dest, si.Mode())
return Copy(srcPath, destPath)
}

// CopyDir copy files recursively from source to target directory.
//
// The filter accepts a function that process the path info.
// and should return true for need to filter.
//
// SyncDirs synchronizes files recursively from source to target directory.
// It returns error when error occurs in underlying functions.
func CopyDir(srcPath, destPath string, filters ...func(filePath string) bool) error {
// Check if target directory exists.
if _, err := os.Stat(destPath); !errors.Is(err, os.ErrNotExist) {
return util.NewAlreadyExistErrorf("file or directory already exists: %s", destPath)
}

func SyncDirs(srcPath, destPath string) error {
err := os.MkdirAll(destPath, os.ModePerm)
if err != nil {
return err
}

// Gather directory info.
infos, err := util.StatDir(srcPath, true)
// find and delete all untracked files
destFiles, err := util.StatDir(destPath, true)
if err != nil {
return err
}

var filter func(filePath string) bool
if len(filters) > 0 {
filter = filters[0]
}

for _, info := range infos {
if filter != nil && filter(info) {
continue
for _, destFile := range destFiles {
destFilePath := filepath.Join(destPath, destFile)
if _, err = os.Stat(filepath.Join(srcPath, destFile)); err != nil {
if os.IsNotExist(err) {
// if src file does not exist, remove dest file
if err = os.RemoveAll(destFilePath); err != nil {
return err
}
} else {
return err
}
}
}

curPath := path.Join(destPath, info)
if strings.HasSuffix(info, "/") {
err = os.MkdirAll(curPath, os.ModePerm)
// sync src files to dest
srcFiles, err := util.StatDir(srcPath, true)
if err != nil {
return err
}
for _, srcFile := range srcFiles {
destFilePath := filepath.Join(destPath, srcFile)
// util.StatDir appends a slash to the directory name
if strings.HasSuffix(srcFile, "/") {
err = os.MkdirAll(destFilePath, os.ModePerm)
} else {
err = Copy(path.Join(srcPath, info), curPath)
err = Sync(filepath.Join(srcPath, srcFile), destFilePath)
}
if err != nil {
return err
Expand Down
45 changes: 3 additions & 42 deletions models/unittest/testdb.go
Original file line number Diff line number Diff line change
Expand Up @@ -164,35 +164,13 @@ func MainTest(m *testing.M, testOpts ...*TestOptions) {
if err = storage.Init(); err != nil {
fatalTestError("storage.Init: %v\n", err)
}
if err = util.RemoveAll(repoRootPath); err != nil {
fatalTestError("util.RemoveAll: %v\n", err)
}
if err = CopyDir(filepath.Join(giteaRoot, "tests", "gitea-repositories-meta"), setting.RepoRootPath); err != nil {
fatalTestError("util.CopyDir: %v\n", err)
if err = SyncDirs(filepath.Join(giteaRoot, "tests", "gitea-repositories-meta"), setting.RepoRootPath); err != nil {
fatalTestError("util.SyncDirs: %v\n", err)
}

if err = git.InitFull(context.Background()); err != nil {
fatalTestError("git.Init: %v\n", err)
}
ownerDirs, err := os.ReadDir(setting.RepoRootPath)
if err != nil {
fatalTestError("unable to read the new repo root: %v\n", err)
}
for _, ownerDir := range ownerDirs {
if !ownerDir.Type().IsDir() {
continue
}
repoDirs, err := os.ReadDir(filepath.Join(setting.RepoRootPath, ownerDir.Name()))
if err != nil {
fatalTestError("unable to read the new repo root: %v\n", err)
}
for _, repoDir := range repoDirs {
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "objects", "pack"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "objects", "info"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "refs", "heads"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "refs", "tag"), 0o755)
}
}

if len(testOpts) > 0 && testOpts[0].SetUp != nil {
if err := testOpts[0].SetUp(); err != nil {
Expand Down Expand Up @@ -255,24 +233,7 @@ func PrepareTestDatabase() error {
// by tests that use the above MainTest(..) function.
func PrepareTestEnv(t testing.TB) {
assert.NoError(t, PrepareTestDatabase())
assert.NoError(t, util.RemoveAll(setting.RepoRootPath))
metaPath := filepath.Join(giteaRoot, "tests", "gitea-repositories-meta")
assert.NoError(t, CopyDir(metaPath, setting.RepoRootPath))
ownerDirs, err := os.ReadDir(setting.RepoRootPath)
assert.NoError(t, err)
for _, ownerDir := range ownerDirs {
if !ownerDir.Type().IsDir() {
continue
}
repoDirs, err := os.ReadDir(filepath.Join(setting.RepoRootPath, ownerDir.Name()))
assert.NoError(t, err)
for _, repoDir := range repoDirs {
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "objects", "pack"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "objects", "info"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "refs", "heads"), 0o755)
_ = os.MkdirAll(filepath.Join(setting.RepoRootPath, ownerDir.Name(), repoDir.Name(), "refs", "tag"), 0o755)
}
}

assert.NoError(t, SyncDirs(metaPath, setting.RepoRootPath))
base.SetupGiteaRoot() // Makes sure GITEA_ROOT is set
}
9 changes: 6 additions & 3 deletions modules/queue/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ package queue

import (
"context"
"errors"
"sync"
"time"

Expand Down Expand Up @@ -32,6 +33,7 @@ type ManagedWorkerPoolQueue interface {

// FlushWithContext tries to make the handler process all items in the queue synchronously.
// It is for testing purpose only. It's not designed to be used in a cluster.
// Negative timeout means discarding all items in the queue.
FlushWithContext(ctx context.Context, timeout time.Duration) error

// RemoveAllItems removes all items in the base queue (on-the-fly items are not affected)
Expand Down Expand Up @@ -76,15 +78,16 @@ func (m *Manager) ManagedQueues() map[int64]ManagedWorkerPoolQueue {

// FlushAll tries to make all managed queues process all items synchronously, until timeout or the queue is empty.
// It is for testing purpose only. It's not designed to be used in a cluster.
// Negative timeout means discarding all items in the queue.
func (m *Manager) FlushAll(ctx context.Context, timeout time.Duration) error {
var finalErr error
var finalErrors []error
qs := m.ManagedQueues()
for _, q := range qs {
if err := q.FlushWithContext(ctx, timeout); err != nil {
finalErr = err // TODO: in Go 1.20: errors.Join
finalErrors = append(finalErrors, err)
}
}
return finalErr
return errors.Join(finalErrors...)
}

// CreateSimpleQueue creates a simple queue from global setting config provider by name
Expand Down
19 changes: 17 additions & 2 deletions modules/queue/workergroup.go
Original file line number Diff line number Diff line change
Expand Up @@ -197,15 +197,30 @@ func (q *WorkerPoolQueue[T]) doFlush(wg *workerGroup[T], flush flushType) {
defer log.Debug("Queue %q finishes flushing", q.GetName())

// stop all workers, and prepare a new worker context to start new workers

wg.ctxWorkerCancel()
wg.wg.Wait()

defer func() {
close(flush)
close(flush.c)
wg.doPrepareWorkerContext()
}()

if flush.timeout < 0 {
// discard everything
wg.batchBuffer = nil
for {
select {
case <-wg.popItemChan:
case <-wg.popItemErr:
case <-q.batchChan:
case <-q.ctxRun.Done():
return
default:
return
}
}
}

// drain the batch channel first
loop:
for {
Expand Down
11 changes: 7 additions & 4 deletions modules/queue/workerqueue.go
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,10 @@ type WorkerPoolQueue[T any] struct {
workerNumMu sync.Mutex
}

type flushType chan struct{}
type flushType struct {
timeout time.Duration
c chan struct{}
}

var _ ManagedWorkerPoolQueue = (*WorkerPoolQueue[any])(nil)

Expand Down Expand Up @@ -104,12 +107,12 @@ func (q *WorkerPoolQueue[T]) FlushWithContext(ctx context.Context, timeout time.
if timeout > 0 {
after = time.After(timeout)
}
c := make(flushType)
flush := flushType{timeout: timeout, c: make(chan struct{})}

// send flush request
// if it blocks, it means that there is a flush in progress or the queue hasn't been started yet
select {
case q.flushChan <- c:
case q.flushChan <- flush:
case <-ctx.Done():
return ctx.Err()
case <-q.ctxRun.Done():
Expand All @@ -120,7 +123,7 @@ func (q *WorkerPoolQueue[T]) FlushWithContext(ctx context.Context, timeout time.

// wait for flush to finish
select {
case <-c:
case <-flush.c:
return nil
case <-ctx.Done():
return ctx.Err()
Expand Down
2 changes: 1 addition & 1 deletion modules/testlogger/testlogger.go
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ func PrintCurrentTest(t testing.TB, skip ...int) func() {
_, _ = fmt.Fprintf(os.Stdout, "+++ %s ... still flushing after %v ...\n", t.Name(), SlowFlush)
}
})
if err := queue.GetManager().FlushAll(context.Background(), time.Minute); err != nil {
if err := queue.GetManager().FlushAll(context.Background(), -1); err != nil {
t.Errorf("Flushing queues failed with error %v", err)
}
timer.Stop()
Expand Down
2 changes: 1 addition & 1 deletion services/repository/adopt_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ func TestListUnadoptedRepositories_ListOptions(t *testing.T) {

func TestAdoptRepository(t *testing.T) {
assert.NoError(t, unittest.PrepareTestDatabase())
assert.NoError(t, unittest.CopyDir(filepath.Join(setting.RepoRootPath, "user2", "repo1.git"), filepath.Join(setting.RepoRootPath, "user2", "test-adopt.git")))
assert.NoError(t, unittest.SyncDirs(filepath.Join(setting.RepoRootPath, "user2", "repo1.git"), filepath.Join(setting.RepoRootPath, "user2", "test-adopt.git")))
user2 := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2})
_, err := AdoptRepository(db.DefaultContext, user2, user2, CreateRepoOptions{Name: "test-adopt"})
assert.NoError(t, err)
Expand Down
Loading
Loading