Make our own low-level driver for testing

.golangci.yml

@@ -36,6 +36,7 @@ linters-settings:
         files:
           - $all
           - "!**/internal/bson/*_test.go"
+          - "!**/internal/driver/*.go"
           - "!**/internal/util/testutil/*.go"
           - "!**/internal/wire/*.go"
         deny:
@@ -44,6 +45,7 @@ linters-settings:
         files:
           - $all
           - "!**/internal/bson/*.go"
+          - "!**/internal/driver/*.go"
         deny:
           - pkg: github.com/cristalhq/bson
           - pkg: github.com/cristalhq/bson/bsonproto

internal/driver/driver.go

@@ -37,6 +37,8 @@
 	r *bufio.Reader
 	w *bufio.Writer
 	l *slog.Logger
+
+	lastRequestID int32
 }
 
 // Connect creates a new connection for the given MongoDB URI and logger.
@@ -120,7 +122,7 @@
 	c.l.Debug(
 		fmt.Sprintf("<<<\n%s", body.String()),
 		slog.Int("length", int(header.MessageLength)),
-		slog.Int("id", int(header.ResponseTo)),
+		slog.Int("id", int(header.RequestID)),
 		slog.Int("response_to", int(header.ResponseTo)),
 		slog.String("opcode", header.OpCode.String()),
 	)
@@ -133,7 +135,7 @@
 	c.l.Debug(
 		fmt.Sprintf(">>>\n%s", body.String()),
 		slog.Int("length", int(header.MessageLength)),
-		slog.Int("id", int(header.ResponseTo)),
+		slog.Int("id", int(header.RequestID)),
 		slog.Int("response_to", int(header.ResponseTo)),
 		slog.String("opcode", header.OpCode.String()),
 	)
@@ -165,7 +167,71 @@
 }
 
 // Request sends the given request to the connection and returns the response.
+// If header MessageLength or RequestID is not specified, it assings the proper values.
+// For header.OpCode the wire.OpCodeMsg is used as default.
 func (c *Conn) Request(ctx context.Context, header *wire.MsgHeader, body wire.MsgBody) (*wire.MsgHeader, wire.MsgBody, error) {
-	// TODO https://github.com/FerretDB/FerretDB/issues/4146
-	panic("not implemented")
+	if header.MessageLength == 0 {
+		msgBin, err := body.MarshalBinary()
+		if err != nil {
+			return nil, nil, lazyerrors.Error(err)
+		}
+
+		header.MessageLength = int32(len(msgBin) + wire.MsgHeaderLen)
+	}
+
+	if header.OpCode == 0 {
+		header.OpCode = wire.OpCodeMsg
+	}
+
+	if header.RequestID == 0 {
+		header.RequestID = c.nextRequestID()
+	}
+	c.lastRequestID = header.RequestID
+
+	if header.ResponseTo != 0 {
+		return nil, nil, lazyerrors.Errorf("response_to is not allowed")
+	}
+
+	if m, ok := body.(*wire.OpMsg); ok {
+		if m.Flags != 0 {
+			return nil, nil, lazyerrors.Errorf("unsupported request flags %s", m.Flags)
+		}
+	}
+
+	if err := c.Write(header, body); err != nil {
+		return nil, nil, lazyerrors.Error(err)
+	}
+
+	resHeader, resBody, err := c.Read()
+	if err != nil {
+		return nil, nil, lazyerrors.Error(err)
+	}
+
+	if resHeader.ResponseTo != header.RequestID {
+		c.l.Error(
+			fmt.Sprintf("response_to not equal to request_id"),
+			slog.Int("request_id", int(header.RequestID)),
+			slog.Int("response_id", int(resHeader.RequestID)),
+			slog.Int("response_to", int(resHeader.ResponseTo)),
+		)
+		return nil, nil, lazyerrors.Errorf(
+			"response_to not equal to request_id (response_to=%d; expected=%d)",
+			resHeader.ResponseTo,
+			header.RequestID,
+		)
+	}
+
+	if m, ok := resBody.(*wire.OpMsg); ok {
+		if m.Flags != 0 {
+			return nil, nil, lazyerrors.Errorf("unsupported response flags %s", m.Flags)
+		}
+	}
+
+	return resHeader, resBody, nil
+}
+
+// nextRequestID returns the incremented value of last recorded request header ID from `Request` function.
+func (c *Conn) nextRequestID() int32 {
+	c.lastRequestID += 1
+	return c.lastRequestID
 }

internal/driver/driver_test.go

@@ -17,9 +17,14 @@
 import (
 	"testing"
 
+	"github.com/cristalhq/bson/bsonproto"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/FerretDB/FerretDB/internal/bson"
+	"github.com/FerretDB/FerretDB/internal/util/must"
 	"github.com/FerretDB/FerretDB/internal/util/testutil"
+	"github.com/FerretDB/FerretDB/internal/wire"
 )
 
 func TestDriver(t *testing.T) {
@@ -33,6 +38,147 @@
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, c.Close()) })
 
