diff --git a/test/engine/setup/env.go b/test/engine/setup/env.go
index 6d5bdcbce6..49b375c595 100644
--- a/test/engine/setup/env.go
+++ b/test/engine/setup/env.go
@@ -58,3 +58,13 @@ func SetAgentPID(ctx context.Context) (context.Context, error) {
 	}
 	return context.WithValue(ctx, config.AgentPIDKey, result), nil
 }
+
+func RunCommandOnSource(ctx context.Context, command string) (context.Context, error) {
+	_, err := Env.ExecOnSource(ctx, command)
+	return ctx, err
+}
+
+func RunCommandOnLoongCollector(ctx context.Context, command string) (context.Context, error) {
+	_, err := Env.ExecOnLoongCollector(command)
+	return ctx, err
+}
diff --git a/test/engine/setup/subscriber/sls.go b/test/engine/setup/subscriber/sls.go
index b817bee8fe..c2f074e8c7 100644
--- a/test/engine/setup/subscriber/sls.go
+++ b/test/engine/setup/subscriber/sls.go
@@ -32,6 +32,7 @@ type SLSSubscriber struct {
 	Aliuid        string
 	Region        string
 	Endpoint      string
+	QueryEndpoint string
 	Project       string
 	Logstore      string
 }
@@ -112,7 +113,7 @@ func (s *SLSSubscriber) getLogFromSLS(sql string, from int32) (*sls.GetLogsRespo
 		From:  tea.Int32(from),
 		To:    tea.Int32(now),
 	}
-	resp, err := s.client.GetLogs(tea.String(config.TestConfig.Project), tea.String(config.TestConfig.GetLogstore(s.TelemetryType)), req)
+	resp, err := s.client.GetLogs(tea.String(s.Project), tea.String(s.Logstore), req)
 	if err != nil {
 		return nil, err
 	}
@@ -134,15 +135,7 @@ func createSLSClient(accessKeyID, accessKeySecret, endpoint string) *sls.Client
 
 func init() {
 	RegisterCreator(slsName, func(spec map[string]interface{}) (Subscriber, error) {
-		telemetryType := "logs"
-		if v, ok := spec["telemetry_type"]; ok {
-			telemetryType = v.(string)
-		}
-		fmt.Println("create sls subscriber with telemetry type", telemetryType)
-		l := &SLSSubscriber{
-			client:        createSLSClient(config.TestConfig.AccessKeyID, config.TestConfig.AccessKeySecret, config.GetQueryEndpoint()),
-			TelemetryType: telemetryType,
-		}
+		l := &SLSSubscriber{}
 		if v, ok := spec["aliuid"]; ok {
 			l.Aliuid = v.(string)
 		} else {
@@ -166,8 +159,19 @@ func init() {
 		if v, ok := spec["logstore"]; ok {
 			l.Logstore = v.(string)
 		} else {
-			l.Logstore = config.TestConfig.GetLogstore(telemetryType)
+			l.Logstore = config.TestConfig.Logstore
+		}
+		if v, ok := spec["query_endpoint"]; ok {
+			l.QueryEndpoint = v.(string)
+		} else {
+			l.QueryEndpoint = config.TestConfig.Endpoint
+		}
+		if v, ok := spec["telemetry_type"]; ok {
+			l.TelemetryType = v.(string)
+		} else {
+			l.TelemetryType = "logs"
 		}
+		l.client = createSLSClient(config.TestConfig.AccessKeyID, config.TestConfig.AccessKeySecret, l.QueryEndpoint)
 		return l, nil
 	})
 	doc.Register("subscriber", slsName, new(SLSSubscriber))
diff --git a/test/engine/steps.go b/test/engine/steps.go
index dbbb2524dc..ed921b8b08 100644
--- a/test/engine/steps.go
+++ b/test/engine/steps.go
@@ -32,6 +32,8 @@ func ScenarioInitializer(ctx *godog.ScenarioContext) {
 	ctx.Given(`^subcribe data from \{(\S+)\} with config`, subscriber.InitSubscriber)
 	ctx.Given(`^mkdir \{(.*)\}`, setup.Mkdir)
 	ctx.Given(`^docker-compose boot type \{(\S+)\}$`, setup.SetDockerComposeBootType)
+	ctx.Given(`^run command on datasource \{(.*)\}$`, setup.RunCommandOnSource)
+	ctx.Given(`^run command on loongcollector \{(.*)\}$`, setup.RunCommandOnLoongCollector)
 
 	// chaos
 	ctx.Given(`^network delay package \{(\d+)\}ms for ip \{(.*)\}`, chaos.NetworkDelay)
diff --git a/test/main.go b/test/main.go
index 1fb14fc6d7..7f459e532c 100644
--- a/test/main.go
+++ b/test/main.go
@@ -10,8 +10,9 @@ import (
 	"strings"
 	"time"
 
-	"github.com/alibaba/ilogtail/test/engine"
 	"github.com/cucumber/godog"
+
+	"github.com/alibaba/ilogtail/test/engine"
 )
 
 func main() {