MiFaZhan
250 lines
8.5 KiB
Go
250 lines
8.5 KiB
Go
package monthlyusage
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
|
)
|
|
|
|
func newScopeProducer(t *testing.T, cards []Card) (*Producer, *producerRedis) {
|
|
t.Helper()
|
|
store := testStore(t)
|
|
scanner := scannerFor(cards, cards[len(cards)-1].ID)
|
|
for i := range scanner.channels {
|
|
for _, c := range cards {
|
|
if c.ChanKey == scanner.channels[i].ChanKey {
|
|
scanner.channels[i].TempKey = c.TempKey
|
|
scanner.channels[i].SyncEnabled = c.SyncEnabled
|
|
break
|
|
}
|
|
}
|
|
}
|
|
p, err := New(scanner, store, Config{PageSize: 100})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return p, &producerRedis{enqueueBatches: map[string]int{}}
|
|
}
|
|
func scopedMessage(runID string, channels ...string) model.TrafficUsageMonthlyStartMessage {
|
|
return model.TrafficUsageMonthlyStartMessage{
|
|
RunID: runID,
|
|
Operation: "current_month_usage_full_sync",
|
|
QueryMonth: "202608",
|
|
Scope: &model.SyncScope{Channels: channels},
|
|
}
|
|
}
|
|
|
|
func TestProduceScopeEnqueuesOnlySelectedChannels(t *testing.T) {
|
|
producer, redis := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", Operator: 1},
|
|
{ID: 3, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
{ID: 4, ChanKey: "189", TempKey: "sim_yd_v5", Operator: 1},
|
|
})
|
|
|
|
if err := producer.Produce(context.Background(), scopedMessage("scoped", "217")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
redis.enqueueBatches = map[string]int{}
|
|
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
|
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
|
n, _ := producer.store.SizeKey(context.Background(), key)
|
|
if n > 0 {
|
|
redis.enqueueBatches["flow:"+ch] = int(n)
|
|
}
|
|
}
|
|
redis.mu.Lock()
|
|
defer redis.mu.Unlock()
|
|
if len(redis.enqueueBatches) != 1 {
|
|
t.Fatalf("enqueue batches = %+v, want only the 217 channel", redis.enqueueBatches)
|
|
}
|
|
if redis.enqueueBatches["flow:217"] != 2 {
|
|
t.Fatalf("enqueue batches = %+v, want 2 tasks on flow:217", redis.enqueueBatches)
|
|
}
|
|
}
|
|
|
|
func TestProduceScopeMultipleChannels(t *testing.T) {
|
|
producer, redis := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", Operator: 1},
|
|
{ID: 3, ChanKey: "189", TempKey: "sim_yd_v5", Operator: 1},
|
|
})
|
|
|
|
if err := producer.Produce(context.Background(), scopedMessage("scoped", "189", "217")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
redis.enqueueBatches = map[string]int{}
|
|
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
|
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
|
n, _ := producer.store.SizeKey(context.Background(), key)
|
|
if n > 0 {
|
|
redis.enqueueBatches["flow:"+ch] = int(n)
|
|
}
|
|
}
|
|
redis.mu.Lock()
|
|
defer redis.mu.Unlock()
|
|
if len(redis.enqueueBatches) != 2 {
|
|
t.Fatalf("enqueue batches = %+v, want 217 and 189", redis.enqueueBatches)
|
|
}
|
|
if _, ok := redis.enqueueBatches["flow:173"]; ok {
|
|
t.Fatalf("channel 173 was enqueued despite not being selected: %+v", redis.enqueueBatches)
|
|
}
|
|
}
|
|
|
|
func TestProduceFullScopeKeepsEveryChannel(t *testing.T) {
|
|
producer, redis := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", Operator: 1},
|
|
})
|
|
|
|
if err := producer.Produce(context.Background(), model.TrafficUsageMonthlyStartMessage{
|
|
RunID: "full",
|
|
Operation: "current_month_usage_full_sync",
|
|
QueryMonth: "202608",
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
redis.enqueueBatches = map[string]int{}
|
|
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
|
key, _ := producer.store.RoundActiveKey("full", 1, ch)
|
|
n, _ := producer.store.SizeKey(context.Background(), key)
|
|
if n > 0 {
|
|
redis.enqueueBatches["flow:"+ch] = int(n)
|
|
}
|
|
}
|
|
redis.mu.Lock()
|
|
defer redis.mu.Unlock()
|
|
if len(redis.enqueueBatches) != 2 {
|
|
t.Fatalf("enqueue batches = %+v, want both channels for a full sync", redis.enqueueBatches)
|
|
}
|
|
}
|
|
|
|
func TestProduceScopeCombinesWithSyncDisabledGuard(t *testing.T) {
|
|
producer, redis := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
// Selected but excluded by sync_enabled: scope must not override it.
|
|
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", SyncEnabled: true, Operator: 1},
|
|
})
|
|
|
|
if err := producer.Produce(context.Background(), scopedMessage("scoped", "173")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
redis.enqueueBatches = map[string]int{}
|
|
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
|
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
|
n, _ := producer.store.SizeKey(context.Background(), key)
|
|
if n > 0 {
|
|
redis.enqueueBatches["flow:"+ch] = int(n)
|
|
}
|
|
}
|
|
redis.mu.Lock()
|
|
defer redis.mu.Unlock()
|
|
if len(redis.enqueueBatches) != 0 {
|
|
t.Fatalf("enqueue batches = %+v, want none (selected channel is sync-disabled)", redis.enqueueBatches)
|
|
}
|
|
}
|
|
|
|
func TestProduceScopeSelectingUnknownChannelEnqueuesNothing(t *testing.T) {
|
|
producer, redis := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
})
|
|
|
|
if err := producer.Produce(context.Background(), scopedMessage("scoped", "does-not-exist")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
redis.enqueueBatches = map[string]int{}
|
|
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
|
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
|
n, _ := producer.store.SizeKey(context.Background(), key)
|
|
if n > 0 {
|
|
redis.enqueueBatches["flow:"+ch] = int(n)
|
|
}
|
|
}
|
|
redis.mu.Lock()
|
|
defer redis.mu.Unlock()
|
|
if len(redis.enqueueBatches) != 0 {
|
|
t.Fatalf("enqueue batches = %+v, want none", redis.enqueueBatches)
|
|
}
|
|
}
|
|
|
|
func TestProduceRejectsInvalidScope(t *testing.T) {
|
|
producer, redis := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
})
|
|
|
|
if err := producer.Produce(context.Background(), scopedMessage("bad", "invalid/chan")); err == nil {
|
|
t.Fatal("expected invalid chanKey to be rejected")
|
|
}
|
|
redis.enqueueBatches = map[string]int{}
|
|
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
|
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
|
n, _ := producer.store.SizeKey(context.Background(), key)
|
|
if n > 0 {
|
|
redis.enqueueBatches["flow:"+ch] = int(n)
|
|
}
|
|
}
|
|
redis.mu.Lock()
|
|
defer redis.mu.Unlock()
|
|
if len(redis.enqueueBatches) != 0 {
|
|
t.Fatalf("rejected scope must not enqueue: %+v", redis.enqueueBatches)
|
|
}
|
|
}
|
|
|
|
func TestProduceScopeDeduplicatesChannels(t *testing.T) {
|
|
producer, redis := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
})
|
|
|
|
if err := producer.Produce(context.Background(), scopedMessage("scoped", "217", "217")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
redis.enqueueBatches = map[string]int{}
|
|
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
|
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
|
n, _ := producer.store.SizeKey(context.Background(), key)
|
|
if n > 0 {
|
|
redis.enqueueBatches["flow:"+ch] = int(n)
|
|
}
|
|
}
|
|
redis.mu.Lock()
|
|
defer redis.mu.Unlock()
|
|
if redis.enqueueBatches["flow:217"] != 1 {
|
|
t.Fatalf("enqueue batches = %+v, want 1 task despite a duplicated chanKey", redis.enqueueBatches)
|
|
}
|
|
}
|
|
|
|
func TestScopeDigestIsOrderIndependent(t *testing.T) {
|
|
first := &model.SyncScope{Channels: []string{"173", "217"}}
|
|
second := &model.SyncScope{Channels: []string{"217", "173"}}
|
|
// Normalization sorts the channels, so both requests produce one digest.
|
|
normalizedFirst, err := model.NormalizeScope(first)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
normalizedSecond, err := model.NormalizeScope(second)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if scopeDigest(normalizedFirst) != scopeDigest(normalizedSecond) {
|
|
t.Fatalf("digest depends on channel order: %s vs %s", scopeDigest(normalizedFirst), scopeDigest(normalizedSecond))
|
|
}
|
|
if scopeDigest(nil) != "" {
|
|
t.Fatalf("full sync must have an empty digest")
|
|
}
|
|
}
|
|
|
|
func TestRunAlreadyFinishedTreatsMissingRunAsFirstDelivery(t *testing.T) {
|
|
producer, _ := newScopeProducer(t, []Card{
|
|
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
|
})
|
|
// The fake Redis returns no metadata, which must not be treated as finished.
|
|
finished, err := producer.runAlreadyFinished(context.Background(), "fresh")
|
|
if err != nil {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
if finished {
|
|
t.Fatal("a run without stored metadata must not be reported as finished")
|
|
}
|
|
}
|