flow lints: add testifylint & wastedassign (#1122)

Also fix %v/%w flagged by errorlint,
but don't enable for now since it had more complaints

flow/.golangci.yml

@@ -10,21 +10,23 @@ linters:
     - gofumpt
     - gosec
     - gosimple
+    - ineffassign
+    - lll
     - misspell
     - nakedret
     - nolintlint
+    - prealloc
     - staticcheck
     - stylecheck
     - sqlclosecheck
+    - testifylint
+    - thelper
     - unconvert
     - unparam
-    - whitespace
-    - prealloc
-    - thelper
-    - ineffassign
     - unparam
     - unused
-    - lll
+    - wastedassign
+    - whitespace
 linters-settings:
   stylecheck:
     checks:

flow/connectors/postgres/client.go

@@ -246,7 +246,7 @@ func (c *PostgresConnector) checkSlotAndPublication(slot string, publication str
 // If slotName input is empty, all slot info rows are returned - this is for UI.
 // Else, only the row pertaining to that slotName will be returned.
 func (c *PostgresConnector) GetSlotInfo(slotName string) ([]*protos.SlotInfo, error) {
-	whereClause := ""
+	var whereClause string
 	if slotName != "" {
 		whereClause = fmt.Sprintf(" WHERE slot_name = %s", QuoteLiteral(slotName))
 	} else {

flow/connectors/postgres/postgres_schema_delta_test.go

@@ -250,9 +250,9 @@ func TestPostgresSchemaDeltaTestSuite(t *testing.T) {
 		err = teardownTx.Commit(context.Background())
 		require.NoError(s.t, err)
 
-		require.True(s.t, s.connector.ConnectionActive() == nil)
+		require.NoError(s.t, s.connector.ConnectionActive())
 		err = s.connector.Close()
 		require.NoError(s.t, err)
-		require.False(s.t, s.connector.ConnectionActive() == nil)
+		require.Error(s.t, s.connector.ConnectionActive())
 	})
 }

flow/connectors/postgres/qrep_partition_test.go

@@ -184,7 +184,7 @@ func TestGetQRepPartitions(t *testing.T) {
 			// for now, but ideally we should check that the partition ranges
 			// are correct as well.
 			if tc.expectedNumPartitions != 0 {
-				assert.Equal(t, tc.expectedNumPartitions, len(got))
+				assert.Len(t, got, tc.expectedNumPartitions)
 				return
 			}
 

flow/connectors/utils/cdc_records/cdc_records_storage_test.go

@@ -83,7 +83,7 @@ func TestSingleRecord(t *testing.T) {
 	err := cdcRecordsStore.Set(key, rec)
 	require.NoError(t, err)
 	// should not spill into DB
-	require.Equal(t, 1, len(cdcRecordsStore.inMemoryRecords))
+	require.Len(t, cdcRecordsStore.inMemoryRecords, 1)
 	require.Nil(t, cdcRecordsStore.pebbleDB)
 
 	reck, ok, err := cdcRecordsStore.Get(key)
@@ -100,11 +100,11 @@ func TestRecordsTillSpill(t *testing.T) {
 	cdcRecordsStore.numRecordsSwitchThreshold = 10
 
 	// add records upto set limit
-	for i := 0; i < 10; i++ {
+	for i := 1; i <= 10; i++ {
 		key, rec := genKeyAndRec(t)
 		err := cdcRecordsStore.Set(key, rec)
 		require.NoError(t, err)
-		require.Equal(t, i+1, len(cdcRecordsStore.inMemoryRecords))
+		require.Len(t, cdcRecordsStore.inMemoryRecords, i)
 		require.Nil(t, cdcRecordsStore.pebbleDB)
 	}
 

flow/e2e/s3/qrep_flow_s3_test.go

@@ -143,7 +143,7 @@ func (s PeerFlowE2ETestSuiteS3) Test_Complete_QRep_Flow_S3() {
 
 	require.NoError(s.t, err)
 
-	require.Equal(s.t, 1, len(files))
+	require.Len(s.t, files, 1)
 }
 
 func (s PeerFlowE2ETestSuiteS3) Test_Complete_QRep_Flow_S3_CTID() {
@@ -192,5 +192,5 @@ func (s PeerFlowE2ETestSuiteS3) Test_Complete_QRep_Flow_S3_CTID() {
 
 	require.NoError(s.t, err)
 
-	require.Equal(s.t, 10, len(files))
+	require.Len(s.t, files, 10)
 }

flow/e2e/snowflake/peer_flow_sf_test.go

@@ -754,7 +754,7 @@ func (s PeerFlowE2ETestSuiteSF) Test_Types_SF() {
 	require.NoError(s.t, err)
 
 	// Make sure that there are no nulls
-	require.Equal(s.t, noNulls, true)
+	require.True(s.t, noNulls)
 }
 
 func (s PeerFlowE2ETestSuiteSF) Test_Multi_Table_SF() {
@@ -1259,9 +1259,9 @@ func (s PeerFlowE2ETestSuiteSF) Test_Column_Exclusion() {
 	require.NoError(s.t, err)
 
 	for _, field := range sfRows.Schema.Fields {
-		require.NotEqual(s.t, field.Name, "c2")
+		require.NotEqual(s.t, "c2", field.Name)
 	}
-	require.Equal(s.t, 5, len(sfRows.Schema.Fields))
+	require.Len(s.t, sfRows.Schema.Fields, 5)
 }
 
 func (s PeerFlowE2ETestSuiteSF) Test_Soft_Delete_Basic() {

flow/model/conversion_avro.go

@@ -42,7 +42,7 @@ func (qac *QRecordAvroConverter) Convert() (map[string]interface{}, error) {
 		)
 		avroVal, err := avroConverter.ToAvroValue()
 		if err != nil {
-			return nil, fmt.Errorf("failed to convert QValue to Avro-compatible value: %v", err)
+			return nil, fmt.Errorf("failed to convert QValue to Avro-compatible value: %w", err)
 		}
 
 		m[key] = avroVal
@@ -100,7 +100,7 @@ func GetAvroSchemaDefinition(
 
 	avroSchemaJSON, err := json.Marshal(avroSchema)
 	if err != nil {
-		return nil, fmt.Errorf("failed to marshal Avro schema to JSON: %v", err)
+		return nil, fmt.Errorf("failed to marshal Avro schema to JSON: %w", err)
 	}
 
 	return &QRecordAvroSchemaDefinition{

flow/model/qrecord_test.go

@@ -79,7 +79,7 @@ func TestEquals(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			assert.Equal(t, e2eshared.CheckQRecordEquality(t, tt.q1, tt.q2), tt.want)
+			assert.Equal(t, tt.want, e2eshared.CheckQRecordEquality(t, tt.q1, tt.q2))
 		})
 	}
 }

flow/model/qvalue/avro_converter.go

@@ -505,7 +505,7 @@ func (c *QValueAvroConverter) processUUID() (interface{}, error) {
 
 	u, err := uuid.FromBytes(byteData[:])
 	if err != nil {
-		return nil, fmt.Errorf("[conversion] conversion of invalid UUID value: %v", err)
+		return nil, fmt.Errorf("[conversion] conversion of invalid UUID value: %w", err)
 	}
 
 	uuidString := u.String()