Skip to content

Commit fa51163

Browse files
siherrmannsiherrmann
committed
add queuer tests, add more database tests, update worker options check
1 parent 78da4aa commit fa51163

5 files changed

Lines changed: 325 additions & 8 deletions

File tree

‎helper/database_test.go‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,62 @@ func TestMain(m *testing.M) {
2828
}
2929
}
3030

31+
func TestNewDatabaseConfig(t *testing.T) {
32+
tests := []struct {
33+
name string
34+
envs map[string]string
35+
expectError bool
36+
}{
37+
{
38+
name: "Valid environment variables",
39+
envs: map[string]string{
40+
"QUEUER_DB_HOST": "localhost",
41+
"QUEUER_DB_PORT": dbPort,
42+
"QUEUER_DB_DATABASE": "database",
43+
"QUEUER_DB_USERNAME": "user",
44+
"QUEUER_DB_PASSWORD": "password",
45+
"QUEUER_DB_SCHEMA": "public",
46+
},
47+
expectError: false,
48+
},
49+
{
50+
name: "Missing environment variable",
51+
envs: map[string]string{
52+
"QUEUER_DB_HOST": "localhost",
53+
"QUEUER_DB_PORT": dbPort,
54+
"QUEUER_DB_DATABASE": "database",
55+
"QUEUER_DB_USERNAME": "user",
56+
// "QUEUER_DB_PASSWORD": "password", // Intentionally missing
57+
"QUEUER_DB_SCHEMA": "public",
58+
},
59+
expectError: true,
60+
},
61+
}
62+
63+
for _, test := range tests {
64+
t.Run(test.name, func(t *testing.T) {
65+
for key, value := range test.envs {
66+
t.Setenv(key, value)
67+
}
68+
69+
dbConfig, err := NewDatabaseConfiguration()
70+
if test.expectError {
71+
assert.NotNil(t, err, "expected an error for %s, but got none", test.name)
72+
assert.Nil(t, dbConfig, "expected NewTestDatabaseConfig to return nil on error")
73+
} else {
74+
assert.NotNil(t, dbConfig, "expected NewTestDatabaseConfig to return a non-nil instance")
75+
assert.Equal(t, test.envs["QUEUER_DB_HOST"], dbConfig.Host, "expected host to be 'localhost', got %s", dbConfig.Host)
76+
assert.Equal(t, test.envs["QUEUER_DB_PORT"], dbConfig.Port, "expected port to be %s, got %s", dbPort, dbConfig.Port)
77+
assert.Equal(t, test.envs["QUEUER_DB_DATABASE"], dbConfig.Database, "expected database to be 'queuer_test', got %s", dbConfig.Database)
78+
assert.Equal(t, test.envs["QUEUER_DB_USERNAME"], dbConfig.Username, "expected username to be 'postgres', got %s", dbConfig.Username)
79+
assert.Equal(t, test.envs["QUEUER_DB_PASSWORD"], dbConfig.Password, "expected password to be empty, got %s", dbConfig.Password)
80+
assert.Equal(t, test.envs["QUEUER_DB_SCHEMA"], dbConfig.Schema, "expected schema to be 'public', got %s", dbConfig.Schema)
81+
}
82+
83+
})
84+
}
85+
}
86+
3187
func TestNew(t *testing.T) {
3288
dbConfig := NewTestDatabaseConfig(dbPort)
3389
database := NewTestDatabase(dbConfig)

‎model/worker.go‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,9 @@ func NewWorker(name string, maxConcurrency int) (*Worker, error) {
4242
}
4343

4444
func NewWorkerWithOptions(name string, maxConcurrency int, options *OnError) (*Worker, error) {
45-
if len(name) == 0 || len(name) > 100 {
46-
return nil, fmt.Errorf("name must have a length between 1 and 100")
47-
}
48-
49-
if maxConcurrency < 1 || maxConcurrency > 1000 {
50-
return nil, fmt.Errorf("maxConcurrency must be between 1 and 100")
45+
err := options.IsValid()
46+
if err != nil {
47+
return nil, fmt.Errorf("invalid options: %w", err)
5148
}
5249

5350
return &Worker{

‎queuer.go‎

Lines changed: 64 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ type Queuer struct {
3737
}
3838

3939
// NewQueuer creates a new Queuer instance with the given name and max concurrency.
40-
// It initializes the database connection, job listeners, and worker.
40+
// It initializes the database connection, job listeners and worker.
4141
// If options are provided, it creates a worker with those options.
4242
// If any error occurs during initialization, it logs a panic error and exits the program.
4343
// It returns a pointer to the newly created Queuer instance.
@@ -99,7 +99,8 @@ func NewQueuer(name string, maxConcurrency int, options ...*model.OnError) *Queu
9999
if err != nil {
100100
logger.Panicf("error inserting worker: %v", err)
101101
}
102-
logger.Printf("Worker %s created with RID %s", worker.Name, worker.RID.String())
102+
103+
logger.Printf("Queuer %s created with worker RID %s", worker.Name, worker.RID.String())
103104

104105
return &Queuer{
105106
worker: worker,
@@ -115,6 +116,67 @@ func NewQueuer(name string, maxConcurrency int, options ...*model.OnError) *Queu
115116
}
116117
}
117118

119+
// NewQueuerWithoutWorker creates a new Queuer instance without a worker.
120+
// This is useful for scenarios where the queuer needs to be initialized without a worker,
121+
// such as when a seperate service is responsible for job status endpoints without processing jobs.
122+
// It initializes the database connection and job listeners.
123+
// If any error occurs during initialization, it logs a panic error and exits the program.
124+
// It returns a pointer to the newly created Queuer instance.
125+
func NewQueuerWithoutWorker() *Queuer {
126+
// Logger
127+
logger := log.New(os.Stdout, "Queuer: ", log.Ltime)
128+
129+
// Database
130+
dbConfig, err := helper.NewDatabaseConfiguration()
131+
if err != nil {
132+
logger.Panicf("failed to create database configuration: %v", err)
133+
}
134+
dbConnection := helper.NewDatabase(
135+
"queuer",
136+
dbConfig,
137+
)
138+
139+
// DBs
140+
var dbJob database.JobDBHandlerFunctions
141+
var dbWorker database.WorkerDBHandlerFunctions
142+
dbJob, err = database.NewJobDBHandler(dbConnection)
143+
if err != nil {
144+
logger.Panicf("failed to create job db handler: %v", err)
145+
}
146+
dbWorker, err = database.NewWorkerDBHandler(dbConnection)
147+
if err != nil {
148+
logger.Panicf("failed to create worker db handler: %v", err)
149+
}
150+
151+
// Job listeners
152+
// jobInsertListener, err := database.NewQueuerListener(dbConfig, "job.INSERT")
153+
// if err != nil {
154+
// logger.Panicf("failed to create job insert listener: %v", err)
155+
// }
156+
// jobUpdateListener, err := database.NewQueuerListener(dbConfig, "job.UPDATE")
157+
// if err != nil {
158+
// logger.Panicf("failed to create job update listener: %v", err)
159+
// }
160+
// jobDeleteListener, err := database.NewQueuerListener(dbConfig, "job.DELETE")
161+
// if err != nil {
162+
// logger.Panicf("failed to create job delete listener: %v", err)
163+
// }
164+
165+
logger.Println("Queuer without worker created")
166+
167+
return &Queuer{
168+
dbJob: dbJob,
169+
dbWorker: dbWorker,
170+
// jobInsertListener: jobInsertListener,
171+
// jobUpdateListener: jobUpdateListener,
172+
// jobDeleteListener: jobDeleteListener,
173+
JobPollInterval: 1 * time.Minute,
174+
tasks: map[string]*model.Task{},
175+
nextIntervalFuncs: map[string]model.NextIntervalFunc{},
176+
log: logger,
177+
}
178+
}
179+
118180
// Start starts the queuer by initializing the job listeners and starting the job poll ticker.
119181
// It checks if the queuer is initialized properly, and if not, it logs a panic error and exits the program.
120182
// It runs the job processing in a separate goroutine and listens for job events.

‎queuerNextInterval_test.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
func MockNextIntervalFunc1(start time.Time, currentCount int) time.Time {
1313
return start.Add(time.Hour * time.Duration(currentCount))
1414
}
15+
1516
func MockNextIntervalFunc2(start time.Time, currentCount int) time.Time {
1617
return start.Add(time.Hour * time.Duration(currentCount))
1718
}

‎queuer_test.go‎

Lines changed: 201 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,201 @@
1+
package queuer
2+
3+
import (
4+
"queuer/model"
5+
"testing"
6+
7+
"github.com/stretchr/testify/assert"
8+
"github.com/stretchr/testify/require"
9+
)
10+
11+
func TestNewQueuer(t *testing.T) {
12+
tests := []struct {
13+
name string
14+
maxConcurrency int
15+
options []*model.OnError
16+
dbEnvs map[string]string
17+
expectError bool
18+
}{
19+
{
20+
name: "Valid queuer",
21+
maxConcurrency: 100,
22+
options: nil,
23+
dbEnvs: map[string]string{
24+
"QUEUER_DB_HOST": "localhost",
25+
"QUEUER_DB_PORT": dbPort,
26+
"QUEUER_DB_DATABASE": "database",
27+
"QUEUER_DB_USERNAME": "user",
28+
"QUEUER_DB_PASSWORD": "password",
29+
"QUEUER_DB_SCHEMA": "public",
30+
},
31+
expectError: false,
32+
},
33+
{
34+
name: "Valid queuer with options",
35+
maxConcurrency: 100,
36+
options: []*model.OnError{
37+
{
38+
Timeout: 10.0,
39+
MaxRetries: 3,
40+
RetryDelay: 1.0,
41+
RetryBackoff: model.RETRY_BACKOFF_LINEAR,
42+
},
43+
},
44+
dbEnvs: map[string]string{
45+
"QUEUER_DB_HOST": "localhost",
46+
"QUEUER_DB_PORT": dbPort,
47+
"QUEUER_DB_DATABASE": "database",
48+
"QUEUER_DB_USERNAME": "user",
49+
"QUEUER_DB_PASSWORD": "password",
50+
"QUEUER_DB_SCHEMA": "public",
51+
},
52+
expectError: false,
53+
},
54+
{
55+
name: "Invalid max concurrency",
56+
maxConcurrency: -1,
57+
options: nil,
58+
dbEnvs: map[string]string{
59+
"QUEUER_DB_HOST": "localhost",
60+
"QUEUER_DB_PORT": dbPort,
61+
"QUEUER_DB_DATABASE": "database",
62+
"QUEUER_DB_USERNAME": "user",
63+
"QUEUER_DB_PASSWORD": "password",
64+
"QUEUER_DB_SCHEMA": "public",
65+
},
66+
expectError: true,
67+
},
68+
{
69+
name: "Invalid options",
70+
maxConcurrency: 100,
71+
options: []*model.OnError{
72+
{
73+
Timeout: -10.0, // Invalid timeout value
74+
MaxRetries: 3,
75+
RetryDelay: 1.0,
76+
RetryBackoff: model.RETRY_BACKOFF_LINEAR,
77+
},
78+
},
79+
dbEnvs: map[string]string{
80+
"QUEUER_DB_HOST": "localhost",
81+
"QUEUER_DB_PORT": dbPort,
82+
"QUEUER_DB_DATABASE": "database",
83+
"QUEUER_DB_USERNAME": "user",
84+
"QUEUER_DB_PASSWORD": "password",
85+
"QUEUER_DB_SCHEMA": "public",
86+
},
87+
expectError: true,
88+
},
89+
{
90+
name: "Missing DB environment variable",
91+
maxConcurrency: 100,
92+
options: nil,
93+
dbEnvs: map[string]string{
94+
"QUEUER_DB_HOST": "localhost",
95+
"QUEUER_DB_PORT": dbPort,
96+
// "QUEUER_DB_DATABASE": "database", // Intentionally missing
97+
"QUEUER_DB_USERNAME": "user",
98+
"QUEUER_DB_PASSWORD": "password",
99+
"QUEUER_DB_SCHEMA": "public",
100+
},
101+
expectError: true,
102+
},
103+
}
104+
105+
for _, test := range tests {
106+
t.Run(test.name, func(t *testing.T) {
107+
for key, value := range test.dbEnvs {
108+
t.Setenv(key, value)
109+
}
110+
111+
if test.expectError {
112+
defer func() {
113+
if r := recover(); r == nil {
114+
t.Errorf("Expected panic for %s, but did not get one", test.name)
115+
}
116+
}()
117+
}
118+
119+
queuer := NewQueuer(test.name, test.maxConcurrency, test.options...)
120+
if !test.expectError {
121+
require.NotNil(t, queuer, "Expected Queuer to be created successfully")
122+
assert.Equal(t, test.name, queuer.worker.Name, "Expected Queuer name to match")
123+
assert.Equal(t, test.maxConcurrency, queuer.worker.MaxConcurrency, "Expected Queuer max concurrency to match")
124+
}
125+
})
126+
}
127+
}
128+
129+
func TestNewQueuerWithoutWorker(t *testing.T) {
130+
tests := []struct {
131+
name string
132+
dbEnvs map[string]string
133+
expectError bool
134+
}{
135+
{
136+
name: "Valid queuer without worker",
137+
dbEnvs: map[string]string{
138+
"QUEUER_DB_HOST": "localhost",
139+
"QUEUER_DB_PORT": dbPort,
140+
"QUEUER_DB_DATABASE": "database",
141+
"QUEUER_DB_USERNAME": "user",
142+
"QUEUER_DB_PASSWORD": "password",
143+
"QUEUER_DB_SCHEMA": "public",
144+
},
145+
expectError: false,
146+
},
147+
{
148+
name: "Missing DB environment variable",
149+
dbEnvs: map[string]string{
150+
"QUEUER_DB_HOST": "localhost",
151+
"QUEUER_DB_PORT": dbPort,
152+
// "QUEUER_DB_DATABASE": "database", // Intentionally missing
153+
"QUEUER_DB_USERNAME": "user",
154+
"QUEUER_DB_PASSWORD": "password",
155+
"QUEUER_DB_SCHEMA": "public",
156+
},
157+
expectError: true,
158+
},
159+
}
160+
161+
for _, test := range tests {
162+
t.Run(test.name, func(t *testing.T) {
163+
for key, value := range test.dbEnvs {
164+
t.Setenv(key, value)
165+
}
166+
167+
if test.expectError {
168+
defer func() {
169+
if r := recover(); r == nil {
170+
t.Errorf("Expected panic for %s, but did not get one", test.name)
171+
}
172+
}()
173+
}
174+
175+
queuer := NewQueuerWithoutWorker()
176+
if !test.expectError {
177+
require.NotNil(t, queuer, "Expected Queuer to be created successfully")
178+
}
179+
})
180+
}
181+
}
182+
183+
func TestStop(t *testing.T) {
184+
envs := map[string]string{
185+
"QUEUER_DB_HOST": "localhost",
186+
"QUEUER_DB_PORT": dbPort,
187+
"QUEUER_DB_DATABASE": "database",
188+
"QUEUER_DB_USERNAME": "user",
189+
"QUEUER_DB_PASSWORD": "password",
190+
"QUEUER_DB_SCHEMA": "public",
191+
}
192+
for key, value := range envs {
193+
t.Setenv(key, value)
194+
}
195+
196+
queuer := NewQueuer("test", 10)
197+
require.NotNil(t, queuer, "Expected Queuer to be created successfully")
198+
199+
err := queuer.Stop()
200+
assert.NoError(t, err, "Expected Stop to complete without error")
201+
}

0 commit comments

Comments
 (0)