Implement closing request

Adjust test for CloseRequest.

Change-Id: I0bc0399206f3285cdd8e20eacf1a5d25815e54f7
Signed-off-by: Lukasz Wojciechowski <l.wojciechow@partner.samsung.com>

diff --git a/matcher/jobs.go b/matcher/jobs.go
index 9eb79e8..7b0cbd5 100644 (file)
--- a/matcher/jobs.go
+++ b/matcher/jobs.go
@@ -118,8 +118,11 @@ func (m *JobsManagerImpl) Get(worker WorkerUUID) (*workers.Job, error) {
 }
 
 // Finish closes the job execution, breaks the tunnel to Dryad and removes job
-// from jobs collection. It is a part of JobsManager interface implementation.
+// from jobs collection.
+// The Dryad should be notified and prepared for next job with key regeneration.
+// It is a part of JobsManager interface implementation.
 func (m *JobsManagerImpl) Finish(worker WorkerUUID) error {
+       defer m.workers.PrepareWorker(worker, true)
        m.mutex.Lock()
        defer m.mutex.Unlock()
 

diff --git a/matcher/jobs_test.go b/matcher/jobs_test.go
index f934744..558011d 100644 (file)
--- a/matcher/jobs_test.go
+++ b/matcher/jobs_test.go
@@ -171,6 +171,7 @@ var _ = Describe("Jobs", func() {
                                        ttm.EXPECT().Create(nil, ip).Return(nil),
                                        ttm.EXPECT().Addr().Return(addr),
                                        ttm.EXPECT().Close(),
+                                       w.EXPECT().PrepareWorker(worker, true),
                                )
 
                                err := jm.Create(req, worker)
@@ -182,6 +183,7 @@ var _ = Describe("Jobs", func() {
                                Expect(jm.(*JobsManagerImpl).jobs).To(BeEmpty())
                        })
                        It("should fail to finish nonexistent job", func() {
+                               w.EXPECT().PrepareWorker(worker, true)
                                err := jm.Finish(worker)
                                Expect(err).To(Equal(NotFoundError("Job")))
                        })

diff --git a/requests/requests.go b/requests/requests.go
index adc93fb..89253d6 100644 (file)
--- a/requests/requests.go
+++ b/requests/requests.go
@@ -130,8 +130,9 @@ func (reqs *ReqsCollection) NewRequest(caps Capabilities,
 // all required conditions to close request are met.
 // The method must be called in reqs.mutex critical section.
 func (reqs *ReqsCollection) closeRequest(req *ReqInfo) {
+       worker := req.Job.WorkerUUID
+       reqs.jobs.Finish(worker)
        req.State = DONE
-       // TODO(mwereski): release worker
 }
 
 // CloseRequest is part of implementation of Requests interface.

diff --git a/requests/requests_requestsmanager_test.go b/requests/requests_requestsmanager_test.go
index 1a23ce7..42e44fd 100644 (file)
--- a/requests/requests_requestsmanager_test.go
+++ b/requests/requests_requestsmanager_test.go
@@ -31,13 +31,15 @@ var _ = Describe("Requests as RequestsManager", func() {
        Describe("With RequestsManager created", func() {
                var ctrl *gomock.Controller
                var wm *MockWorkersManager
+               var jm *MockJobsManager
                var R *ReqsCollection
                testErr := errors.New("Test Error")
 
                BeforeEach(func() {
                        ctrl = gomock.NewController(GinkgoT())
                        wm = NewMockWorkersManager(ctrl)
-                       R = NewRequestQueue(wm, nil)
+                       jm = NewMockJobsManager(ctrl)
+                       R = NewRequestQueue(wm, jm)
                })
                AfterEach(func() {
                        R.Finish()
@@ -288,16 +290,20 @@ var _ = Describe("Requests as RequestsManager", func() {
                                        Expect(err).To(Equal(ErrModificationForbidden))
                                })
                                It("should close request and release worker", func() {
+                                       testWorker := WorkerUUID("TestWorker")
                                        R.mutex.Lock()
                                        rinfo.State = INPROGRESS
                                        rinfo.Job = &JobInfo{
-                                               Timeout: time.Now().AddDate(0, 0, -1),
+                                               Timeout:    time.Now().AddDate(0, 0, -1),
+                                               WorkerUUID: testWorker,
                                        }
                                        R.mutex.Unlock()
+                                       gomock.InOrder(
+                                               jm.EXPECT().Finish(testWorker),
+                                       )
                                        err := R.Close(req)
                                        Expect(err).NotTo(HaveOccurred())
                                        Expect(rinfo.State).To(Equal(DONE))
-                                       // TODO verify releasing worker when implemented
                                })
                        })
                        Describe("Run", func() {

diff --git a/requests/requests_test.go b/requests/requests_test.go
index 30cb9c1..6c414f6 100644 (file)
--- a/requests/requests_test.go
+++ b/requests/requests_test.go
@@ -120,9 +120,13 @@ func TestNewRequest(t *testing.T) {
 }
 
 func TestCloseRequest(t *testing.T) {
-       assert, rqueue, ctrl, _ := initTest(t)
+       assert, rqueue, ctrl, jm := initTest(t)
        defer finiTest(rqueue, ctrl)
+
        req := requestsTests[0].req
+       jobInfo := JobInfo{
+               WorkerUUID: "Test WorkerUUID",
+       }
 
        // Add valid request to the queue.
        reqid, err := rqueue.NewRequest(req.Caps, req.Priority, req.Owner, req.ValidAfter, req.Deadline)
@@ -152,9 +156,13 @@ func TestCloseRequest(t *testing.T) {
        assert.Nil(err)
        rqueue.mutex.Lock()
        rqueue.requests[reqid].State = INPROGRESS
+       rqueue.requests[reqid].Job = &jobInfo
        rqueue.queue.removeRequest(&reqinfo)
        rqueue.mutex.Unlock()
        // Close request.
+       gomock.InOrder(
+               jm.EXPECT().Finish(jobInfo.WorkerUUID),
+       )
        err = rqueue.CloseRequest(reqid)
        assert.Nil(err)
        rqueue.mutex.RLock()
@@ -470,7 +478,6 @@ func TestListRequests(t *testing.T) {
 func TestAcquireWorker(t *testing.T) {
        assert, rqueue, ctrl, jm := initTest(t)
        defer finiTest(rqueue, ctrl)
-
        req := requestsTests[0].req
        empty := AccessInfo{}
        testErr := errors.New("Test Error")