func TestExtensions(t *testing.T) {
	log := common.Log
	var tests = []struct {
		name string
		rj   string
		r    [][]map[string]interface{}
	}{
		{
			name: `$$test1`,
			rj:   "{\"sql\": \"SELECT count(echo(count)) as c, echo(count) as e, countPlusOne(count) as p FROM ext where count > 49\",\"actions\": [{\"file\":  {\"path\":\"" + CACHE_FILE + "\"}}]}",
		},
	}
	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
	rp := setup()
	done := make(chan struct{})
	defer close(done)
	for i, tt := range tests {
		rp.ExecDrop("$$test1")
		rs, err := rp.ExecCreate(tt.name, tt.rj)
		if err != nil {
			t.Errorf("failed to create rule: %s.", err)
			continue
		}

		tp, err := rp.ExecInitRule(rs)
		if err != nil {
			t.Errorf("fail to init rule: %v", err)
			continue
		}

		go func() {
			select {
			case err := <-tp.Open():
				log.Println(err)
				tp.Cancel()
			}
		}()
		time.Sleep(5000 * time.Millisecond)
		log.Printf("exit main program after 5 seconds")
		results := getResults()
		if len(results) == 0 {
			t.Errorf("no result found")
			continue
		}
		log.Debugf("get results %v", results)
		var maps [][]map[string]interface{}
		for _, v := range results {
			var mapRes []map[string]interface{}
			err := json.Unmarshal([]byte(v), &mapRes)
			if err != nil {
				t.Errorf("Failed to parse the input into map")
				continue
			}
			maps = append(maps, mapRes)
		}

		for _, r := range maps {
			if len(r) != 1 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
			r := r[0]
			c := int((r["c"]).(float64))
			if c != 1 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
			e := int((r["e"]).(float64))
			if e != 50 && e != 51 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
			p := int(r["p"].(float64))
			if p != 2 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
		}
		tp.Cancel()
	}
}
//Test for source, sink, func and agg func extensions
//The .so files must be in the plugins folder
Test for source, sink, func and agg func extensions
The .so files must be in the plugins folder
// Test for source sink func and agg func extensions // The . so files must be in the plugins folder
func TestExtensions(t *testing.T) {
	log := common.Log
	var tests = []struct {
		name string
		rj   string
		r    [][]map[string]interface{}
	}{
		{
			name: `$$test1`,
			rj:   "{\"sql\": \"SELECT count(echo(count)) as c, echo(count) as e, countPlusOne(count) as p FROM ext where count > 49\",\"actions\": [{\"file\":  {\"path\":\"" + CACHE_FILE + "\"}}]}",
		},
	}
	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
	rp := setup()
	done := make(chan struct{})
	defer close(done)
	for i, tt := range tests {
		rp.ExecDrop("$$test1")
		rs, err := rp.ExecCreate(tt.name, tt.rj)
		if err != nil {
			t.Errorf("failed to create rule: %s.", err)
			continue
		}
		tp, err := rp.ExecInitRule(rs)
		if err != nil {
			t.Errorf("fail to init rule: %v", err)
			continue
		}
		go func() {
			select {
			case err := <-tp.Open():
				log.Println(err)
				tp.Cancel()
			}
		}()
		time.Sleep(5000 * time.Millisecond)
		log.Printf("exit main program after 5 seconds")
		results := getResults()
		if len(results) == 0 {
			t.Errorf("no result found")
			continue
		}
		log.Debugf("get results %v", results)
		var maps [][]map[string]interface{}
		for _, v := range results {
			var mapRes []map[string]interface{}
			err := json.Unmarshal([]byte(v), &mapRes)
			if err != nil {
				t.Errorf("Failed to parse the input into map")
				continue
			}
			maps = append(maps, mapRes)
		}
		for _, r := range maps {
			if len(r) != 1 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
			r := r[0]
			c := int((r["c"]).(float64))
			if c != 1 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
			e := int((r["e"]).(float64))
			if e != 50 && e != 51 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
			p := int(r["p"].(float64))
			if p != 2 {
				t.Errorf("%d. %q\n\nresult mismatch:\n\ngot=%#v\n\n", i, tt.rj, maps)
				break
			}
		}
		tp.Cancel()
	}
}
func TestExtensions ( t * testing . T ) { log := common . Log var tests = [ ] struct { name string rj string r [ ] [ ] map [ string ] interface { } } { { name : `$$test1` , rj : " \" \" \" \" \" \" \" \" \" \" \" " + CACHE_FILE + " \" " , } , } fmt . Printf ( " \n \n " , len ( tests ) ) rp := setup ( ) done := make ( chan struct { } ) defer close ( done ) for i , tt := range tests { rp . ExecDrop ( " " ) rs , err := rp . ExecCreate ( tt . name , tt . rj ) if err != nil { t . Errorf ( " " , err ) continue } tp , err := rp . ExecInitRule ( rs ) if err != nil { t . Errorf ( " " , err ) continue } go func ( ) { select { case err := <- tp . Open ( ) : log . Println ( err ) tp . Cancel ( ) } } ( ) time . Sleep ( 5000 * time . Millisecond ) log . Printf ( " " ) results := getResults ( ) if len ( results ) == 0 { t . Errorf ( " " ) continue } log . Debugf ( " " , results ) var maps [ ] [ ] map [ string ] interface { } for _ , v := range results { var mapRes [ ] map [ string ] interface { } err := json . Unmarshal ( [ ] byte ( v ) , & mapRes ) if err != nil { t . Errorf ( " " ) continue } maps = append ( maps , mapRes ) } for _ , r := range maps { if len ( r ) != 1 { t . Errorf ( " \n \n \n \n \n \n " , i , tt . rj , maps ) break } r := r [ 0 ] c := int ( ( r [ " " ] ) . ( float64 ) ) if c != 1 { t . Errorf ( " \n \n \n \n \n \n " , i , tt . rj , maps ) break } e := int ( ( r [ " " ] ) . ( float64 ) ) if e != 50 && e != 51 { t . Errorf ( " \n \n \n \n \n \n " , i , tt . rj , maps ) break } p := int ( r [ " " ] . ( float64 ) ) if p != 2 { t . Errorf ( " \n \n \n \n \n \n " , i , tt . rj , maps ) break } } tp . Cancel ( ) } }
Test for source, sink, func and agg func extensions
The .so files must be in the plugins folder
Test for source sink func and agg func extensions The . so files must be in the plugins folder