14442149348

Committed 14 Apr 2025 09:24AM UTC coverage: 72.673% (+0.4%) from 72.224%

Build # 14442149348

Build Type

push

github

Committed by

web-flow

Commit Message

Run operations (#422)

* Add operation messages mapping

* Implement operation requests policy

* Listen for operation requests

* Wrap error messages

* Add tests to policy

* Use lowercase to wrap errors

Run Details

174 of 201 new or added lines in 4 files covered. (86.57%)

4662 of 6415 relevant lines covered (72.67%)

19.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.18

/internal/operations/mapper.go

package operations

import (
        "fmt"

        "github.com/google/uuid"
        "github.com/pkg/errors"
        "google.golang.org/protobuf/types/known/structpb"

        "github.com/trento-project/contracts/go/pkg/events"
        "github.com/trento-project/workbench/pkg/operator"
)

const (
        EventSource = "https://github.com/trento-project/agent"
)

type OperatorExecutionRequestedTarget struct {
        AgentID   string
        Arguments map[string]interface{}
}

type OperatorExecutionRequested struct {
        OperationID string
        GroupID     string
        StepNumber  int32
        Operator    string
        Targets     []OperatorExecutionRequestedTarget
}

func (r *OperatorExecutionRequested) GetTargetAgent(agentID string) *OperatorExecutionRequestedTarget {
        var currentAgent *OperatorExecutionRequestedTarget

        for _, target := range r.Targets {
                if target.AgentID == agentID {
                        currentAgent = &target
                }
        }

        return currentAgent
}

func OperatorExecutionRequestedFromEvent(event []byte) (*OperatorExecutionRequested, error) {
        var operatorExecutionRequested events.OperatorExecutionRequested

        err := events.FromEvent(event, &operatorExecutionRequested, events.WithExpirationCheck())
        if err != nil {
                return nil, err
        }

        targets := []OperatorExecutionRequestedTarget{}

        for _, target := range operatorExecutionRequested.Targets {
                arguments := make(map[string]interface{})
                for key, value := range target.GetArguments() {
                        arguments[key] = value.AsInterface()
                }

                newTarget := OperatorExecutionRequestedTarget{
                        AgentID:   target.GetAgentId(),
                        Arguments: arguments,
                }

                targets = append(targets, newTarget)
        }

        return &OperatorExecutionRequested{
                OperationID: operatorExecutionRequested.GetOperationId(),
                GroupID:     operatorExecutionRequested.GetGroupId(),
                StepNumber:  operatorExecutionRequested.GetStepNumber(),
                Operator:    operatorExecutionRequested.GetOperator(),
                Targets:     targets,
        }, nil
}

func OperatorExecutionCompletedToEvent(
        operationID,
        groupID,
        agentID string,
        stepNumber int32,
        report *operator.ExecutionReport,
) ([]byte, error) {
        event := events.OperatorExecutionCompleted{
                OperationId: operationID,
                GroupId:     groupID,
                StepNumber:  stepNumber,
                AgentId:     agentID,
        }

        if report.Success != nil {
                before, beforeFound := report.Success.Diff["before"]
                if !beforeFound {
                        return nil, fmt.Errorf("before not found in report")
                }

                beforeValue, err := structpb.NewValue(before)
                if err != nil {
                        return nil, err
                }

                after, afterFound := report.Success.Diff["after"]
                if !afterFound {
                        return nil, fmt.Errorf("after not found in report")
                }

                afterValue, err := structpb.NewValue(after)
                if err != nil {
                        return nil, err
                }

                result := &events.OperatorExecutionCompleted_Value{
                        Value: &events.OperatorResponse{
                                Phase: events.OperatorPhase(events.OperatorPhase_value[string(report.Success.LastPhase)]),
                                Diff: &events.OperatorDiff{
                                        Before: beforeValue,
                                        After:  afterValue,
                                },
                        },
                }
                event.Result = result
        } else {
                result := &events.OperatorExecutionCompleted_Error{
                        Error: &events.OperatorError{
                                Phase:   events.OperatorPhase(events.OperatorPhase_value[string(report.Error.ErrorPhase)]),
                                Message: report.Error.Message,
                        },
                }
                event.Result = result
        }

        eventBytes, err := events.ToEvent(
                &event,
                events.WithSource(EventSource),
                events.WithID(uuid.New().String()),
        )
        if err != nil {
                return nil, errors.Wrap(err, "error creating event")
        }

        return eventBytes, nil
}

1	package operations
2
3	import (
4	"fmt"
5
6	"github.com/google/uuid"
7	"github.com/pkg/errors"
8	"google.golang.org/protobuf/types/known/structpb"
9
10	"github.com/trento-project/contracts/go/pkg/events"
11	"github.com/trento-project/workbench/pkg/operator"
12	)
13
14	const (
15	EventSource = "https://github.com/trento-project/agent"
16	)
17
18	type OperatorExecutionRequestedTarget struct {
19	AgentID string
20	Arguments map[string]interface{}
21	}
22
23	type OperatorExecutionRequested struct {
24	OperationID string
25	GroupID string
26	StepNumber int32
27	Operator string
28	Targets []OperatorExecutionRequestedTarget
29	}
30
31	func (r OperatorExecutionRequested) GetTargetAgent(agentID string) OperatorExecutionRequestedTarget {	8✔
32	var currentAgent *OperatorExecutionRequestedTarget	8✔
33		8✔
34	for _, target := range r.Targets {	20✔
35	if target.AgentID == agentID {	18✔
36	currentAgent = &target	6✔
37	}	6✔
38	}
39
40	return currentAgent	8✔
41	}
42
43	func OperatorExecutionRequestedFromEvent(event []byte) (*OperatorExecutionRequested, error) {	9✔
44	var operatorExecutionRequested events.OperatorExecutionRequested	9✔
45		9✔
46	err := events.FromEvent(event, &operatorExecutionRequested, events.WithExpirationCheck())	9✔
47	if err != nil {	11✔
48	return nil, err	2✔
49	}	2✔
50
51	targets := []OperatorExecutionRequestedTarget{}	7✔
52		7✔
53	for _, target := range operatorExecutionRequested.Targets {	17✔
54	arguments := make(map[string]interface{})	10✔
55	for key, value := range target.GetArguments() {	14✔
56	arguments[key] = value.AsInterface()	4✔
57	}	4✔
58
59	newTarget := OperatorExecutionRequestedTarget{	10✔
60	AgentID: target.GetAgentId(),	10✔
61	Arguments: arguments,	10✔
62	}	10✔
63		10✔
64	targets = append(targets, newTarget)	10✔
65	}
66
67	return &OperatorExecutionRequested{	7✔
68	OperationID: operatorExecutionRequested.GetOperationId(),	7✔
69	GroupID: operatorExecutionRequested.GetGroupId(),	7✔
70	StepNumber: operatorExecutionRequested.GetStepNumber(),	7✔
71	Operator: operatorExecutionRequested.GetOperator(),	7✔
72	Targets: targets,	7✔
73	}, nil	7✔
74	}
75
76	func OperatorExecutionCompletedToEvent(
77	operationID,
78	groupID,
79	agentID string,
80	stepNumber int32,
81	report *operator.ExecutionReport,
82	) ([]byte, error) {	8✔
83	event := events.OperatorExecutionCompleted{	8✔
84	OperationId: operationID,	8✔
85	GroupId: groupID,	8✔
86	StepNumber: stepNumber,	8✔
87	AgentId: agentID,	8✔
88	}	8✔
89		8✔
90	if report.Success != nil {	15✔
91	before, beforeFound := report.Success.Diff["before"]	7✔
92	if !beforeFound {	8✔
93	return nil, fmt.Errorf("before not found in report")	1✔
94	}	1✔
95
96	beforeValue, err := structpb.NewValue(before)	6✔
97	if err != nil {	6✔
NEW 98	return nil, err	×
NEW 99	}	×
100
101	after, afterFound := report.Success.Diff["after"]	6✔
102	if !afterFound {	8✔
103	return nil, fmt.Errorf("after not found in report")	2✔
104	}	2✔
105
106	afterValue, err := structpb.NewValue(after)	4✔
107	if err != nil {	4✔
NEW 108	return nil, err	×
NEW 109	}	×
110
111	result := &events.OperatorExecutionCompleted_Value{	4✔
112	Value: &events.OperatorResponse{	4✔
113	Phase: events.OperatorPhase(events.OperatorPhase_value[string(report.Success.LastPhase)]),	4✔
114	Diff: &events.OperatorDiff{	4✔
115	Before: beforeValue,	4✔
116	After: afterValue,	4✔
117	},	4✔
118	},	4✔
119	}	4✔
120	event.Result = result	4✔
121	} else {	1✔
122	result := &events.OperatorExecutionCompleted_Error{	1✔
123	Error: &events.OperatorError{	1✔
124	Phase: events.OperatorPhase(events.OperatorPhase_value[string(report.Error.ErrorPhase)]),	1✔
125	Message: report.Error.Message,	1✔
126	},	1✔
127	}	1✔
128	event.Result = result	1✔
129	}	1✔
130
131	eventBytes, err := events.ToEvent(	5✔
132	&event,	5✔
133	events.WithSource(EventSource),	5✔
134	events.WithID(uuid.New().String()),	5✔
135	)	5✔
136	if err != nil {	5✔
NEW 137	return nil, errors.Wrap(err, "error creating event")	×
NEW 138	}	×
139
140	return eventBytes, nil	5✔
141	}

trento-project / agent / 14442149348

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous