fix(tests): improve concurrency handling in worker tests and enhance module cloning
quality / commitlint (push) Skipped
quality / changes (push) Successful in 8s
quality / openapi (push) Skipped
quality / web (push) Skipped
quality / docker-check (push) Skipped
quality / go (push) Successful in 1m14s
quality / bird2 (push) Successful in 18s
CD / quality (push) Successful in 1m47s
CD / publish (push) Successful in 2m53s
quality / commitlint (push) Skipped
quality / changes (push) Successful in 8s
quality / openapi (push) Skipped
quality / web (push) Skipped
quality / docker-check (push) Skipped
quality / go (push) Successful in 1m14s
quality / bird2 (push) Successful in 18s
CD / quality (push) Successful in 1m47s
CD / publish (push) Successful in 2m53s
- Added environment variable `EVOBGP_JOB_MAX_CONCURRENT` to control job concurrency in tests. - Modified worker test to ensure proper synchronization of job processing by holding workers until both jobs are enqueued. - Updated memory store methods to return cloned module instances, preventing unintended mutations of original modules during operations.
This commit is contained in:
@@ -10,6 +10,7 @@ import (
|
||||
func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) {
|
||||
t.Setenv("EVOBGP_ASN_RESOLVE", "0")
|
||||
t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "") // skip bird binary path in deploy_apply
|
||||
t.Setenv("EVOBGP_JOB_MAX_CONCURRENT", "8")
|
||||
|
||||
m := store.NewMemory()
|
||||
m.SeedDemo()
|
||||
@@ -35,8 +36,15 @@ func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Hold workers until both jobs are enqueued so inflightRefresh=2 before either
|
||||
// finishModuleRefreshSuccess. Otherwise a fast ingest can finalize+deploy before
|
||||
// the second Enqueue — sequential refreshes correctly produce two deploy_apply jobs.
|
||||
start := make(chan struct{})
|
||||
wk := &Worker{Store: m}
|
||||
reg := NewRegistry(wk.Process)
|
||||
reg := NewRegistry(func(j *Job) {
|
||||
<-start
|
||||
wk.Process(j)
|
||||
})
|
||||
wk.Registry = reg
|
||||
|
||||
mid1 := modIP
|
||||
@@ -47,6 +55,7 @@ func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) {
|
||||
if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid2, map[string]any{"module_id": mod2.ID}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
close(start)
|
||||
|
||||
waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, 2)
|
||||
|
||||
|
||||
@@ -414,6 +414,9 @@ func (m *Memory) ListModules(tenantID string) []*Module {
|
||||
}
|
||||
return out[i].Name < out[j].Name
|
||||
})
|
||||
for i := range out {
|
||||
out[i] = cloneModule(out[i])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -446,7 +449,7 @@ func (m *Memory) GetModule(tenantID, moduleID string) (*Module, error) {
|
||||
if mod.TenantID != tenantID {
|
||||
return nil, ErrTenantScope
|
||||
}
|
||||
return mod, nil
|
||||
return cloneModule(mod), nil
|
||||
}
|
||||
|
||||
func (m *Memory) GetRevision(tenantID, revisionID string) (*Revision, error) {
|
||||
@@ -684,3 +687,24 @@ func (m *Memory) ListRevisions(tenantID, moduleID string, cursor string, limit i
|
||||
}
|
||||
return page, nextCursor, hasMore
|
||||
}
|
||||
|
||||
func cloneStringPtr(s *string) *string {
|
||||
if s == nil {
|
||||
return nil
|
||||
}
|
||||
v := *s
|
||||
return &v
|
||||
}
|
||||
|
||||
func cloneModule(m *Module) *Module {
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
cp := *m
|
||||
cp.DefaultCommunityID = cloneStringPtr(m.DefaultCommunityID)
|
||||
cp.DohProfileID = cloneStringPtr(m.DohProfileID)
|
||||
cp.DohProfileIDs = append([]string(nil), m.DohProfileIDs...)
|
||||
cp.LastRefreshedAt = cloneTime(m.LastRefreshedAt)
|
||||
cp.DeletedAt = cloneTime(m.DeletedAt)
|
||||
return &cp
|
||||
}
|
||||
|
||||
@@ -38,7 +38,7 @@ func (m *Memory) CreateModule(tenantID string, in *Module) (*Module, error) {
|
||||
}
|
||||
NormalizeModuleDoh(mod)
|
||||
m.modules[id] = mod
|
||||
return mod, nil
|
||||
return cloneModule(mod), nil
|
||||
}
|
||||
|
||||
func (m *Memory) UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*Module, error) {
|
||||
@@ -74,12 +74,14 @@ func (m *Memory) UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*M
|
||||
mod.DefaultCommunityID = &v
|
||||
}
|
||||
}
|
||||
ApplyModuleDohPatch(mod, patch)
|
||||
if patch.DohProfileIDs != nil || patch.DohProfileID != nil || patch.DohResolverPolicy != nil {
|
||||
ApplyModuleDohPatch(mod, patch)
|
||||
}
|
||||
if patch.LastRefreshedAt != nil {
|
||||
t := patch.LastRefreshedAt.UTC()
|
||||
mod.LastRefreshedAt = &t
|
||||
}
|
||||
return mod, nil
|
||||
return cloneModule(mod), nil
|
||||
}
|
||||
|
||||
func (m *Memory) SoftDeleteModule(tenantID, moduleID string) error {
|
||||
|
||||
Reference in New Issue
Block a user