diff --git a/CHANGELOG.md b/CHANGELOG.md index 0363bd27ce7bb53382ca7f4733529de2a547784c..62833bd3834784a3364447e0dad87ef8042af25e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +## 20.03.1 + +IMPROVEMENTS +* change behavior when trying to get data from a substream that not exist - return EndOfStream/NoData instead of WrongInput + ## 20.03 FEATURES * introduced substreams for producer/consumer diff --git a/broker/src/asapo_broker/database/mongodb.go b/broker/src/asapo_broker/database/mongodb.go index fe9ab0f7ee0279e3de39235dc8fbac1b648ea6d1..b219654128adefd675da8199549614cdb9e8ed15 100644 --- a/broker/src/asapo_broker/database/mongodb.go +++ b/broker/src/asapo_broker/database/mongodb.go @@ -307,10 +307,14 @@ func (db *Mongodb) checkDatabaseOperationPrerequisites(db_name string, collectio return &DBError{utils.StatusServiceUnavailable, no_session_msg} } - if err := db.getParentDB().dataBaseExist(db_name); err != nil { - return err + if len(db_name)==0 || len(collection_name) ==0 { + return &DBError{utils.StatusWrongInput, "beamtime_id ans substream must be set"} } + // if err := db.getParentDB().dataBaseExist(db_name); err != nil { +// return err +// } + if len(group_id) > 0 { db.getParentDB().generateLocationPointersInDbIfNeeded(db_name, collection_name, group_id) } @@ -471,7 +475,7 @@ func (db *Mongodb) getSubstreams(db_name string) ([]byte, error) { return db.processQueryError("get substreams", db_name, err) } - var rec SubstreamsRecord + var rec = SubstreamsRecord{[]string{}} for _, coll := range result { if strings.HasPrefix(coll, data_collection_name_prefix) { rec.Substreams = append(rec.Substreams, strings.TrimPrefix(coll, data_collection_name_prefix)) diff --git a/broker/src/asapo_broker/database/mongodb_test.go b/broker/src/asapo_broker/database/mongodb_test.go index 9f11a8a5871778e1d6193825f30734e9d948fc60..c0e13537896aafd3af6fb30d312d04d4244a2df1 100644 --- a/broker/src/asapo_broker/database/mongodb_test.go +++ b/broker/src/asapo_broker/database/mongodb_test.go @@ -4,11 +4,8 @@ package database import ( "asapo_common/utils" - "context" "encoding/json" "github.com/stretchr/testify/assert" - "go.mongodb.org/mongo-driver/bson" - "go.mongodb.org/mongo-driver/mongo/options" "sync" "testing" ) @@ -97,21 +94,32 @@ func TestMongoDBGetNextErrorWhenWrongDatabasename(t *testing.T) { assert.Equal(t, utils.StatusWrongInput, err.(*DBError).Code) } -func TestMongoDBGetNextErrorWhenWrongDatacollectionname(t *testing.T) { +func TestMongoDBGetNextErrorWhenNonExistingDatacollectionname(t *testing.T) { db.Connect(dbaddress) defer cleanup() - _, err := db.ProcessRequest(dbname, "", groupId, "next", "") - assert.Equal(t, utils.StatusWrongInput, err.(*DBError).Code) + _, err := db.ProcessRequest(dbname, "bla", groupId, "next", "") + assert.Equal(t, utils.StatusNoData, err.(*DBError).Code) + assert.Equal(t, "{\"op\":\"get_record_by_id\",\"id\":0,\"id_max\":0,\"next_substream\":\"\"}", err.Error()) } -func TestMongoDBGetNextErrorWhenEmptyCollection(t *testing.T) { +func TestMongoDBGetLastErrorWhenNonExistingDatacollectionname(t *testing.T) { db.Connect(dbaddress) - db.databases = append(db.databases, dbname) defer cleanup() - _, err := db.ProcessRequest(dbname, collection, groupId, "next", "") + _, err := db.ProcessRequest(dbname, "bla", groupId, "last", "") + assert.Equal(t, utils.StatusNoData, err.(*DBError).Code) + assert.Equal(t, "{\"op\":\"get_record_by_id\",\"id\":0,\"id_max\":0,\"next_substream\":\"\"}", err.Error()) +} + +func TestMongoDBGetByIdErrorWhenNonExistingDatacollectionname(t *testing.T) { + db.Connect(dbaddress) + defer cleanup() + _, err := db.ProcessRequest(dbname, collection, groupId, "id", "2") + assert.Equal(t, utils.StatusNoData, err.(*DBError).Code) + assert.Equal(t, "{\"op\":\"get_record_by_id\",\"id\":2,\"id_max\":0,\"next_substream\":\"\"}", err.Error()) } + func TestMongoDBGetNextErrorWhenRecordNotThereYet(t *testing.T) { db.Connect(dbaddress) defer cleanup() @@ -321,22 +329,12 @@ func TestMongoDBGetSize(t *testing.T) { func TestMongoDBGetSizeNoRecords(t *testing.T) { db.Connect(dbaddress) defer cleanup() - // to have empty collection - db.insertRecord(dbname, collection, &rec1) - db.client.Database(dbname).Collection(data_collection_name_prefix+collection).DeleteOne(context.TODO(), bson.M{"_id": 1}, options.Delete()) res, err := db.ProcessRequest(dbname, collection, "", "size", "0") assert.Nil(t, err) assert.Equal(t, string(recs2_expect), string(res)) } -func TestMongoDBGetSizeNoDatabase(t *testing.T) { - db.Connect(dbaddress) - defer cleanup() - _, err := db.ProcessRequest(dbname, collection, "", "size", "0") - assert.NotNil(t, err) -} - func TestMongoPing(t *testing.T) { db.Connect(dbaddress) defer cleanup() @@ -480,6 +478,22 @@ func TestMongoDBQueryImagesOK(t *testing.T) { } +func TestMongoDBQueryImagesOnEmptyDatabase(t *testing.T) { + db.Connect(dbaddress) + defer cleanup() + for _, test := range tests { + res_string, err := db.ProcessRequest(dbname, collection, "", "queryimages", test.query) + var res []TestRecordMeta + json.Unmarshal(res_string, &res) + assert.Equal(t, 0, len(res)) + if test.ok { + assert.Nil(t, err, test.query) + } else { + assert.NotNil(t, err, test.query) + } + } +} + var rec_dataset1 = TestDataset{1, 3, []TestRecord{rec1, rec2, rec3}} var rec_dataset1_incomplete = TestDataset{1, 4, []TestRecord{rec1, rec2, rec3}} var rec_dataset2 = TestDataset{2, 4, []TestRecord{rec1, rec2, rec3}} @@ -570,7 +584,7 @@ var testsSubstreams = []struct { test string ok bool }{ - {SubstreamsRecord{[]string{}}, "no substreams", false}, + {SubstreamsRecord{[]string{}}, "no substreams", true}, {SubstreamsRecord{[]string{"ss1"}}, "one substream", true}, {SubstreamsRecord{[]string{"ss1", "ss2"}}, "two substreams", true}, } diff --git a/broker/src/asapo_broker/server/process_request_test.go b/broker/src/asapo_broker/server/process_request_test.go index 818d0040983b9f286ebea5625921b6c9891b0931..f4198f5e662b282688bdffd7819ba8e042aa0a20 100644 --- a/broker/src/asapo_broker/server/process_request_test.go +++ b/broker/src/asapo_broker/server/process_request_test.go @@ -122,13 +122,13 @@ func (suite *ProcessRequestTestSuite) TestProcessRequestWithNoToken() { func (suite *ProcessRequestTestSuite) TestProcessRequestWithWrongDatabaseName() { suite.mock_db.On("ProcessRequest", expectedDBName, expectedSubstream, expectedGroupID, "next", "0").Return([]byte(""), - &database.DBError{utils.StatusWrongInput, ""}) + &database.DBError{utils.StatusNoData, ""}) - logger.MockLog.On("Error", mock.MatchedBy(containsMatcher("processing request next"))) + logger.MockLog.On("Debug", mock.MatchedBy(containsMatcher("processing request next"))) w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedStream + "/" + expectedSubstream + "/" + expectedGroupID + "/next" + correctTokenSuffix) - suite.Equal(http.StatusBadRequest, w.Code, "wrong database name") + suite.Equal(http.StatusConflict, w.Code, "wrong database name") } func (suite *ProcessRequestTestSuite) TestProcessRequestWithConnectionError() { @@ -141,7 +141,7 @@ func (suite *ProcessRequestTestSuite) TestProcessRequestWithConnectionError() { w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedStream + "/" + expectedSubstream + "/" + expectedGroupID + "/next" + correctTokenSuffix) time.Sleep(time.Second) - suite.Equal(http.StatusNotFound, w.Code, "wrong database name") + suite.Equal(http.StatusNotFound, w.Code, "data not found") } func (suite *ProcessRequestTestSuite) TestProcessRequestWithInternalDBError() {