-	// TODO https://github.com/FerretDB/FerretDB/issues/4146
-	_ = c
+	dbName := "TestDriver"
+
+	var lsid bson.Binary
+
+	lsid = bson.Binary{
+		B: []byte{
+			0xa3, 0x19, 0xf2, 0xb4, 0xa1, 0x75, 0x40, 0xc7,
+			0xb8, 0xe7, 0xa3, 0xa3, 0x2e, 0xc2, 0x56, 0xbe,
+		},
+		Subtype: bsonproto.BinaryUUID,
+	}
+
+	t.Run("Drop", func(t *testing.T) {
+		doc := must.NotFail(bson.NewDocument(
+			"dropDatabase", int32(1),
+			"lsid", must.NotFail(bson.NewDocument("id", lsid)),
+			"$db", dbName,
+		))
+
+		body, err := wire.NewOpMsg(must.NotFail(doc.Encode()))
+		require.NoError(t, err)
+
+		_, _, err = c.Request(ctx, new(wire.MsgHeader), body)
+		require.NoError(t, err)
+	})
+
+	insertDocs := must.NotFail(bson.NewArray(
+		must.NotFail(bson.NewDocument("w", int32(2), "v", int32(1), "_id", int32(0))),
+		must.NotFail(bson.NewDocument("v", int32(2), "_id", int32(1))),
+		must.NotFail(bson.NewDocument("v", int32(3), "_id", int32(2))),
+	))
+
+	t.Run("Insert", func(t *testing.T) {
+		insertCmd := must.NotFail(bson.NewDocument(
+			"insert", "values",
+			"documents", insertDocs,
+			"ordered", true,
+			"lsid", must.NotFail(bson.NewDocument("id", lsid)),
+			"$db", dbName,
+		))
+
+		body, err := wire.NewOpMsg(must.NotFail(insertCmd.Encode()))
+		require.NoError(t, err)
+
+		resHeader, _, err := c.Request(ctx, new(wire.MsgHeader), body)
+		require.NoError(t, err)
+
+		assert.NotZero(t, resHeader.RequestID)
+	})
+
+	var cursorID int64
+
+	t.Run("Find", func(t *testing.T) {
+		findCmd := must.NotFail(bson.NewDocument(
+			"find", "values",
+			"filter", must.NotFail(bson.NewDocument()),
+			"sort", must.NotFail(bson.NewDocument("_id", int32(1))),
+			"lsid", must.NotFail(bson.NewDocument("id", lsid)),
+			"batchSize", int32(1),
+			"$db", dbName,
+		))
+
+		body, err := wire.NewOpMsg(must.NotFail(findCmd.Encode()))
+		require.NoError(t, err)
+
+		resHeader, resBody, err := c.Request(ctx, new(wire.MsgHeader), body)
+		require.NoError(t, err)
+		assert.NotZero(t, resHeader.RequestID)
+
+		resMsg, err := must.NotFail(resBody.(*wire.OpMsg).RawDocument()).Decode()
+		require.NoError(t, err)
+
+		cursor, err := resMsg.Get("cursor").(bson.RawDocument).Decode()
+		require.NoError(t, err)
+
+		firstBatch := cursor.Get("firstBatch").(bson.RawArray)
+		cursorID = cursor.Get("id").(int64)
+
+		expectedDocs := must.NotFail(bson.NewArray(
+			must.NotFail(bson.NewDocument("_id", int32(0), "w", int32(2), "v", int32(1))),
+		))
+
+		testutil.AssertEqual(t, must.NotFail(expectedDocs.Convert()), must.NotFail(firstBatch.Convert()))
+		require.NotZero(t, cursorID)
+	})
+
+	getMoreCmd := must.NotFail(bson.NewDocument(
+		"getMore", cursorID,
+		"collection", "values",
+		"lsid", must.NotFail(bson.NewDocument("id", lsid)),
+		"batchSize", int32(1),
+		"$db", dbName,
+	))
+
+	t.Run("GetMore", func(t *testing.T) {
+		for i := 0; i < 2; i++ {
+			body, err := wire.NewOpMsg(must.NotFail(getMoreCmd.Encode()))
+			require.NoError(t, err)
+
+			resHeader, resBody, err := c.Request(ctx, new(wire.MsgHeader), body)
+			require.NoError(t, err)
+			assert.NotZero(t, resHeader.RequestID)
+
+			resMsg, err := must.NotFail(resBody.(*wire.OpMsg).RawDocument()).Decode()
+			require.NoError(t, err)
+
+			cursor, err := resMsg.Get("cursor").(bson.RawDocument).Decode()
+			require.NoError(t, err)
+
+			nextBatch := cursor.Get("nextBatch").(bson.RawArray)
+			newCursorID := cursor.Get("id").(int64)
+
+			expectedDocs := must.NotFail(bson.NewArray(
+				must.NotFail(bson.NewDocument("_id", int32(i+1), "v", int32(i+2))),
+			))
+
+			testutil.AssertEqual(t, must.NotFail(expectedDocs.Convert()), must.NotFail(nextBatch.Convert()))
+			assert.Equal(t, cursorID, newCursorID)
+		}
+	})
+
+	t.Run("GetMoreEmpty", func(t *testing.T) {
+		body, err := wire.NewOpMsg(must.NotFail(getMoreCmd.Encode()))
+		require.NoError(t, err)
+
+		resHeader, resBody, err := c.Request(ctx, new(wire.MsgHeader), body)
+		require.NoError(t, err)
+		assert.NotZero(t, resHeader.RequestID)
+
+		resMsg, err := must.NotFail(resBody.(*wire.OpMsg).RawDocument()).Decode()
+		require.NoError(t, err)
+
+		cursor, err := resMsg.Get("cursor").(bson.RawDocument).Decode()
+		require.NoError(t, err)
+
+		nextBatch := cursor.Get("nextBatch").(bson.RawArray)
+		newCursorID := cursor.Get("id").(int64)
+
+		expectedDocs := must.NotFail(bson.NewArray())
+
+		testutil.AssertEqual(t, must.NotFail(expectedDocs.Convert()), must.NotFail(nextBatch.Convert()))
+		assert.Zero(t, newCursorID)
+	})
 }