test:修改订单同步测试

@@ -21,7 +21,7 @@ type OrderBestshop struct {
 	BuyerAddress string
 	//买家备注
 	BuyerRemark string
-	//
+	//买家id
 	BuyerId int64
 	//订单总数
 	OrderCount int
@@ -31,9 +31,14 @@ type OrderBestshop struct {
 	DeliveryTime string
 	//创建时间
 	CreateTime time.Time
-	PartnerId  int64
+	//合伙人id
+	PartnerId int64
 	//是否将数据同步到 order_base ,order_good
-	IsCopy    bool `pg:",use_zero"`
+	IsCopy bool `pg:",use_zero"`
+	//公司id
 	CompanyId int64
+	//订单区域
 	OrderArea string
+	//微信小程序id
+	WxAppletId string
 }

@@ -44,6 +44,7 @@ func (respository OrderBestshopRepository) transformPgModelToDomainModel(orderMo
 		IsCopy:        orderModel.IsCopy,
 		CompanyId:     orderModel.CompanyId,
 		OrderArea:     orderModel.OrderArea,
+		WxAppletId:    orderModel.WxAppletId,
 	}, nil
 }
 
@@ -67,6 +68,7 @@ func (respository OrderBestshopRepository) Add(order *domain.OrderBestShop) erro
 		IsCopy:        order.IsCopy,
 		CompanyId:     order.CompanyId,
 		OrderArea:     order.OrderArea,
+		WxAppletId:    order.WxAppletId,
 	}
 	_, err := tx.Model(&m).Insert()
 	order.Id = m.Id
@@ -94,6 +96,7 @@ func (respository OrderBestshopRepository) Edit(order *domain.OrderBestShop) err
 		IsCopy:        order.IsCopy,
 		CompanyId:     order.CompanyId,
 		OrderArea:     order.OrderArea,
+		WxAppletId:    order.WxAppletId,
 	}
 	_, err := tx.Model(&m).Where("id=?", order.Id).Update()
 	order.Id = m.Id

