api: simplify the code & switch to container mocks for testing

api/api.go

@@ -1,8 +1,16 @@
 package api
 
 import (
+	"fmt"
+	"time"
+
 	"github.com/mesg-foundation/core/container"
 	"github.com/mesg-foundation/core/database"
+	"github.com/mesg-foundation/core/event"
+	"github.com/mesg-foundation/core/execution"
+	"github.com/mesg-foundation/core/pubsub"
+	"github.com/mesg-foundation/core/service"
+	uuid "github.com/satori/go.uuid"
 )
 
 // API exposes all functionalities of MESG core.
@@ -37,3 +45,165 @@ func ContainerOption(container container.Container) Option {
 		a.container = container
 	}
 }
+
+// GetService returns service serviceID.
+func (a *API) GetService(serviceID string) (*service.Service, error) {
+	s, err := a.db.Get(serviceID)
+	if err != nil {
+		return nil, err
+	}
+	return service.FromService(s, service.ContainerOption(a.container))
+}
+
+// ListServices returns all services.
+func (a *API) ListServices() ([]*service.Service, error) {
+	ss, err := a.db.All()
+	if err != nil {
+		return nil, err
+	}
+
+	var services []*service.Service
+	for _, s := range ss {
+		s, err = service.FromService(s, service.ContainerOption(a.container))
+		if err != nil {
+			return nil, err
+		}
+		services = append(services, s)
+	}
+
+	return services, nil
+}
+
+// StartService starts service serviceID.
+func (a *API) StartService(serviceID string) error {
+	sr, err := a.db.Get(serviceID)
+	if err != nil {
+		return err
+	}
+	sr, err = service.FromService(sr, service.ContainerOption(a.container))
+	if err != nil {
+		return err
+	}
+	_, err = sr.Start()
+	return err
+}
+
+// StopService stops service serviceID.
+func (a *API) StopService(serviceID string) error {
+	sr, err := a.db.Get(serviceID)
+	if err != nil {
+		return err
+	}
+	sr, err = service.FromService(sr, service.ContainerOption(a.container))
+	if err != nil {
+		return err
+	}
+	return sr.Stop()
+}
+
+// DeleteService stops and deletes service serviceID.
+// when deleteData is enabled, any persistent data that belongs to
+// the service and to its dependencies also will be deleted.
+func (a *API) DeleteService(serviceID string, deleteData bool) error {
+	s, err := a.db.Get(serviceID)
+	if err != nil {
+		return err
+	}
+	s, err = service.FromService(s, service.ContainerOption(a.container))
+	if err != nil {
+		return err
+	}
+	if err := s.Stop(); err != nil {
+		return err
+	}
+	// delete volumes first before the service. this way if
+	// deleting volumes fails, process can be retried by the user again
+	// because service still will be in the db.
+	if deleteData {
+		if err := s.DeleteVolumes(); err != nil {
+			return err
+		}
+	}
+	return a.db.Delete(serviceID)
+}
+
+// EmitEvent emits a MESG event eventKey with eventData for service token.
+func (a *API) EmitEvent(token, eventKey string, eventData map[string]interface{}) error {
+	s, err := a.db.Get(token)
+	if err != nil {
+		return err
+	}
+	event, err := event.Create(s, eventKey, eventData)
+	if err != nil {
+		return err
+	}
+	event.Publish()
+	return nil
+}
+
+// ExecuteTask executes a task tasKey with inputData and tags for service serviceID.
+func (a *API) ExecuteTask(serviceID, taskKey string, inputData map[string]interface{},
+	tags []string) (executionID string, err error) {
+	s, err := a.db.Get(serviceID)
+	if err != nil {
+		return "", err
+	}
+	s, err = service.FromService(s, service.ContainerOption(a.container))
+	if err != nil {
+		return "", err
+	}
+
+	// a task should be executed only if task's service is running.
+	status, err := s.Status()
+	if err != nil {
+		return "", err
+	}
+	if status != service.RUNNING {
+		return "", &NotRunningServiceError{ServiceID: s.Sid}
+	}
+
+	// execute the task.
+	eventID := uuid.NewV4().String()
+	exec, err := execution.New(s, eventID, taskKey, inputData, tags)
+	if err != nil {
+		return "", err
+	}
+	if err := exec.Execute(); err != nil {
+		return "", err
+	}
+	if err = a.execDB.Save(exec); err != nil {
+		return "", err
+	}
+	go pubsub.Publish(s.TaskSubscriptionChannel(), exec)
+	return exec.ID, nil
+}
+
+// ExecuteAndListen executes given task and listen for result.
+func (a *API) ExecuteAndListen(serviceID, task string, inputs map[string]interface{}) (*execution.Execution, error) {
+	tag := uuid.NewV4().String()
+	result, err := a.ListenResult(serviceID, ListenResultTagFilters([]string{tag}))
+	if err != nil {
+		return nil, err
+	}
+	defer result.Close()
+
+	// XXX: sleep because listen stream may not be ready to stream the data
+	// and execution will done before stream is ready. In that case the response
+	// wlll never come TODO: investigate
+	time.Sleep(1 * time.Second)
+
+	if _, err := a.ExecuteTask(serviceID, task, inputs, []string{tag}); err != nil {
+		return nil, err
+	}
+	return <-result.Executions, nil
+}
+
+// NotRunningServiceError is an error returned when the service is not running that
+// a task needed to be executed on.
+type NotRunningServiceError struct {
+	ServiceID string
+}
+
+func (e *NotRunningServiceError) Error() string {
+	return fmt.Sprintf("Service %q is not running", e.ServiceID)
+}

