From 14442efd282e611de93ef327d9e6fee24da8e6e3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Michal=20Jure=C4=8Dko?= <michal.jurecko@gmail.com>
Date: Wed, 3 Jul 2024 09:11:39 +0200
Subject: [PATCH] tests: Fix writer test, use statistics/collector

---
 .../local/writer/test/testcase/testcase.go    | 36 ++++++++++---------
 1 file changed, 20 insertions(+), 16 deletions(-)

diff --git a/internal/pkg/service/stream/storage/level/local/writer/test/testcase/testcase.go b/internal/pkg/service/stream/storage/level/local/writer/test/testcase/testcase.go
index 9a2622fffd..c32c768fbb 100644
--- a/internal/pkg/service/stream/storage/level/local/writer/test/testcase/testcase.go
+++ b/internal/pkg/service/stream/storage/level/local/writer/test/testcase/testcase.go
@@ -10,12 +10,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/benbjohnson/clock"
 	"github.com/c2h5oh/datasize"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/keboola/keboola-as-code/internal/pkg/log"
+	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/dependencies"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/mapping/table/column"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/compression"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/level/local/volume/disksync"
@@ -23,8 +22,8 @@ import (
 	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/level/local/writer"
 	writerVolume "github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/level/local/writer/volume"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/model"
+	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/statistics/collector"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/test"
-	"github.com/keboola/keboola-as-code/internal/pkg/utils/testhelper"
 	"github.com/keboola/keboola-as-code/internal/pkg/validator"
 )
 
@@ -53,17 +52,19 @@ func (tc *WriterTestCase) Run(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
-	// Setup logger
-	logger := log.NewDebugLogger()
-	logger.ConnectTo(testhelper.VerboseStdout())
+	d, mock := dependencies.NewMockedLocalStorageScope(t)
+	cfg := mock.TestConfig()
+
+	// Start statistics collector
+	events := writer.NewEvents()
+	collector.Start(d, events, cfg.Storage.Statistics.Collector, cfg.NodeID)
 
 	// Open volume
 	opts := []writerVolume.Option{writerVolume.WithWatchDrainFile(false)}
-	clk := clock.New()
-	now := clk.Now()
+	now := d.Clock().Now()
 	volPath := t.TempDir()
 	spec := volume.Spec{NodeID: "my-node", Path: volPath, Type: "hdd", Label: "1"}
-	vol, err := writerVolume.Open(ctx, logger, clk, writer.NewEvents(), writer.NewConfig(), spec, opts...)
+	vol, err := writerVolume.Open(ctx, d.Logger(), d.Clock(), events, cfg.Storage.Level.Local.Writer, spec, opts...)
 	require.NoError(t, err)
 
 	// Create a test slice
@@ -122,17 +123,13 @@ func (tc *WriterTestCase) Run(t *testing.T) {
 
 	// Close the writer
 	require.NoError(t, w.Close(ctx))
-	assert.Equal(t, uint64(rowsCount), w.CompletedWrites())
-	assert.NotEmpty(t, w.CompressedSize())
-	assert.NotEmpty(t, w.UncompressedSize())
 
 	// Close volume
 	assert.NoError(t, vol.Close(ctx))
 
-	// Check compressed size
-	stat, err := os.Stat(filePath)
+	// Get file size
+	fileStat, err := os.Stat(filePath)
 	require.NoError(t, err)
-	assert.Equal(t, int64(w.CompressedSize().Bytes()), stat.Size(), "compressed file size doesn't match")
 
 	// Open file
 	f, err := os.OpenFile(filePath, os.O_RDONLY, 0o640)
@@ -147,13 +144,20 @@ func (tc *WriterTestCase) Run(t *testing.T) {
 	// Read file content
 	content, err := io.ReadAll(reader)
 	require.NoError(t, err)
-	assert.Equal(t, int(w.UncompressedSize().Bytes()), len(content), "uncompressed file size doesn't match")
 
 	// Check written data
 	tc.Validator(t, string(content))
 
 	// Close file
 	assert.NoError(t, f.Close())
+
+	// Check statistics
+	sliceStats, err := d.StatisticsRepository().SliceStats(ctx, slice.SliceKey)
+	if assert.NoError(t, err) {
+		assert.Equal(t, int(sliceStats.Total.RecordsCount), rowsCount, "records count doesn't match")
+		assert.Equal(t, int64(sliceStats.Total.CompressedSize.Bytes()), fileStat.Size(), "compressed file size doesn't match")
+		assert.Equal(t, int(sliceStats.Total.UncompressedSize.Bytes()), len(content), "uncompressed file size doesn't match")
+	}
 }
 
 func (tc *WriterTestCase) newSlice(t *testing.T, volume *writerVolume.Volume) *model.Slice {