@@ -27,13 +27,13 @@ func DataFromXiangMi(message *sarama.ConsumerMessage) error {
 	// TODO 使用小程序id作为module
 	dataAction := msgData.Module + "/" + msgData.Action
 	switch dataAction {
-	case "xiangmi.order/ship":
-		err = syncBestshopOrder(msgData.Data)
-		if err != nil {
-			e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
-			return e
-		}
-		// TODO 统一操作小程序订单
+	//case "xiangmi.order/ship":
+	//	err = syncBestshopOrder(msgData.Data)
+	//	if err != nil {
+	//		e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
+	//		return e
+	//	}
+	// TODO 统一操作小程序订单
 	case "wxapplet.order/ship":
 		err = syncBestshopOrder(msgData.Data)
 		if err != nil {

+/**
+   @author: stevechan
+   @date: 2021/3/11
+   @note:
+**/
+
+package configs
+
+type MqConfig struct {
+	Topics     []string `json:"topics"`
+	Servers    []string `json:"servers"`
+	ConsumerId string   `json:"consumerGroup"`
+}
+
+var Cfg = MqConfig{
+	Topics:     []string{"xiangmi_project_dev"},
+	Servers:    []string{"127.0.0.1:9092"},
+	ConsumerId: "test",
+}

+/**
+   @author: stevechan
+   @date: 2021/3/11
+   @note:
+**/
+
+package sync_order
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/Shopify/sarama"
+	"github.com/gavv/httpexpect"
+	"github.com/go-pg/pg"
+	pg2 "github.com/go-pg/pg/v10"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	pG "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/infrastructure/pg"
+	"gitlab.fjmaimaimai.com/mmm-go/partnermg/test/integration/consumer/sync_order/configs"
+	"net/http"
+	"strconv"
+	"time"
+)
+
+var (
+	cfg      *configs.MqConfig
+	producer sarama.SyncProducer
+)
+
+func produce(topic string, key string, content interface{}) error {
+	data, _ := json.Marshal(content)
+
+	msg := &sarama.ProducerMessage{
+		Topic:     topic,
+		Key:       sarama.StringEncoder(key),
+		Value:     sarama.ByteEncoder(data),
+		Timestamp: time.Now(),
+	}
+
+	_, _, err2 := producer.SendMessage(msg)
+	if err2 != nil {
+		msg1 := fmt.Sprintf("Send Error topic: %v. key: %v. content: %v", topic, key, content)
+		fmt.Println(msg1)
+		return err2
+	}
+
+	fmt.Printf("Send OK topic:%s key:%s value:%s\n", topic, key, content)
+
+	return nil
+}
+
+var _ = Describe("同步小程序订单", func() {
+	var companyId int64
+	var partnerId int64
+	BeforeEach(func() {
+		fmt.Print("init kafka producer, it may take a few seconds to init the connection\n")
+		var err error
+		cfg = &configs.Cfg
+		mqConfig := sarama.NewConfig()
+		mqConfig.Producer.Return.Successes = true
+		mqConfig.Version = sarama.V0_10_2_1
+		if err = mqConfig.Validate(); err != nil {
+			msg := fmt.Sprintf("Kafka producer config invalidate. config: %v. err: %v", *cfg, err)
+			fmt.Println(msg)
+			panic(msg)
+		}
+		producer, err = sarama.NewSyncProducer(cfg.Servers, mqConfig)
+		if err != nil {
+			msg := fmt.Sprintf("Kafak producer create fail. err: %v", err)
+			fmt.Println(msg)
+			panic(msg)
+		}
+
+		// 新增公司
+		_, err2 := pG.DB.QueryOne(
+			pg.Scan(&companyId),
+			"INSERT INTO company (name, phone, logo, admin_company, status, create_at, update_at, delete_at, remarks, enable, abbreviation, applets) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
+			"测试公司", "13162672794", "http://suplus-file-dev.fjmaimaimai.com/upload/image/2020091709304822611.jpg", 431, Default, "2020-09-17 09:30:05.588579+08", "2020-09-17 09:30:05.588579+08", pg2.NullTime{}, "", "测试", 1, "公司", []map[string]interface{}{{"id": "gh_18eb644002fb", "url": "pages/index/index", "name": "海鲜干货啦啦啦啦啦", "imageUrl": ""}})
+		Expect(err2).NotTo(HaveOccurred())
+
+		// 新增合伙人
+		_, err3 := pG.DB.QueryOne(
+			pg.Scan(&partnerId),
+			"INSERT INTO partner_info (partner_name, account, password, status, partner_category, region_info, cooperate_time, create_at, update_at, salesman, partner_category_infos, company_id, remark) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
+			"test_partner", "19121619631", "7c4a8d09ca3762af61e59520943dc26494f8941b", 1, 1, map[string]interface{}{"regionName": "123"}, "2020-09-01 00:00:00+08", "2020-09-01 00:00:00+08", "2020-09-01 00:00:00+08", []map[string]interface{}{{"name": "123", "telephone": "13322223333"}}, []map[string]interface{}{{"id": 2, "code": "001"}}, 1, "测试合伙人")
+		Expect(err3).NotTo(HaveOccurred())
+
+		key := strconv.FormatInt(time.Now().UTC().UnixNano(), 10)
+		value := map[string]interface{}{
+			"module": "wxapplet.order",
+			"action": "ship",
+			"data": map[string]interface{}{
+				"orderCode":     "1234567895",
+				"orderCount":    1,
+				"orderAmount":   2,
+				"orderTime":     "2021-03-05 12:01:54",
+				"orderState":    1,
+				"deliveryState": 1,
+				"buyerName":     "陈志颖",
+				"buyerPhone":    "19121619631",
+				"buyerAddress":  "福建省 福州市 仓山区",
+				"buyerRemark":   "不辣",
+				"buyerId":       123,
+				"partnerId":     partnerId,
+				"goods": []map[string]interface{}{
+					{
+						"sn":     "wertyuf",
+						"id":     1,
+						"bn":     "a;slkdjgaj",
+						"name":   "地瓜",
+						"price":  2,
+						"nums":   1,
+						"amount": 2,
+					},
+				},
+				"companyId":    companyId,
+				"deliveryTime": "2021-03-05 12:01:54",
+				"orderArea":    "福建省 福州市 台江区",
+				"wxAppletId":   "gh_18eb644002fb",
+			},
+		}
+		err4 := produce(cfg.Topics[0], key, value)
+		Expect(err4).NotTo(HaveOccurred())
+	})
+	Describe("提交小程序订单消息", func() {
+		Context("提交正确的小程序订单", func() {
+			It("返回正确的订单数据", func() {
+				httpExpect := httpexpect.New(GinkgoT(), server.URL)
+				httpExpect.GET("").
+					Expect().
+					Status(http.StatusOK).
+					JSON().
+					Object().
+					ContainsKey("code").ValueEqual("code", 0)
+			})
+		})
+	})
+	AfterEach(func() {
+		_, err1 := pG.DB.Exec("DELETE FROM order_bestshop WHERE true")
+		Expect(err1).NotTo(HaveOccurred())
+		_, err2 := pG.DB.Exec("DELETE FROM company WHERE true")
+		Expect(err2).NotTo(HaveOccurred())
+		_, err3 := pG.DB.Exec("DELETE FROM partner_info WHERE true")
+		Expect(err3).NotTo(HaveOccurred())
+	})
+})

+/**
+   @author: stevechan
+   @date: 2021/3/11
+   @note:
+**/
+
+package sync_order
+
+import (
+	"github.com/astaxie/beego"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	_ "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/infrastructure/pg"
+	_ "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/beego"
+)
+
+func TestColumnSetting(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Kafka Consumer Sync Order Correlations Test Case Suite")
+}
+
+var handler http.Handler
+var server *httptest.Server
+
+// 初始化kafka
+var _ = BeforeSuite(func() {
+	handler = beego.BeeApp.Handlers
+	server = httptest.NewServer(handler)
+})
+
+var _ = AfterSuite(func() {
+	server.Close()
+})