<commit_before>// SPDX-License-Identifier: Apache-2.0
// Copyright Authors of Cilium

package stream_test

import (
	"context"
	"errors"
	"testing"
	"time"

	. "github.com/cilium/cilium/pkg/stream"
)

func TestMap(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	double := func(x int) int { return x * 2 }

	// 1. mapping a non-empty source
	{
		src := Range(0, 5)
		src = Map(src, double)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 1", err)
		assertSlice(t, "case 1", []int{0, 2, 4, 6, 8}, result)
	}

	// 2. mapping an empty source
	{
		src := Map(Empty[int](), double)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 2", err)
		assertSlice(t, "case 2", []int{}, result)
	}

	// 3. cancelled context
	checkCancelled(t, "case 3", Map(Range(0, 100), double))
}

func TestFilter(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	isOdd := func(x int) bool { return x%2 != 0 }

	// 1. filtering a non-empty source
	{
		src := Range(0, 5)
		src = Filter(src, isOdd)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 1", err)
		assertSlice(t, "case 1", []int{1, 3}, result)
	}

	// 2. filtering an empty source
	{
		src := Filter(Empty[int](), isOdd)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 2", err)
		assertSlice(t, "case 2", []int{}, result)
	}

	// 3. cancelled context
	checkCancelled(t, "case 3", Filter(Range(0, 100), isOdd))
}

func TestReduce(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sum := func(result, x int) int { return result + x }

	// 1. Reducing a non-empty source
	{
		src := Range(0, 5)
		src = Reduce(src, 0, sum)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 1", err)
		assertSlice(t, "case 1", []int{0 + 0 + 1 + 2 + 3 + 4}, result)
	}

	// 2. Reducing an empty source
	{
		src := Reduce(Empty[int](), 0, sum)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 2", err)
		assertSlice(t, "case 2", []int{0}, result)
	}

	// 3. cancelled context
	checkCancelled(t, "case 3", Reduce(Range(0, 100), 0, sum))
}

func TestThrottle(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	waitMillis := 20
	go func() {
		time.Sleep(time.Duration(waitMillis) * time.Millisecond)
		cancel()
	}()

	ratePerSecond := 2000.0
	values, err := ToSlice(ctx, Throttle(Range(0, 100000), ratePerSecond, 1))
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("expected Canceled error, got %s", err)
	}

	expectedLen := int(float64(waitMillis) / 1000.0 * ratePerSecond)

	lenDiff := len(values) - expectedLen
	if lenDiff < 0 {
		lenDiff *= -1
	}
	// Check that we're within 20%
	if lenDiff > expectedLen/5 {
		t.Fatalf("expected ~%d values, got %d, diff: %d", expectedLen, len(values), lenDiff)<commit_msg>stream: operators_test: remove TestThrottle OS timer dependency<commit_after>// SPDX-License-Identifier: Apache-2.0
// Copyright Authors of Cilium

package stream_test

import (
	"context"
	"errors"
	"testing"
	"time"

	. "github.com/cilium/cilium/pkg/stream"
)

func TestMap(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	double := func(x int) int { return x * 2 }

	// 1. mapping a non-empty source
	{
		src := Range(0, 5)
		src = Map(src, double)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 1", err)
		assertSlice(t, "case 1", []int{0, 2, 4, 6, 8}, result)
	}

	// 2. mapping an empty source
	{
		src := Map(Empty[int](), double)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 2", err)
		assertSlice(t, "case 2", []int{}, result)
	}

	// 3. cancelled context
	checkCancelled(t, "case 3", Map(Range(0, 100), double))
}

func TestFilter(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	isOdd := func(x int) bool { return x%2 != 0 }

	// 1. filtering a non-empty source
	{
		src := Range(0, 5)
		src = Filter(src, isOdd)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 1", err)
		assertSlice(t, "case 1", []int{1, 3}, result)
	}

	// 2. filtering an empty source
	{
		src := Filter(Empty[int](), isOdd)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 2", err)
		assertSlice(t, "case 2", []int{}, result)
	}

	// 3. cancelled context
	checkCancelled(t, "case 3", Filter(Range(0, 100), isOdd))
}

func TestReduce(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sum := func(result, x int) int { return result + x }

	// 1. Reducing a non-empty source
	{
		src := Range(0, 5)
		src = Reduce(src, 0, sum)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 1", err)
		assertSlice(t, "case 1", []int{0 + 0 + 1 + 2 + 3 + 4}, result)
	}

	// 2. Reducing an empty source
	{
		src := Reduce(Empty[int](), 0, sum)
		result, err := ToSlice(ctx, src)
		assertNil(t, "case 2", err)
		assertSlice(t, "case 2", []int{0}, result)
	}

	// 3. cancelled context
	checkCancelled(t, "case 3", Reduce(Range(0, 100), 0, sum))
}

func TestThrottle(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	item, err := First(ctx, Throttle(Range(42, 1000), 100.0, 1))
	assertNil(t, "First", err)

	if item != 42 {
		t.Fatalf("expected 42, got %d", item)
{'fraction_non_alphanumeric': 0.17812142038946163, 'fraction_numerical': 0.03245513554791905, 'mean_word_length': 3.71981981981982, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5259079', 'n_tokens_mistral': 2234, 'n_tokens_neox': 2007, 'n_words': 564}