add api integration test

Signed-off-by: SimFG <[email protected]>
zilliztech · Nov 29, 2024 · 21378a4 · 21378a4
1 parent 7a3ca02
commit 21378a4
Show file tree

Hide file tree

Showing 10 changed files with 245 additions and 209 deletions.
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
@@ -688,20 +688,24 @@ jobs:
                   "db_collections": {
                       "foo": [
                           {
-                            "name": "*" 
+                            "name": "*"
                           }
                       ]
                   }
               }
           }'
 
       - name: Run test
-        timeout-minutes: 15
+        timeout-minutes: 30
         shell: bash
         working-directory: tests
         run: |
           pip install -r requirements.txt --trusted-host https://test.pypi.org
           pytest testcases/test_cdc_database.py --upstream_host 127.0.0.1 --upstream_port 19530 --downstream_host 127.0.0.1 --downstream_port 19500
+          pytest testcases/test_cdc_get.py --upstream_host 127.0.0.1 --upstream_port 19530 --downstream_host 127.0.0.1 --downstream_port 19500
+          pytest testcases/test_cdc_list.py --upstream_host 127.0.0.1 --upstream_port 19530 --downstream_host 127.0.0.1 --downstream_port 19500
+          pytest testcases/test_cdc_pause.py --upstream_host 127.0.0.1 --upstream_port 19530 --downstream_host 127.0.0.1 --downstream_port 19500
+          pytest testcases/test_cdc_resume.py --upstream_host 127.0.0.1 --upstream_port 19530 --downstream_host 127.0.0.1 --downstream_port 19500
 
       - name: List CDC task
         if: ${{ always() }}

diff --git a/server/cdc_impl.go b/server/cdc_impl.go
@@ -327,13 +327,6 @@ func (e *MetaCDC) checkDuplicateCollection(uKey string,
 	}
 	if names, ok := e.collectionNames.data[uKey]; ok {
 		var duplicateCollections []string
-		containsAny := false
-		for _, name := range names {
-			d, c := util.GetCollectionNameFromFull(name)
-			if d == cdcreader.AllDatabase || c == cdcreader.AllCollection {
-				containsAny = true
-			}
-		}
 		for _, newCollectionName := range newCollectionNames {
 			if lo.Contains(names, newCollectionName) {
 				duplicateCollections = append(duplicateCollections, newCollectionName)
@@ -343,9 +336,12 @@ func (e *MetaCDC) checkDuplicateCollection(uKey string,
 			if nd == cdcreader.AllDatabase && nc == cdcreader.AllCollection {
 				continue
 			}
-			if containsAny && !lo.Contains(e.collectionNames.excludeData[uKey], newCollectionName) {
-				duplicateCollections = append(duplicateCollections, newCollectionName)
-				continue
+			for _, name := range names {
+				match, containAny := matchCollectionName(name, newCollectionName)
+				if match && containAny && !lo.Contains(e.collectionNames.excludeData[uKey], newCollectionName) {
+					duplicateCollections = append(duplicateCollections, newCollectionName)
+					break
+				}
 			}
 		}
 		if len(duplicateCollections) > 0 {
@@ -886,7 +882,7 @@ func (e *MetaCDC) newReplicateEntity(info *meta.TaskInfo) (*ReplicateEntity, err
 	}
 	if err != nil {
 		taskLog.Warn("fail to new the data handler", zap.Error(err))
-		return nil, servererror.NewClientError("fail to new the data handler, task_id: ")
+		return nil, servererror.NewClientError("fail to new the data handler, task_id: " + info.TaskID)
 	}
 	writerObj := cdcwriter.NewChannelWriter(dataHandler, config.WriterConfig{
 		MessageBufferSize: bufferSize,

diff --git a/server/cdc_impl_test.go b/server/cdc_impl_test.go
@@ -1591,6 +1591,25 @@ func TestCheckDuplicateCollection(t *testing.T) {
 		assert.Len(t, excludeCollections, 0)
 	})
 
+	t.Run("collection duplicate test", func(t *testing.T) {
+		metaCDC := &MetaCDC{}
+		initMetaCDCMap(metaCDC)
+		excludeCollections, err := metaCDC.checkDuplicateCollection("foo", []string{
+			util.GetFullCollectionName("foo", "*"),
+		}, model.ExtraInfo{
+			EnableUserRole: true,
+		}, nil)
+		assert.NoError(t, err)
+		assert.Len(t, excludeCollections, 0)
+
+		_, err = metaCDC.checkDuplicateCollection("foo", []string{
+			util.GetFullCollectionName("default", "col1"),
+		}, model.ExtraInfo{
+			EnableUserRole: false,
+		}, nil)
+		assert.NoError(t, err)
+	})
+
 	t.Run("map collection name", func(t *testing.T) {
 		metaCDC := &MetaCDC{}
 		initMetaCDCMap(metaCDC)

diff --git a/tests/api/milvus_cdc.py b/tests/api/milvus_cdc.py
@@ -1,6 +1,7 @@
 import json
 import requests
 
+DEFAULT_TOKEN = 'root:Milvus'
 
 class MilvusCdcClient:
 
@@ -17,7 +18,7 @@ def create_task(self, request_data):
         payload = json.dumps(body)
         response = requests.post(url, headers=self.headers, data=payload)
         if response.status_code == 200:
-            return response.json(), True
+            return response.json()['data'], True
         else:
             return response.text, False
 
@@ -28,8 +29,8 @@ def list_tasks(self):
         }
         payload = json.dumps(body)
         response = requests.post(url, headers=self.headers, data=payload)
-        if response.status_code == 200:
-            return response.json(), True
+        if response.status_code == 200 and 'data' in response.json():
+            return response.json()['data'], True
         else:
             return response.text, False
 
@@ -43,8 +44,8 @@ def get_task(self, task_id):
         }
         payload = json.dumps(body)
         response = requests.post(url, headers=self.headers, data=payload)
-        if response.status_code == 200:
-            return response.json(), True
+        if response.status_code == 200 and 'data' in response.json():
+            return response.json()['data']['task'], True
         else:
             return response.text, False