Jaculabilis/intake
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add statePath argument to Execute

Browse Source
This commit is contained in:
Tim Van Baak 2025-01-29 23:51:06 -08:00
parent 34fa784c44
commit f5e9277e26
8 changed files with 16 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/actionExecute.go
View File

@ -90,7 +90,7 @@ func actionExecute() {
log.Fatalf("error: failed to get action: %v", err)
}
newItem, err := core.ExecuteItemAction(item, actionExecuteSource, argv, time.Minute)
newItem, err := core.ExecuteItemAction(item, actionExecuteSource, argv, nil, "", time.Minute)
if err != nil {
log.Fatalf("error executing action: %v", err)
}

2
cmd/sourceFetch.go
View File

@ -51,7 +51,7 @@ func sourceFetch() {
log.Fatalf("error: failed to get fetch action: %v", err)
}
items, err := core.Execute(sourceFetchSource, argv, nil, "", time.Minute)
items, err := core.Execute(sourceFetchSource, argv, nil, "", "", time.Minute)
if err != nil {
log.Fatalf("error: failed to execute fetch: %v", err)
}

2
cmd/sourceTest.go
View File

@ -38,7 +38,7 @@ func init() {
func sourceTest(cmd []string) {
formatter := formatAs(sourceTestFormat)
items, err := core.Execute("", cmd, sourceTestEnv, "", time.Minute)
items, err := core.Execute("", cmd, sourceTestEnv, "", "", time.Minute)
log.Printf("Returned %d items", len(items))
if err != nil {
log.Fatal(err)

9
core/action.go
View File

@ -127,10 +127,11 @@ func Execute(
source string,
argv []string,
env []string,
statePath string,
input string,
timeout time.Duration,
) ([]Item, error) {
log.Printf("Executing %v", argv)
log.Printf("executing %v", argv)
if len(argv) == 0 {
return nil, errors.New("empty argv")
@ -139,7 +140,7 @@ func Execute(
return nil, errors.New("empty source")
}
env = append(env, "STATE_PATH=")
env = append(env, "STATE_PATH="+statePath)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
@ -213,6 +214,8 @@ func ExecuteItemAction(
item Item,
action string,
argv []string,
env []string,
statePath string,
timeout time.Duration,
) (Item, error) {
itemJson, err := json.Marshal(item)
@ -220,7 +223,7 @@ func ExecuteItemAction(
return Item{}, fmt.Errorf("failed to serialize item: %v", err)
}
res, err := Execute(item.Source, argv, nil, string(itemJson), time.Minute)
res, err := Execute(item.Source, argv, env, statePath, string(itemJson), timeout)
if err != nil {
return Item{}, fmt.Errorf("failed to execute %s for %s/%s: %v", action, item.Source, item.Id, err)
}

8
core/action_test.go
View File

@ -72,7 +72,7 @@ func TestExecute(t *testing.T) {
}
}
execute := func(argv []string) ([]Item, error) {
return Execute("_", argv, nil, "", time.Minute)
return Execute("_", argv, nil, "", "", time.Minute)
}
res, err := execute([]string{"true"})
@ -89,7 +89,7 @@ func TestExecute(t *testing.T) {
assertLen(res, 0)
// Timeout
res, err = Execute("_", []string{"sleep", "10"}, nil, "", time.Millisecond)
res, err = Execute("_", []string{"sleep", "10"}, nil, "", "", time.Millisecond)
assertNotNil(err)
assertLen(res, 0)
@ -102,7 +102,7 @@ func TestExecute(t *testing.T) {
}
// Read from stdin
res, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, "bar", time.Minute)
res, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, "", "bar", time.Minute)
assertNil(err)
assertLen(res, 1)
if res[0].Id != "bar" {
@ -110,7 +110,7 @@ func TestExecute(t *testing.T) {
}
// Set env
res, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, "", time.Minute)
res, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, "", "", time.Minute)
assertNil(err)
assertLen(res, 1)
if res[0].Id != "baz" {

2
core/source.go
View File

@ -284,7 +284,7 @@ func UpdateWithFetchedItems(db *DB, source string, items []Item) (int, int, erro
if err == nil {
var updatedNewItems []Item
for _, item := range newItems {
updatedItem, err := ExecuteItemAction(item, "on_create", onCreateArgv, time.Minute)
updatedItem, err := ExecuteItemAction(item, "on_create", onCreateArgv, nil, "", time.Minute)
if err != nil {
log.Printf("error: %v", err)
}

2
core/source_test.go
View File

@ -173,7 +173,7 @@ func TestOnCreateAction(t *testing.T) {
}
execute := func(argv []string) []Item {
items, err := Execute("test", argv, nil, "", time.Minute)
items, err := Execute("test", argv, nil, "", "", time.Minute)
if err != nil {
t.Fatal("unexpected error executing test fetch")
}

2
web/item.go
View File

@ -61,7 +61,7 @@ func (env *Env) doAction(writer http.ResponseWriter, req *http.Request) {
return
}
newItem, err := core.ExecuteItemAction(item, action, argv, time.Minute)
newItem, err := core.ExecuteItemAction(item, action, argv, nil, "", time.Minute)
if err != nil {
http.Error(writer, err.Error(), 500)
return