api/api_test.go

@@ -5,8 +5,10 @@ import (
 	"testing"
 
 	"github.com/mesg-foundation/core/container"
-	"github.com/mesg-foundation/core/container/dockertest"
+	"github.com/mesg-foundation/core/container/mocks"
 	"github.com/mesg-foundation/core/database"
+	"github.com/mesg-foundation/core/service"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
 
@@ -15,26 +17,94 @@ const (
 	execdbname    = "exec.db.test"
 )
 
-func newAPIAndDockerTest(t *testing.T) (*API, *dockertest.Testing, func()) {
-	dt := dockertest.New()
+type apiTesting struct {
+	*testing.T
+	serviceDB   *database.LevelDBServiceDB
+	executionDB *database.LevelDBExecutionDB
+	cm          *mocks.Container
+}
 
-	container, err := container.New(container.ClientOption(dt.Client()))
-	require.NoError(t, err)
+func (t *apiTesting) close() {
+	require.NoError(t, t.serviceDB.Close())
+	require.NoError(t, t.executionDB.Close())
+	require.NoError(t, os.RemoveAll(servicedbname))
+	require.NoError(t, os.RemoveAll(execdbname))
+}
+
+func newTesting(t *testing.T) (*API, *apiTesting) {
+	cm := &mocks.Container{}
 
 	db, err := database.NewServiceDB(servicedbname)
 	require.NoError(t, err)
 
 	execDB, err := database.NewExecutionDB(execdbname)
 	require.NoError(t, err)
 
-	a, err := New(db, execDB, ContainerOption(container))
+	a, err := New(db, execDB, ContainerOption(cm))
 	require.NoError(t, err)
 
-	closer := func() {
-		require.NoError(t, db.Close())
-		require.NoError(t, execDB.Close())
-		require.NoError(t, os.RemoveAll(servicedbname))
-		require.NoError(t, os.RemoveAll(execdbname))
+	return a, &apiTesting{
+		T:           t,
+		serviceDB:   db,
+		executionDB: execDB,
+		cm:          cm,
 	}
-	return a, dt, closer
+}
+
+var testService = &service.Service{
+	Name: "1",
+	Sid:  "2",
+	Hash: "33",
+	Tasks: []*service.Task{
+		{Key: "4"},
+	},
+	Dependencies: []*service.Dependency{
+		{Key: "5"},
+	},
+}
+
+func TestNotRunningServiceError(t *testing.T) {
+	e := NotRunningServiceError{ServiceID: "test"}
+	require.Equal(t, `Service "test" is not running`, e.Error())
+}
+
+func TestExecuteTask(t *testing.T) {
+	a, at := newTesting(t)
+	defer at.close()
+
+	// TODO(ilgooz): use api.Deploy() instead of manually saving the service
+	// and do the same improvement in the similar places.
+	// in order to do this, create a testing helper to build service tarballs
+	// from yml definitions on the fly .
+	require.NoError(t, at.serviceDB.Save(testService))
+	at.cm.On("Status", mock.Anything).Once().Return(container.RUNNING, nil)
+
+	id, err := a.ExecuteTask("2", "4", map[string]interface{}{}, []string{})
+	require.NoError(t, err)
+	require.NotNil(t, id)
+
+	at.cm.AssertExpectations(t)
+}
+
+func TestExecuteTaskWithInvalidTaskName(t *testing.T) {
+	a, at := newTesting(t)
+	defer at.close()
+
+	require.NoError(t, at.serviceDB.Save(testService))
+	at.cm.On("Status", mock.Anything).Once().Return(container.RUNNING, nil)
+
+	_, err := a.ExecuteTask("2", "2a", map[string]interface{}{}, []string{})
+	require.Error(t, err)
+}
+
+func TestExecuteTaskForNotRunningService(t *testing.T) {
+	a, at := newTesting(t)
+	defer at.close()
+
+	require.NoError(t, at.serviceDB.Save(testService))
+	at.cm.On("Status", mock.Anything).Once().Return(container.STOPPED, nil)
+
+	_, err := a.ExecuteTask("2", "4", map[string]interface{}{}, []string{})
+	_, notRunningError := err.(*NotRunningServiceError)
+	require.True(t, notRunningError)
 }