john-okeefe
b3b40b77d6
Problem: TestWorker_ConcurrentJobs was using a fixed 3-second sleep to wait for concurrent scan jobs to complete. However, this wasn't sufficient time for the watch mode to enqueue and process the jobs. When the test function ended, Go's testing framework deleted all t.TempDir() directories, causing the scanner to fail with 'no such file or directory' errors. Error messages: Processing media file: /tmp/.../002/book0.epub Failed to get file info for /tmp/.../002/book0.epub: stat ...: no such file or directory Root Cause: The test created temporary directories and files using t.TempDir(), which are automatically cleaned up when the test function ends. The scanner needs time to process the files, but the test only waited 3 seconds before checking results, causing temp dirs to be deleted mid-scan. Solution: Replaced the fixed 3-second sleep with proper job polling that: 1. Stores job IDs when submitting them to the worker 2. Polls job status every 100ms up to a 15-second timeout 3. Waits until all 3 jobs reach Completed or Failed status 4. Only then checks for media items in the database This ensures the scanner has finished processing all files before the test ends and temp dirs are cleaned up. Matches the polling pattern used in TestWorker_DirectoryScanJob. Files changed: - cmd/server/tests/worker_test.go: Added job tracking and proper polling
340 lines
9.6 KiB
Go
340 lines
9.6 KiB
Go
package main
|
|
|
|
import (
|
|
"bookhoard/internal/database"
|
|
"bookhoard/internal/services"
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func TestWorker_DirectoryScanJob(t *testing.T) {
|
|
setup := setupTestServer(t)
|
|
|
|
token := setup.Token
|
|
|
|
// Create library
|
|
createLibReq := map[string]interface{}{
|
|
"name": "Worker Test Library",
|
|
"description": "Test library for worker",
|
|
"type": "ebooks",
|
|
}
|
|
createLibURL := setup.Server.URL + "/api/libraries"
|
|
createLibReqHTTP, _ := http.NewRequest("POST", createLibURL, bytes.NewBuffer(jsonMarshal(createLibReq)))
|
|
createLibReqHTTP.Header.Set("Content-Type", "application/json")
|
|
createLibReqHTTP.Header.Set("Authorization", "Bearer "+token)
|
|
|
|
client := &http.Client{}
|
|
createLibResp, err := client.Do(createLibReqHTTP)
|
|
require.NoError(t, err)
|
|
require.Equal(t, http.StatusCreated, createLibResp.StatusCode)
|
|
|
|
var createLibResponse map[string]interface{}
|
|
json.NewDecoder(createLibResp.Body).Decode(&createLibResponse)
|
|
createLibResp.Body.Close()
|
|
|
|
libraryID, ok := createLibResponse["id"].(string)
|
|
require.True(t, ok)
|
|
require.NotEmpty(t, libraryID)
|
|
|
|
// Create a temporary directory for testing
|
|
tmpDir := t.TempDir()
|
|
|
|
// Add folder to library
|
|
folderURL := fmt.Sprintf("%s/api/libraries/%s/folders", setup.Server.URL, libraryID)
|
|
folderReq := map[string]interface{}{
|
|
"folder_path": tmpDir,
|
|
}
|
|
req, _ := http.NewRequest("POST", folderURL, bytes.NewBuffer(jsonMarshal(folderReq)))
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("Authorization", "Bearer "+token)
|
|
|
|
resp, err := client.Do(req)
|
|
require.NoError(t, err)
|
|
resp.Body.Close()
|
|
require.Equal(t, http.StatusCreated, resp.StatusCode)
|
|
|
|
// Create test files in the directory
|
|
for i := 0; i < 3; i++ {
|
|
filePath := filepath.Join(tmpDir, fmt.Sprintf("book%d.epub", i))
|
|
err := os.WriteFile(filePath, []byte(fmt.Sprintf("test content %d", i)), 0644)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// Submit directory scan job through WorkerInstance
|
|
job := &services.Job{
|
|
ID: uuid.New().String(),
|
|
Type: services.JobTypeDirectoryScan,
|
|
Params: map[string]interface{}{
|
|
"directory": tmpDir,
|
|
"db": setup.DB,
|
|
},
|
|
Status: services.JobStatusPending,
|
|
}
|
|
|
|
// Use EnqueueJob which returns errors
|
|
err = services.WorkerInstance.EnqueueJob(job)
|
|
require.NoError(t, err, "Failed to enqueue job")
|
|
// Poll for job completion
|
|
timeout := time.Now().Add(10 * time.Second)
|
|
for time.Now().Before(timeout) {
|
|
if status, exists := services.WorkerInstance.GetJobStatus(job.ID); exists {
|
|
if status.Status == services.JobStatusCompleted || status.Status == services.JobStatusFailed {
|
|
break
|
|
}
|
|
}
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
|
|
// Debug: Check job status after polling
|
|
if finalStatus, exists := services.WorkerInstance.GetJobStatus(job.ID); exists {
|
|
t.Logf("Job %s final status: %s, error: %s", job.ID, finalStatus.Status, finalStatus.Error)
|
|
} else {
|
|
t.Logf("Job %s not found in results", job.ID)
|
|
}
|
|
|
|
// Check that items were created in database
|
|
ctx := context.Background()
|
|
items, err := setup.DB.ListMediaItems(ctx, database.ListMediaItemsParams{
|
|
Limit: 1000,
|
|
Offset: 0,
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Should have at least 3 items (our test files)
|
|
greaterOrEqual := func(a, b int) bool {
|
|
return a >= b
|
|
}
|
|
assert.True(t, greaterOrEqual(len(items), 3), fmt.Sprintf("Expected at least 3 items, got %d", len(items)))
|
|
}
|
|
|
|
func TestWorker_SetFoldersJob(t *testing.T) {
|
|
setup := setupTestServer(t)
|
|
|
|
token := setup.Token
|
|
|
|
// Create library
|
|
createLibReq := map[string]interface{}{
|
|
"name": "Set Folders Test Library",
|
|
"description": "Test library for set folders",
|
|
"type": "ebooks",
|
|
}
|
|
createLibURL := setup.Server.URL + "/api/libraries"
|
|
createLibReqHTTP, _ := http.NewRequest("POST", createLibURL, bytes.NewBuffer(jsonMarshal(createLibReq)))
|
|
createLibReqHTTP.Header.Set("Content-Type", "application/json")
|
|
createLibReqHTTP.Header.Set("Authorization", "Bearer "+token)
|
|
|
|
client := &http.Client{}
|
|
createLibResp, err := client.Do(createLibReqHTTP)
|
|
require.NoError(t, err)
|
|
require.Equal(t, http.StatusCreated, createLibResp.StatusCode)
|
|
|
|
var createLibResponse map[string]interface{}
|
|
json.NewDecoder(createLibResp.Body).Decode(&createLibResponse)
|
|
createLibResp.Body.Close()
|
|
|
|
libraryID, ok := createLibResponse["id"].(string)
|
|
require.True(t, ok)
|
|
require.NotEmpty(t, libraryID)
|
|
|
|
// Create temporary directory
|
|
tmpDir := t.TempDir()
|
|
|
|
// Add folder to library via HTTP API
|
|
addFolderToLibrary(t, setup, libraryID, tmpDir)
|
|
|
|
// Submit set folders job through WorkerInstance
|
|
job := &services.Job{
|
|
ID: uuid.New().String(),
|
|
Type: services.JobTypeSetFolders,
|
|
Params: map[string]interface{}{
|
|
"folders": []string{tmpDir},
|
|
"db": setup.DB,
|
|
},
|
|
Status: services.JobStatusPending,
|
|
}
|
|
|
|
err = services.WorkerInstance.EnqueueJob(job)
|
|
require.NoError(t, err, "Failed to enqueue set folders job")
|
|
|
|
// Wait for job to process
|
|
time.Sleep(1 * time.Second)
|
|
|
|
// Verify folder was added to library
|
|
ctx := context.Background()
|
|
folders, err := setup.DB.GetLibraryFolders(ctx, pgtype.UUID{Bytes: [16]byte(uuid.MustParse(libraryID)), Valid: true})
|
|
require.NoError(t, err)
|
|
assert.NotEmpty(t, folders, "Library should have folders")
|
|
|
|
// Find our folder
|
|
found := false
|
|
for _, folder := range folders {
|
|
if folder.FolderPath == tmpDir {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
assert.True(t, found, "Folder should be in library")
|
|
}
|
|
|
|
func TestWorker_ConcurrentJobs(t *testing.T) {
|
|
setup := setupTestServer(t)
|
|
|
|
ctx := context.Background()
|
|
|
|
// Create a library and get admin user
|
|
adminID := getTestUserID(t, setup.DB)
|
|
|
|
libraryType, err := setup.DB.GetLibraryTypeByName(ctx, "ebooks")
|
|
require.NoError(t, err)
|
|
|
|
library, err := setup.DB.CreateLibrary(ctx, database.CreateLibraryParams{
|
|
Name: "Concurrent Job Test Library",
|
|
Description: pgtype.Text{String: "Test library", Valid: true},
|
|
LibraryTypeID: libraryType.ID,
|
|
CreatedByAdminID: pgtype.UUID{Bytes: [16]byte(adminID), Valid: true},
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Create multiple temporary directories
|
|
var tmpDirs []string
|
|
for i := 0; i < 3; i++ {
|
|
tmpDir := t.TempDir()
|
|
tmpDirs = append(tmpDirs, tmpDir)
|
|
|
|
// Add folder to library
|
|
_, err = setup.DB.AddLibraryFolder(ctx, database.AddLibraryFolderParams{
|
|
LibraryID: library.ID,
|
|
FolderPath: tmpDir,
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Create test files in each directory
|
|
for j := 0; j < 2; j++ {
|
|
filePath := filepath.Join(tmpDir, fmt.Sprintf("book%d.epub", j))
|
|
err = os.WriteFile(filePath, []byte(fmt.Sprintf("test content %d", j)), 0644)
|
|
require.NoError(t, err, "Failed to create test file")
|
|
}
|
|
}
|
|
|
|
// Submit multiple concurrent jobs and track their IDs
|
|
var jobs []*services.Job
|
|
for _, tmpDir := range tmpDirs {
|
|
job := &services.Job{
|
|
ID: uuid.New().String(),
|
|
Type: services.JobTypeDirectoryScan,
|
|
Params: map[string]interface{}{
|
|
"directory": tmpDir,
|
|
"db": setup.DB,
|
|
},
|
|
Status: services.JobStatusPending,
|
|
}
|
|
jobs = append(jobs, job)
|
|
err = services.WorkerInstance.EnqueueJob(job)
|
|
require.NoError(t, err, "Failed to enqueue scan job")
|
|
}
|
|
|
|
// Wait for all jobs to complete
|
|
timeout := time.Now().Add(15 * time.Second)
|
|
completedJobs := 0
|
|
for time.Now().Before(timeout) && completedJobs < 3 {
|
|
for _, job := range jobs {
|
|
if status, exists := services.WorkerInstance.GetJobStatus(job.ID); exists {
|
|
if status.Status == services.JobStatusCompleted || status.Status == services.JobStatusFailed {
|
|
completedJobs++
|
|
}
|
|
}
|
|
}
|
|
if completedJobs < 3 {
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// Verify all directories were processed
|
|
allItems, err := setup.DB.ListMediaItems(ctx, database.ListMediaItemsParams{
|
|
Limit: 1000,
|
|
Offset: 0,
|
|
})
|
|
require.NoError(t, err)
|
|
assert.NotEmpty(t, allItems, "Should have media items from concurrent jobs")
|
|
}
|
|
|
|
func TestWorker_JobStatusTracking(t *testing.T) {
|
|
setup := setupTestServer(t)
|
|
|
|
ctx := context.Background()
|
|
|
|
// Create test library
|
|
adminID := getTestUserID(t, setup.DB)
|
|
libraryType, err := setup.DB.GetLibraryTypeByName(ctx, "ebooks")
|
|
require.NoError(t, err)
|
|
|
|
library, err := setup.DB.CreateLibrary(ctx, database.CreateLibraryParams{
|
|
Name: "Job Status Test Library",
|
|
Description: pgtype.Text{String: "Test library", Valid: true},
|
|
LibraryTypeID: libraryType.ID,
|
|
CreatedByAdminID: pgtype.UUID{Bytes: [16]byte(adminID), Valid: true},
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
tmpDir := t.TempDir()
|
|
_, err = setup.DB.AddLibraryFolder(ctx, database.AddLibraryFolderParams{
|
|
LibraryID: library.ID,
|
|
FolderPath: tmpDir,
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Submit job and track status
|
|
jobID := uuid.New().String()
|
|
job := &services.Job{
|
|
ID: jobID,
|
|
Type: services.JobTypeDirectoryScan,
|
|
Params: map[string]interface{}{
|
|
"directory": tmpDir,
|
|
"db": setup.DB,
|
|
},
|
|
Status: services.JobStatusPending,
|
|
}
|
|
|
|
err = services.WorkerInstance.EnqueueJob(job)
|
|
require.NoError(t, err, "Failed to enqueue set folders job")
|
|
|
|
// Poll job status
|
|
var finalStatus string
|
|
for i := 0; i < 10; i++ {
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
result, exists := services.WorkerInstance.GetJobStatus(jobID)
|
|
if exists {
|
|
finalStatus = string(result.Status)
|
|
if result.Status == services.JobStatusCompleted || result.Status == services.JobStatusFailed {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
// Job should complete (success or failure)
|
|
assert.True(t, finalStatus == "completed" || finalStatus == "failed",
|
|
fmt.Sprintf("Job should complete, got status: %s", finalStatus))
|
|
}
|
|
|
|
// Helper function to marshal JSON
|
|
func jsonMarshal(v interface{}) []byte {
|
|
b, err := json.Marshal(v)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return b
|
|
}
|