test FSM and calling hooks

Makefile

@@ -70,7 +70,7 @@ INSTALL_WHAT:=$(patsubst %, install_%, $(WHAT))
 
 GENERATE_DIRS := ./apricot ./coconut/cmd ./common ./common/runtype ./common/system ./core ./core/integration/ccdb ./core/integration/dcs ./core/integration/ddsched ./core/integration/kafka ./core/integration/odc ./executor ./walnut ./core/integration/trg ./core/integration/bookkeeping
 SRC_DIRS := ./apricot ./cmd/* ./core ./coconut ./executor ./common ./configuration ./occ/peanut ./walnut
-TEST_DIRS := ./apricot/local ./configuration/cfgbackend ./configuration/componentcfg ./configuration/template ./core/task/sm ./core/workflow ./core/integration/odc/fairmq ./core/integration
+TEST_DIRS := ./apricot/local ./configuration/cfgbackend ./configuration/componentcfg ./configuration/template ./core/task/sm ./core/workflow ./core/integration/odc/fairmq ./core/integration ./core/environment
 GO_TEST_DIRS := ./core/repos ./core/integration/dcs
 
 coverage:COVERAGE_PREFIX := ./coverage_results

core/environment/environment.go

@@ -907,6 +907,10 @@ func (env *Environment) handlerFunc() func(e *fsm.Event) {
 			"partition": env.id.String(),
 		}).Debug("environment.sm starting transition")
 
+		if len(e.Args) == 0 {
+			e.Cancel(errors.New("transition missing in FSM event"))
+			return
+		}
 		transition, ok := e.Args[0].(Transition)
 		if !ok {
 			e.Cancel(errors.New("transition wrapping error"))

core/environment/environment_test.go

@@ -0,0 +1,12 @@
+package environment
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"testing"
+)
+
+func TestCoreEnvironment(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Core Environment Test Suite")
+}

core/environment/fsm_test.go

@@ -0,0 +1,288 @@
+package environment
+
+import (
+	"context"
+	"github.com/AliceO2Group/Control/common/utils/uid"
+	"github.com/AliceO2Group/Control/core/integration"
+	"github.com/AliceO2Group/Control/core/integration/testplugin"
+	"github.com/AliceO2Group/Control/core/task"
+	"github.com/AliceO2Group/Control/core/workflow"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/spf13/viper"
+	"io"
+	"os"
+	"time"
+)
+
+func NewDummyTransition(transition string) Transition {
+	return &DummyTransition{
+		baseTransition: baseTransition{
+			name:    transition,
+			taskman: nil,
+		},
+	}
+}
+
+type DummyTransition struct {
+	baseTransition
+}
+
+func (t DummyTransition) do(env *Environment) (err error) {
+	return nil
+}
+
+const fsmTestConfig = "fsm_test.yaml"
+
+var tmpDir *string
+
+var _ = BeforeSuite(func() {
+	var err error
+	tmpDir = new(string)
+	*tmpDir, err = os.MkdirTemp("", "o2control-core-environment")
+	Expect(err).NotTo(HaveOccurred())
+
+	// copy config files
+	configFiles := []string{fsmTestConfig}
+	for _, configFile := range configFiles {
+		from, err := os.Open("./" + configFile)
+		Expect(err).NotTo(HaveOccurred())
+		defer from.Close()
+
+		to, err := os.OpenFile(*tmpDir+"/"+configFile, os.O_RDWR|os.O_CREATE, 0666)
+		Expect(err).NotTo(HaveOccurred())
+		defer to.Close()
+
+		_, err = io.Copy(to, from)
+		Expect(err).NotTo(HaveOccurred())
+	}
+
+	viper.Set("coreWorkingDir", tmpDir) // used by NewRunNumber with YAML backend
+
+	integration.Reset()
+	integration.RegisterPlugin("testplugin", "testPluginEndpoint", testplugin.NewPlugin)
+	viper.Reset()
+	viper.Set("integrationPlugins", []string{"testplugin"})
+	viper.Set("testPluginEndpoint", "http://example.com")
+	viper.Set("config_endpoint", "file://"+*tmpDir+"/"+fsmTestConfig)
+})
+
+var _ = AfterSuite(func() {
+	os.RemoveAll(*tmpDir)
+})
+
+var _ = Describe("environment FSM", func() {
+
+	Describe("allowed states and transitions", func() {
+		var env *Environment
+		BeforeEach(func() {
+			envId, err := uid.FromString("2oDvieFrVTi")
+			Expect(err).NotTo(HaveOccurred())
+
+			env, err = newEnvironment(nil, envId)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(env).NotTo(BeNil())
+		})
+		When("FSM is created", func() {
+			It("should be in STANDBY", func() {
+				Expect(env.Sm.Current()).To(Equal("STANDBY"))
+			})
+		})
+		When("FSM is in STANDBY", func() {
+			It("should allow for DEPLOY, GO_ERROR and EXIT transitions", func() {
+				env.Sm.SetState("STANDBY")
+				Expect(env.Sm.Can("DEPLOY")).To(BeTrue())
+				Expect(env.Sm.Can("GO_ERROR")).To(BeTrue())
+				Expect(env.Sm.Can("EXIT")).To(BeTrue())
+			})
+			It("should not allow for other transitions", func() {
+				env.Sm.SetState("STANDBY")
+				Expect(env.Sm.Cannot("CONFIGURE")).To(BeTrue())
+				Expect(env.Sm.Cannot("RESET")).To(BeTrue())
+				Expect(env.Sm.Cannot("START_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("STOP_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RECOVER")).To(BeTrue())
+			})
+		})
+		When("FSM is in DEPLOYED", func() {
+			It("should allow for CONFIGURED, GO_ERROR and EXIT transitions", func() {
+				env.Sm.SetState("DEPLOYED")
+				Expect(env.Sm.Can("CONFIGURE")).To(BeTrue())
+				Expect(env.Sm.Can("GO_ERROR")).To(BeTrue())
+				Expect(env.Sm.Can("EXIT")).To(BeTrue())
+			})
+			It("should not allow for other transitions", func() {
+				env.Sm.SetState("DEPLOYED")
+				Expect(env.Sm.Cannot("DEPLOY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RESET")).To(BeTrue())
+				Expect(env.Sm.Cannot("START_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("STOP_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RECOVER")).To(BeTrue())
+			})
+		})
+		When("FSM is in CONFIGURED", func() {
+			It("should allow for START_ACTIVITY, RESET, GO_ERROR and EXIT transitions", func() {
+				env.Sm.SetState("CONFIGURED")
+				Expect(env.Sm.Can("START_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Can("RESET")).To(BeTrue())
+				Expect(env.Sm.Can("GO_ERROR")).To(BeTrue())
+				Expect(env.Sm.Can("EXIT")).To(BeTrue())
+			})
+			It("should not allow for other transitions", func() {
+				env.Sm.SetState("CONFIGURED")
+				Expect(env.Sm.Cannot("DEPLOY")).To(BeTrue())
+				Expect(env.Sm.Cannot("CONFIGURE")).To(BeTrue())
+				Expect(env.Sm.Cannot("STOP_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RECOVER")).To(BeTrue())
+			})
+		})
+		When("FSM is in RUNNING", func() {
+			It("should allow for STOP_ACTIVITY and GO_ERROR transitions", func() {
+				env.Sm.SetState("RUNNING")
+				Expect(env.Sm.Can("STOP_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Can("GO_ERROR")).To(BeTrue())
+			})
+			It("should not allow for other transitions", func() {
+				env.Sm.SetState("RUNNING")
+				Expect(env.Sm.Cannot("DEPLOY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RESET")).To(BeTrue())
+				Expect(env.Sm.Cannot("CONFIGURE")).To(BeTrue())
+				Expect(env.Sm.Cannot("START_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RECOVER")).To(BeTrue())
+				Expect(env.Sm.Cannot("EXIT")).To(BeTrue())
+			})
+		})
+		When("FSM is in ERROR", func() {
+			It("should allow for RECOVER transition", func() {
+				env.Sm.SetState("ERROR")
+				Expect(env.Sm.Can("RECOVER")).To(BeTrue())
+				// fixme: is this correct that we do not allow for EXIT?
+				//  in principle, in TeardownEnvironment we ignore the FSM and do it ourselves,
+				//  but I don't see a reason not to use allow for it
+				// fixme: shouldn't we add also DESTROY and TEARDOWN in FSM?
+				//  is even DESTROY different from TEARDOWN?
+			})
+			It("should not allow for other transitions", func() {
+				env.Sm.SetState("ERROR")
+				Expect(env.Sm.Cannot("GO_ERROR")).To(BeTrue())
+				Expect(env.Sm.Cannot("DEPLOY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RESET")).To(BeTrue())
+				Expect(env.Sm.Cannot("CONFIGURE")).To(BeTrue())
+				Expect(env.Sm.Cannot("START_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("STOP_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("EXIT")).To(BeTrue()) // should it be? (see the comment above)
+			})
+		})
+		When("FSM is in DONE", func() {
+			It("should not allow for any transitions", func() {
+				env.Sm.SetState("DONE")
+				Expect(env.Sm.Cannot("GO_ERROR")).To(BeTrue())
+				Expect(env.Sm.Cannot("DEPLOY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RESET")).To(BeTrue())
+				Expect(env.Sm.Cannot("CONFIGURE")).To(BeTrue())
+				Expect(env.Sm.Cannot("START_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("STOP_ACTIVITY")).To(BeTrue())
+				Expect(env.Sm.Cannot("RECOVER")).To(BeTrue())
+				Expect(env.Sm.Cannot("EXIT")).To(BeTrue())
+			})
+		})
+
+	})
+
+	Describe("calling hooks on FSM events", func() {
+		var env *Environment
+		BeforeEach(func() {
+			envId, err := uid.FromString("2oDvieFrVTi")
+			Expect(err).NotTo(HaveOccurred())
+
+			env, err = newEnvironment(map[string]string{}, envId)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(env).NotTo(BeNil())
+		})
+		It("should execute the requested plugin call without errors", func() {
+			env.workflow = workflow.NewAggregatorRole("root", []workflow.Role{
+				workflow.NewCallRole(
+					"call",
+					task.Traits{Trigger: "before_CONFIGURE", Timeout: "5s", Critical: true, Await: "before_CONFIGURE"},
+					"testplugin.Test()",
+					"")})
+			workflow.LinkChildrenToParents(env.workflow)
+			env.Sm.SetState("DEPLOYED")
+			env.workflow.GetUserVars().Del("root.call_called")
+
+			err := env.Sm.Event(context.Background(), "CONFIGURE", NewDummyTransition("CONFIGURE"))
+
+			Expect(err).NotTo(HaveOccurred())
+			v, ok := env.workflow.GetUserVars().Get("root.call_called")
+			Expect(ok).To(BeTrue())
+			Expect(v).To(Equal("true"))
+		})
+
+		It("should return an error if a critical hook fails", func() {
+			env.workflow = workflow.NewAggregatorRole("root", []workflow.Role{
+				workflow.NewCallRole(
+					"call",
+					task.Traits{Trigger: "before_CONFIGURE", Timeout: "5s", Critical: true, Await: "before_CONFIGURE"},
+					"testplugin.Test()",
+					"")})
+			workflow.LinkChildrenToParents(env.workflow)
+			env.Sm.SetState("DEPLOYED")
+			env.workflow.GetUserVars().Del("root.call_called")
+			env.workflow.GetUserVars().Set("testplugin_fail", "true")
+
+			err := env.Sm.Event(context.Background(), "CONFIGURE", NewDummyTransition("CONFIGURE"))
+
+			Expect(err).To(HaveOccurred())
+			v, ok := env.workflow.GetUserVars().Get("root.call_called")
+			Expect(ok).To(BeTrue())
+			Expect(v).To(Equal("true"))
+		})
+
+		It("should not return an error if an non-critical hook fails", func() {
+			env.workflow = workflow.NewAggregatorRole("root", []workflow.Role{
+				workflow.NewCallRole(
+					"call",
+					task.Traits{Trigger: "before_CONFIGURE", Timeout: "5s", Critical: false, Await: "before_CONFIGURE"},
+					"testplugin.Test()",
+					"")})
+			workflow.LinkChildrenToParents(env.workflow)
+			env.Sm.SetState("DEPLOYED")
+			env.workflow.GetUserVars().Del("root.call_called")
+			env.workflow.GetUserVars().Set("testplugin_fail", "true")
+
+			err := env.Sm.Event(context.Background(), "CONFIGURE", NewDummyTransition("CONFIGURE"))
+
+			Expect(err).NotTo(HaveOccurred())
+			v, ok := env.workflow.GetUserVars().Get("root.call_called")
+			Expect(ok).To(BeTrue())
+			Expect(v).To(Equal("true"))
+		})
+
+		It("should return an error if a critical hook times out", func(ctx SpecContext) {
+			// fixme: this fails
+			env.workflow = workflow.NewAggregatorRole("root", []workflow.Role{
+				workflow.NewCallRole(
+					"call",
+					task.Traits{Trigger: "before_CONFIGURE", Timeout: "1s", Critical: true, Await: "before_CONFIGURE"},
+					"testplugin.Test()",
+					"")})
+			workflow.LinkChildrenToParents(env.workflow)
+			env.Sm.SetState("DEPLOYED")
+			env.workflow.GetUserVars().Del("root.call_called")
+			env.workflow.GetUserVars().Set("testplugin_hang", "true")
+
+			err := env.Sm.Event(context.Background(), "CONFIGURE", NewDummyTransition("CONFIGURE"))
+
+			Expect(err).To(HaveOccurred())
+			v, ok := env.workflow.GetUserVars().Get("root.call_called")
+			Expect(ok).To(BeTrue())
+			Expect(v).To(Equal("true"))
+		}, SpecTimeout(10*time.Second))
+
+		/// TODO
+		// It should write the SOSOR, EOSOR, SOEOR and EOEOR timestamps correctly and clear the unneeded ones
+		// the above incl. GO_ERROR
+		// Await within the same transition should work.
+		// Await across different transitions should work
+	})
+})

core/environment/fsm_test.yaml

@@ -0,0 +1,12 @@
+o2:
+  components:
+    qc:
+      TECHNICAL:
+        any:
+          entry: "config"
+  runtime:
+    aliecs:
+      defaults:
+        key1: value1
+      vars:
+        key2: value2

core/environment/transition.go

@@ -22,7 +22,6 @@
  * Intergovernmental Organization or submit itself to any jurisdiction.
  */
 
-
 package environment
 
 import (
@@ -61,13 +60,13 @@ func MakeTransition(taskman *task.Manager, optype pb.ControlEnvironmentRequest_O
 }
 
 type baseTransition struct {
-	taskman         *task.Manager
-	name            string
+	taskman *task.Manager
+	name    string
 }
 
 func (t baseTransition) check() (err error) {
 	if t.taskman == nil {
-		err = errors.New("cannot configure environment with nil roleman")
+		err = errors.New("cannot transition environment with nil taskman")
 	}
 	return
 }