api: add ability to mock connections in tests

Create a mock implementations `MockRequest`, `MockResponse` and
`MockDoer`.
The last one allows to mock not the full `Connection`, but its
part -- a structurem that implements new `Doer` interface
(a `Do` function).

Also added missing comments, all the changes are recorded in
the `CHANGELOG` and `README` files. Added new tests and
examples.

So this entity is easier to implement and it is enough to
mock tests that require working `Connection`.
All new mock structs and an example for `MockDoer` usege
are added to the `test_helpers` package.

Added new structs `MockDoer`, `MockRequest` to `test_helpers`.
Renamed `StrangerResponse` to `MockResponse`.

Closes #237

Author: DerekBum

CHANGELOG.md

@@ -28,6 +28,11 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
   in requests instead of their IDs.
 - `GetSchema` function to get the actual schema (#7)
 - Support connection via an existing socket fd (#321)
+- Ability to mock connections for tests (#237). Added new structs `MockDoer`, 
+  `MockRequest` to `test_helpers`.
+- `Response` method added to the `Request` interface (#237)
+- `Header` struct for the response header (#237). It can be accessed via 
+  `Header()` method of the `Response` interface.
 
 ### Changed
 
@@ -67,6 +72,21 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
   it (#321)
 - Rename `pool.GetPoolInfo` to `pool.GetInfo`. Change return type to
   `map[string]ConnectionInfo` (#321)
+- All responses are now implementations of an `Response` interface (#237).
+  `SelectResponse`, `ExecuteResponse`, `PrepareResponse`, `PushResponse` are part 
+  of a public API. `Pos()`, `MetaData()`, `SQLInfo()` methods created for them 
+  to get specific info.
+  Special types of responses are used with special requests.
+- `IsPush()` method is added to the response iterator (#237). It returns 
+  the information if the current response is a `PushResponse`. 
+  `PushCode` constant is removed.
+- `Future` constructors now accept `Request` as their argument (#237). 
+  Method `Get` now returns response data. To get the actual response new method 
+  added `GetResponse`. Methods `AppendPush` and `SetResponse` accepts 
+  response `Header` and data as their arguments.
+- All deprecated asynchronous requests operations on a `Connector` return 
+  response data instead of an actual responses (#237)
+- Renamed `StrangerResponse` to `MockResponse` (#237)
 
 ### Deprecated
 

README.md

@@ -128,13 +128,13 @@ func main() {
 	if err != nil {
 		fmt.Println("Connection refused:", err)
 	}
-	resp, err := conn.Do(tarantool.NewInsertRequest(999).
+	data, err := conn.Do(tarantool.NewInsertRequest(999).
 		Tuple([]interface{}{99999, "BB"}),
 	).Get()
 	if err != nil {
 		fmt.Println("Error", err)
-		fmt.Println("Code", resp.Code)
 	}
+	fmt.Printf("Data: %v", data)
 }
 ```
 
@@ -201,12 +201,18 @@ The subpackage has been deleted. You could use `pool` instead.
   unique string ID, which allows them to be distinguished.
 * `pool.GetPoolInfo` has been renamed to `pool.GetInfo`. Return type has been changed
   to `map[string]ConnectionInfo`.
+* All deprecated asynchronous requests operations on a `Pooler` return response 
+  data instead of an actual responses.
 
 #### crud package
 
 * `crud` operations `Timeout` option has `crud.OptFloat64` type
   instead of `crud.OptUint`.
 
+#### test_helpers package
+
+Renamed `StrangerResponse` to `MockResponse`.
+
 #### msgpack.v5
 
 Most function names and argument types in `msgpack.v5` and `msgpack.v2`
@@ -241,7 +247,10 @@ of the requests is an array instead of array of arrays.
 
 #### IPROTO constants
 
-IPROTO constants have been moved to a separate package [go-iproto](https://github.com/tarantool/go-iproto).
+* IPROTO constants have been moved to a separate package [go-iproto](https://github.com/tarantool/go-iproto).
+* `PushCode` constant is removed. To get information, if the current response is 
+  a push response, new `IsPush()` method of the response iterator could be used 
+  instead.
 
 #### Request changes
 
@@ -254,6 +263,34 @@ longer accept `ops` argument (operations) as an `interface{}`. `*Operations`
 needs to be passed instead.
 * `UpdateRequest` and `UpsertRequest` structs no longer accept `interface{}`
 for an `ops` field. `*Operations` needs to be used instead.
+* `Response` method added to the `Request` interface.
+
+#### Response changes
+
+* New interface `Response` added.
+* For each request type, a different response type is created. They all 
+  implement a `Response` interface. `SelectResponse`, `PrepareResponse`, 
+  `ExecuteResponse`, `PushResponse` are a part of a public API. 
+  `Pos()`, `MetaData()`, `SQLInfo()` methods created for them to get specific info.
+  Special types of responses are used with special requests.
+* Response header stored in a new `Header` struct. It could be accessed via 
+  `Header()` method.
+* Method `IsPush()` is added to the `ResponseIterator` interface.
+  It returns true if the current response is a push response.
+
+#### Future changes
+
+* `Future` constructors now accept `Request` as their argument.
+* Method `Get` now returns response data instead of the actual response.
+* New method `GetResponse` added to get an actual response.
+* Methods `AppendPush` and `SetResponse` accepts response `Header` and data 
+  as their arguments.
+
+#### Connector changes
+
+* All deprecated asynchronous requests operations on a `Connector` return 
+  response data instead of an actual responses.
+* New interface `Doer` is added as a child-interface instead of a `Do` method.
 
 #### Connect function
 
@@ -270,7 +307,7 @@ for creating a connection are now stored in corresponding `Dialer`, not in `Opts
 #### Connection schema
 
 * Removed `Schema` field from the `Connection` struct. Instead, new
-`GetSchema(Connector)` function was added to get the actual connection
+`GetSchema(Doer)` function was added to get the actual connection
 schema on demand.
 * `OverrideSchema(*Schema)` method replaced with the `SetSchema(Schema)`.
 

connection.go

@@ -830,7 +830,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 				conn.opts.Logger.Report(LogWatchEventReadFailed, conn, err)
 			}
 			continue
-		} else if header.Code == PushCode {
+		} else if iproto.Type(header.Code) == iproto.IPROTO_CHUNK {
 			if fut = conn.peekFuture(header.RequestId); fut != nil {
 				fut.AppendPush(header, &buf)
 			}
@@ -874,8 +874,7 @@ func (conn *Connection) eventer(events <-chan connWatchEvent) {
 
 func (conn *Connection) newFuture(req Request) (fut *Future) {
 	ctx := req.Ctx()
-	fut = NewFuture()
-	fut.SetRequest(req)
+	fut = NewFuture(req)
 	if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
 		select {
 		case conn.rlimit <- struct{}{}:
@@ -1204,7 +1203,7 @@ func (conn *Connection) nextRequestId(context bool) (requestId uint32) {
 func (conn *Connection) Do(req Request) *Future {
 	if connectedReq, ok := req.(ConnectedRequest); ok {
 		if connectedReq.Conn() != conn {
-			fut := NewFuture()
+			fut := NewFuture(req)
 			fut.SetError(errUnknownRequest)
 			return fut
 		}

connector.go

@@ -2,14 +2,20 @@ package tarantool
 
 import "time"
 
+// Doer if an interface that performs requests asynchronously.
+type Doer interface {
+	// Do performs a request asynchronously.
+	Do(req Request) (fut *Future)
+}
+
 type Connector interface {
+	Doer
 	ConnectedNow() bool
 	Close() error
 	ConfiguredTimeout() time.Duration
 	NewPrepared(expr string) (*Prepared, error)
 	NewStream() (*Stream, error)
 	NewWatcher(key string, callback WatchCallback) (Watcher, error)
-	Do(req Request) (fut *Future)
 
 	// Deprecated: the method will be removed in the next major version,
 	// use a PingRequest object + Do() instead.

const.go

@@ -9,6 +9,6 @@ const (
 )
 
 const (
-	OkCode   = uint32(iproto.IPROTO_OK)
-	PushCode = uint32(iproto.IPROTO_CHUNK)
+	// OkCode is an iproto code for a successful request or command.
+	OkCode = int(iproto.IPROTO_OK)
 )

crud/select.go

@@ -2,11 +2,9 @@ package crud
 
 import (
 	"context"
-	"io"
-
-	"github.com/vmihailenco/msgpack/v5"
 
 	"github.com/tarantool/go-tarantool/v2"
+	"github.com/vmihailenco/msgpack/v5"
 )
 
 // SelectOpts describes options for `crud.select` method.
@@ -134,9 +132,3 @@ func (req SelectRequest) Context(ctx context.Context) SelectRequest {
 
 	return req
 }
-
-// Response creates a response for the SelectRequest.
-func (req SelectRequest) Response(header tarantool.Header,
-	body io.Reader) (tarantool.Response, error) {
-	return req.impl.Response(header, body)
-}

dial.go

@@ -398,16 +398,17 @@ func identify(w writeFlusher, r io.Reader) (ProtocolInfo, error) {
 		return info, err
 	}
 
-	resp, err := readResponse(r)
+	resp, err := readResponse(r, req)
 	if err != nil {
+		if resp != nil &&
+			iproto.Error(resp.Header().Code) == iproto.ER_UNKNOWN_REQUEST_TYPE {
+			// IPROTO_ID requests are not supported by server.
+			return info, nil
+		}
 		return info, err
 	}
 	data, err := resp.Decode()
 	if err != nil {
-		if iproto.Error(resp.Header().Code) == iproto.ER_UNKNOWN_REQUEST_TYPE {
-			// IPROTO_ID requests are not supported by server.
-			return info, nil
-		}
 		return info, err
 	}
 
@@ -477,7 +478,7 @@ func authenticate(c Conn, auth Auth, user string, pass string, salt string) erro
 	if err = writeRequest(c, req); err != nil {
 		return err
 	}
-	if _, err = readResponse(c); err != nil {
+	if _, err = readResponse(c, req); err != nil {
 		return err
 	}
 	return nil
@@ -501,19 +502,31 @@ func writeRequest(w writeFlusher, req Request) error {
 }
 
 // readResponse reads a response from the reader.
-func readResponse(r io.Reader) (Response, error) {
+func readResponse(r io.Reader, req Request) (Response, error) {
 	var lenbuf [packetLengthBytes]byte
 
 	respBytes, err := read(r, lenbuf[:])
 	if err != nil {
-		return &BaseResponse{}, fmt.Errorf("read error: %w", err)
+		return nil, fmt.Errorf("read error: %w", err)
 	}
 
 	buf := smallBuf{b: respBytes}
 	header, err := decodeHeader(msgpack.NewDecoder(&smallBuf{}), &buf)
-	resp := &BaseResponse{header: header, buf: buf}
 	if err != nil {
-		return resp, fmt.Errorf("decode response header error: %w", err)
+		return nil, fmt.Errorf("decode response header error: %w", err)
+	}
+	resp, err := req.Response(header, &buf)
+	if err != nil {
+		return nil, fmt.Errorf("creating response error: %w", err)
+	}
+	_, err = resp.Decode()
+	if err != nil {
+		switch err.(type) {
+		case Error:
+			return resp, err
+		default:
+			return resp, fmt.Errorf("decode response body error: %w", err)
+		}
 	}
 	return resp, nil
 }

example_test.go

@@ -198,16 +198,34 @@ func ExampleSelectRequest() {
 	}
 
 	key := []interface{}{uint(1111)}
-	data, err := conn.Do(tarantool.NewSelectRequest(617).
+	resp, err := conn.Do(tarantool.NewSelectRequest(617).
 		Limit(100).
 		Iterator(tarantool.IterEq).
 		Key(key),
-	).Get()
+	).GetResponse()
 
 	if err != nil {
 		fmt.Printf("error in select is %v", err)
 		return
 	}
+	selResp, ok := resp.(*tarantool.SelectResponse)
+	if !ok {
+		fmt.Print("wrong response type")
+		return
+	}
+
+	pos, err := selResp.Pos()
+	if err != nil {
+		fmt.Printf("error in Pos: %v", err)
+		return
+	}
+	fmt.Printf("pos for Select is %v\n", pos)
+
+	data, err := resp.Decode()
+	if err != nil {
+		fmt.Printf("error while decoding: %v", err)
+		return
+	}
 	fmt.Printf("response is %#v\n", data)
 
 	var res []Tuple
@@ -224,6 +242,7 @@ func ExampleSelectRequest() {
 	fmt.Printf("response is %v\n", res)
 
 	// Output:
+	// pos for Select is []
 	// response is []interface {}{[]interface {}{0x457, "hello", "world"}}
 	// response is [{{} 1111 hello world}]
 }
@@ -567,17 +586,21 @@ func ExampleExecuteRequest() {
 	resp, err := conn.Do(req).GetResponse()
 	fmt.Println("Execute")
 	fmt.Println("Error", err)
+
 	data, err := resp.Decode()
 	fmt.Println("Error", err)
 	fmt.Println("Data", data)
+
 	exResp, ok := resp.(*tarantool.ExecuteResponse)
 	if !ok {
 		fmt.Printf("wrong response type")
 		return
 	}
+
 	metaData, err := exResp.MetaData()
 	fmt.Println("MetaData", metaData)
 	fmt.Println("Error", err)
+
 	sqlInfo, err := exResp.SQLInfo()
 	fmt.Println("SQL Info", sqlInfo)
 	fmt.Println("Error", err)

future.go

@@ -123,12 +123,26 @@ func (it *asyncResponseIterator) nextResponse() (resp Response) {
 	return resp
 }
 
-// NewFuture creates a new empty Future.
-func NewFuture() (fut *Future) {
+// PushResponse is used for push requests for the Future.
+type PushResponse struct {
+	baseResponse
+}
+
+func createPushResponse(header Header, body io.Reader) (Response, error) {
+	resp, err := createBaseResponse(header, body)
+	if err != nil {
+		return nil, err
+	}
+	return &PushResponse{resp}, nil
+}
+
+// NewFuture creates a new empty Future for a given Request.
+func NewFuture(req Request) (fut *Future) {
 	fut = &Future{}
 	fut.ready = make(chan struct{}, 1000000000)
 	fut.done = make(chan struct{})
 	fut.pushes = make([]Response, 0)
+	fut.req = req
 	return fut
 }
 
@@ -154,11 +168,6 @@ func (fut *Future) AppendPush(header Header, body io.Reader) error {
 	return nil
 }
 
-// SetRequest sets a request, for which the future was created.
-func (fut *Future) SetRequest(req Request) {
-	fut.req = req
-}
-
 // SetResponse sets a response for the future and finishes the future.
 func (fut *Future) SetResponse(header Header, body io.Reader) error {
 	fut.mutex.Lock()