Roughing in some transfer tests.

Author: jeff-cohere

databases/databases.go

@@ -132,11 +132,11 @@ const (
 // registers a database creation function under the given database name
 // to allow for e.g. test database implementations
 func RegisterDatabase(dbName string, createDb func() (Database, error)) error {
-	if firstTime {
+	if firstTime_ {
 		// register types that appear in Frictionless Descriptors (for manifests)
 		gob.Register(credit.CreditMetadata{})
 
-		firstTime = false
+		firstTime_ = false
 	}
 
 	// make one to check the configuration
@@ -155,6 +155,15 @@ func RegisterDatabase(dbName string, createDb func() (Database, error)) error {
 	}
 }
 
+// returns a list of names of registered databases
+func RegisteredDatabases() []string {
+	dbs := make([]string, 0)
+	for name, _ := range createDatabaseFuncs_ {
+		dbs = append(dbs, name)
+	}
+	return dbs
+}
+
 // returns true if a database has been registered with the given name, false if not
 func HaveDatabase(dbName string) bool {
 	_, found := createDatabaseFuncs_[dbName]
@@ -222,7 +231,7 @@ func Load(states DatabaseSaveStates) error {
 //-----------
 
 // set to false after the first database is registered
-var firstTime = true
+var firstTime_ = true
 
 // we maintain a table of database instances, identified by their names
 var allDatabases_ = make(map[string]Database)

transfers/dispatcher.go

@@ -54,6 +54,16 @@ type dispatcherChannels struct {
 	Stop  chan struct{} // used by client to stop task management
 }
 
+func (channels *dispatcherChannels) close() {
+	close(channels.RequestTransfer)
+	close(channels.ReturnTransferId)
+	close(channels.CancelTransfer)
+	close(channels.RequestStatus)
+	close(channels.ReturnStatus)
+	close(channels.Error)
+	close(channels.Stop)
+}
+
 func (d *dispatcherState) Start() error {
 	d.Channels = dispatcherChannels{
 		RequestTransfer:  make(chan Specification, 32),
@@ -71,7 +81,9 @@ func (d *dispatcherState) Start() error {
 
 func (d *dispatcherState) Stop() error {
 	d.Channels.Stop <- struct{}{}
-	return <-d.Channels.Error
+	err := <-d.Channels.Error
+	d.Channels.close()
+	return err
 }
 
 func (d *dispatcherState) CreateTransfer(spec Specification) (uuid.UUID, error) {
@@ -99,9 +111,9 @@ func (d *dispatcherState) CancelTransfer(id uuid.UUID) error {
 	return <-d.Channels.Error
 }
 
-//----------------------------------------------------
+//---------------------------------------------------------
 // everything past here runs in the dispatcher's goroutine
-//----------------------------------------------------
+//---------------------------------------------------------
 
 // the goroutine itself
 func (d *dispatcherState) process() {
@@ -125,6 +137,7 @@ func (d *dispatcherState) process() {
 			transferId, numFiles, err := d.create(spec)
 			if err != nil {
 				returnError <- err
+				break
 			}
 			returnTransferId <- transferId
 			slog.Info(fmt.Sprintf("Created new transfer %s (%d file(s) requested)", transferId.String(),
@@ -144,6 +157,7 @@ func (d *dispatcherState) process() {
 			returnStatus <- status
 		case <-stopRequested:
 			running = false
+			returnError <- nil
 		}
 	}
 }

transfers/dispatcher_test.go

@@ -0,0 +1,36 @@
+// Copyright (c) 2023 The KBase Project and its Contributors
+// Copyright (c) 2023 Cohere Consulting, LLC
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+// of the Software, and to permit persons to whom the Software is furnished to do
+// so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package transfers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStartAndStopDispatcher(t *testing.T) {
+	assert := assert.New(t)
+	err := dispatcher.Start()
+	assert.Nil(err)
+	err = dispatcher.Stop()
+	assert.Nil(err)
+}

transfers/manifestor.go

@@ -57,6 +57,13 @@ type manifestorChannels struct {
 	Stop                chan struct{}
 }
 
+func (channels *manifestorChannels) close() {
+	close(channels.RequestGeneration)
+	close(channels.RequestCancellation)
+	close(channels.Error)
+	close(channels.Stop)
+}
+
 // starts the mover
 func (m *manifestorState) Start() error {
 	m.Channels = manifestorChannels{
@@ -73,7 +80,9 @@ func (m *manifestorState) Start() error {
 // stops the manifestor goroutine
 func (m *manifestorState) Stop() error {
 	m.Channels.Stop <- struct{}{}
-	return <-m.Channels.Error
+	err := <-m.Channels.Error
+	m.Channels.close()
+	return err
 }
 
 // starts generating a manifest for the given transfer, moving it subsequently to that transfer's
@@ -117,8 +126,9 @@ func (m *manifestorState) process() {
 			} else {
 				manifestor.Channels.Error <- NotFoundError{Id: transferId}
 			}
-		case <-mover.Channels.Stop:
+		case <-manifestor.Channels.Stop:
 			running = false
+			manifestor.Channels.Error <- nil
 		}
 
 		time.Sleep(pollInterval)

transfers/manifestor_test.go

@@ -0,0 +1,36 @@
+// Copyright (c) 2023 The KBase Project and its Contributors
+// Copyright (c) 2023 Cohere Consulting, LLC
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+// of the Software, and to permit persons to whom the Software is furnished to do
+// so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package transfers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStartAndStopManifestor(t *testing.T) {
+	assert := assert.New(t)
+	err := manifestor.Start()
+	assert.Nil(err)
+	err = manifestor.Stop()
+	assert.Nil(err)
+}

transfers/mover.go

@@ -52,10 +52,18 @@ type moverChannels struct {
 	Stop                chan struct{}
 }
 
+func (channels *moverChannels) close() {
+	close(channels.RequestMove)
+	close(channels.RequestCancellation)
+	close(channels.Error)
+	close(channels.Stop)
+}
+
 // starts the mover
 func (m *moverState) Start() error {
 	m.Channels = moverChannels{
 		RequestMove: make(chan uuid.UUID, 32),
+		RequestCancellation: make(chan uuid.UUID, 32),
 		Error:       make(chan error, 32),
 		Stop:        make(chan struct{}),
 	}
@@ -67,7 +75,9 @@ func (m *moverState) Start() error {
 // stops the mover goroutine
 func (m *moverState) Stop() error {
 	m.Channels.Stop <- struct{}{}
-	return <-m.Channels.Error
+	err := <-m.Channels.Error
+	m.Channels.close()
+	return err
 }
 
 // starts moving files associated with the given transfer ID
@@ -111,6 +121,7 @@ func (m *moverState) process() {
 			}
 		case <-mover.Channels.Stop:
 			running = false
+			mover.Channels.Error <- nil
 		}
 
 		time.Sleep(pollInterval)

transfers/mover_test.go

@@ -0,0 +1,36 @@
+// Copyright (c) 2023 The KBase Project and its Contributors
+// Copyright (c) 2023 Cohere Consulting, LLC
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+// of the Software, and to permit persons to whom the Software is furnished to do
+// so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package transfers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStartAndStopMover(t *testing.T) {
+	assert := assert.New(t)
+	err := mover.Start()
+	assert.Nil(err)
+	err = mover.Stop()
+	assert.Nil(err)
+}

transfers/stager.go

@@ -51,10 +51,18 @@ type stagerChannels struct {
 	Stop                chan struct{}
 }
 
+func (channels *stagerChannels) close() {
+	close(channels.RequestStaging)
+	close(channels.RequestCancellation)
+	close(channels.Error)
+	close(channels.Stop)
+}
+
 // starts the stager
 func (s *stagerState) Start() error {
 	s.Channels = stagerChannels{
 		RequestStaging: make(chan uuid.UUID, 32),
+		RequestCancellation: make(chan uuid.UUID, 32),
 		Error:          make(chan error, 32),
 		Stop:           make(chan struct{}),
 	}
@@ -65,7 +73,9 @@ func (s *stagerState) Start() error {
 // stops the stager goroutine
 func (s *stagerState) Stop() error {
 	s.Channels.Stop <- struct{}{}
-	return <-s.Channels.Error
+	err := <-s.Channels.Error
+	s.Channels.close()
+	return err
 }
 
 // requests that files be staged for the transfer with the given ID
@@ -106,6 +116,7 @@ func (s *stagerState) process() {
 			}
 		case <-stager.Channels.Stop:
 			running = false
+			stager.Channels.Error <- nil
 		}
 
 		time.Sleep(pollInterval)

transfers/stager_test.go

@@ -0,0 +1,36 @@
+// Copyright (c) 2023 The KBase Project and its Contributors
+// Copyright (c) 2023 Cohere Consulting, LLC
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+// of the Software, and to permit persons to whom the Software is furnished to do
+// so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package transfers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStartAndStopStager(t *testing.T) {
+	assert := assert.New(t)
+	err := stager.Start()
+	assert.Nil(err)
+	err = stager.Stop()
+	assert.Nil(err)
+}

transfers/store.go

@@ -61,6 +61,21 @@ type storeChannels struct {
 	Stop  chan struct{}
 }
 
+func (channels *storeChannels) close() {
+	close(channels.RequestNewTransfer)
+	close(channels.ReturnNewTransfer)
+	close(channels.RequestSpec)
+	close(channels.ReturnSpec)
+	close(channels.RequestDescriptors)
+	close(channels.ReturnDescriptors)
+	close(channels.SetStatus)
+	close(channels.RequestStatus)
+	close(channels.ReturnStatus)
+	close(channels.RequestRemoval)
+	close(channels.Error)
+	close(channels.Stop)
+}
+
 type transferIdAndStatus struct {
 	Id     uuid.UUID
 	Status TransferStatus
@@ -88,7 +103,10 @@ func (s *storeState) Start() error {
 
 // stops the store goroutine
 func (s *storeState) Stop() error {
-	return nil
+	s.Channels.Stop <- struct{}{}
+	err := <-s.Channels.Error
+	s.Channels.close()
+	return err
 }
 
 // creates a new entry for a transfer within the store, populating it with relevant metadata and
@@ -205,6 +223,7 @@ func (s *storeState) process() {
 				store.Channels.Error <- TransferNotFoundError{Id: id}
 			}
 		case <-store.Channels.Stop:
+			store.Channels.Error <- nil
 			running = false
 		}
 	}