We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add tests for evaluation metrics reporting service
We previously added service to report evaluation metrics in #510. However, there isn't any test coverage for the added code yet. We should add some tests to cover `MasterServicer.ReportEvaluationMetrics`.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticdl/python/elasticdl/master/servicer.py`
Content:
```
1 import logging
2 import threading
3 import numpy as np
4 
5 import tensorflow as tf
6 
7 assert tf.executing_eagerly()
8 
9 from google.protobuf import empty_pb2
10 
11 from elasticdl.proto import elasticdl_pb2
12 from elasticdl.proto import elasticdl_pb2_grpc
13 from elasticdl.python.elasticdl.common.ndarray import ndarray_to_tensor, tensor_to_ndarray
14 
15 
16 class MasterServicer(elasticdl_pb2_grpc.MasterServicer):
17     """Master service implementation"""
18 
19     def __init__(
20         self,
21         grads_to_wait,
22         minibatch_size,
23         optimizer,
24         task_q,
25         *,
26         init_var=[]
27     ):
28         # TODO: group params together into a single object.
29         self._logger = logging.getLogger(__name__)
30         self._opt = optimizer
31         self._task_q = task_q
32         self._lock = threading.Lock()
33         # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable
34         # instead ndarray to avoid copying and conversion when calling
35         # optimizer's apply_gradients() function.
36         self._model = {}
37         self._version = 0
38         self._gradient_sum = {}
39         self._grad_to_wait = grads_to_wait
40         self._grad_n = 0
41         self._minibatch_size = minibatch_size
42         self._evaluation_metrics = {}
43         for var in init_var:
44             self.set_model_var(var.name, var.numpy())
45 
46     def set_model_var(self, name, value):
47         """Add or set model variable. Value should be a float32 ndarray"""
48         if value.dtype != np.float32:
49             raise ValueError("Value should be a float32 numpy array")
50         self._model[name] = tf.Variable(
51             value, name=MasterServicer.var_name_encode(name)
52         )
53 
54     @staticmethod
55     def var_name_encode(name):
56         return name.replace(":", "-")
57 
58     def GetTask(self, request, _):
59         res = elasticdl_pb2.Task()
60         res.model_version = self._version
61         res.minibatch_size = self._minibatch_size
62         task_id, task = self._task_q.get(request.worker_id)
63         if task:
64             res.task_id = task_id
65             res.shard_file_name = task.file_name
66             res.start = task.start
67             res.end = task.end
68         return res
69 
70     def GetModel(self, request, _):
71         _ = self._validate_model_version(request.min_version)
72 
73         res = elasticdl_pb2.Model()
74         with self._lock:
75             res.version = self._version
76             for k, v in self._model.items():
77                 res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))
78         return res
79 
80     def _update_model(self):
81         assert self._lock.locked()
82         grad_var = []
83         for k in self._gradient_sum:
84             self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait
85             grad_var.append((self._gradient_sum[k], self._model[k]))
86         self._opt.apply_gradients(grad_var)
87         self._version += 1
88         self._gradient_sum.clear()
89         self._grad_n = 0
90 
91     def _validate_model_version(self, request_model_version):
92         if request_model_version > self._version:
93             err_msg = "Model version %d not available yet, current version: %d" % (
94                 request_model_version,
95                 self._version,
96             )
97             self._logger.warning(err_msg)
98             raise ValueError(err_msg)
99 
100         invalid_model_version = request_model_version < self._version
101         if invalid_model_version:
102             self._logger.warning(
103                 "Task result for outdated version %d dropped",
104                 request_model_version,
105             )
106         return invalid_model_version
107 
108     def ReportGradient(self, request, _):
109         invalid_model_version = self._validate_model_version(request.model_version)
110 
111         res = elasticdl_pb2.ReportGradientResponse()
112         if invalid_model_version:
113             res.accepted = False
114             res.model_version = self._version
115             return res
116 
117         # TODO: Update task queue with task_id
118         with self._lock:
119             tmp = {}
120             # Do sanity check before accumulating gradients.
121             for k, v in request.gradient.items():
122                 if k not in self._model:
123                     raise ValueError(
124                         "Gradient key: %s is not part of model", k
125                     )
126                 arr = tensor_to_ndarray(v)
127                 if arr.shape != self._model[k].numpy().shape:
128                     raise ValueError(
129                         "Gradient key: %s has incompatible dimension", k
130                     )
131                 tmp[k] = arr
132 
133             for k, v in tmp.items():
134                 if k in self._gradient_sum:
135                     self._gradient_sum[k] = self._gradient_sum[k] + v
136                 else:
137                     self._gradient_sum[k] = v
138 
139             self._grad_n += 1
140             if self._grad_n >= self._grad_to_wait:
141                 self._update_model()
142         res.accepted = True
143         res.model_version = self._version
144         return res
145 
146     def ReportTaskResult(self, request, _):
147         if request.err_message:
148             self._logger.warning(
149                 "Worker reported error: " + request.err_message
150             )
151             self._task_q.report(request.task_id, False)
152         else:
153             self._task_q.report(request.task_id, True)
154         return empty_pb2.Empty()
155 
156     def ReportEvaluationMetrics(self, request, _):
157         invalid_model_version = self._validate_model_version(request.model_version)
158 
159         res = elasticdl_pb2.ReportEvaluationMetricsResponse()
160         if invalid_model_version:
161             res.accepted = False
162             res.model_version = self._version
163             return res
164 
165         with self._lock:
166             for k, v in request.evaluation_metrics.items():
167                 arr = tensor_to_ndarray(v)
168                 self._evaluation_metrics[k] = arr
169 
170             self._update_model()
171         res.accepted = True
172         res.model_version = self._version
173         return res
174
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/elasticdl/python/elasticdl/master/servicer.py b/elasticdl/python/elasticdl/master/servicer.py
--- a/elasticdl/python/elasticdl/master/servicer.py
+++ b/elasticdl/python/elasticdl/master/servicer.py
@@ -77,6 +77,10 @@
                 res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))
         return res
 
+    def _update_model_version(self):
+        assert self._lock.locked()
+        self._version += 1
+
     def _update_model(self):
         assert self._lock.locked()
         grad_var = []
@@ -84,7 +88,7 @@
             self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait
             grad_var.append((self._gradient_sum[k], self._model[k]))
         self._opt.apply_gradients(grad_var)
-        self._version += 1
+        self._update_model_version()
         self._gradient_sum.clear()
         self._grad_n = 0
 
@@ -167,7 +171,7 @@
                 arr = tensor_to_ndarray(v)
                 self._evaluation_metrics[k] = arr
 
-            self._update_model()
+            self._update_model_version()
         res.accepted = True
         res.model_version = self._version
         return res

{"golden_diff": "diff --git a/elasticdl/python/elasticdl/master/servicer.py b/elasticdl/python/elasticdl/master/servicer.py\n--- a/elasticdl/python/elasticdl/master/servicer.py\n+++ b/elasticdl/python/elasticdl/master/servicer.py\n@@ -77,6 +77,10 @@\n                 res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))\n         return res\n \n+    def _update_model_version(self):\n+        assert self._lock.locked()\n+        self._version += 1\n+\n     def _update_model(self):\n         assert self._lock.locked()\n         grad_var = []\n@@ -84,7 +88,7 @@\n             self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait\n             grad_var.append((self._gradient_sum[k], self._model[k]))\n         self._opt.apply_gradients(grad_var)\n-        self._version += 1\n+        self._update_model_version()\n         self._gradient_sum.clear()\n         self._grad_n = 0\n \n@@ -167,7 +171,7 @@\n                 arr = tensor_to_ndarray(v)\n                 self._evaluation_metrics[k] = arr\n \n-            self._update_model()\n+            self._update_model_version()\n         res.accepted = True\n         res.model_version = self._version\n         return res\n", "issue": "Add tests for evaluation metrics reporting service\nWe previously added service to report evaluation metrics in #510. However, there isn't any test coverage for the added code yet. We should add some tests to cover `MasterServicer.ReportEvaluationMetrics`.\n", "before_files": [{"content": "import logging\nimport threading\nimport numpy as np\n\nimport tensorflow as tf\n\nassert tf.executing_eagerly()\n\nfrom google.protobuf import empty_pb2\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.proto import elasticdl_pb2_grpc\nfrom elasticdl.python.elasticdl.common.ndarray import ndarray_to_tensor, tensor_to_ndarray\n\n\nclass MasterServicer(elasticdl_pb2_grpc.MasterServicer):\n    \"\"\"Master service implementation\"\"\"\n\n    def __init__(\n        self,\n        grads_to_wait,\n        minibatch_size,\n        optimizer,\n        task_q,\n        *,\n        init_var=[]\n    ):\n        # TODO: group params together into a single object.\n        self._logger = logging.getLogger(__name__)\n        self._opt = optimizer\n        self._task_q = task_q\n        self._lock = threading.Lock()\n        # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable\n        # instead ndarray to avoid copying and conversion when calling\n        # optimizer's apply_gradients() function.\n        self._model = {}\n        self._version = 0\n        self._gradient_sum = {}\n        self._grad_to_wait = grads_to_wait\n        self._grad_n = 0\n        self._minibatch_size = minibatch_size\n        self._evaluation_metrics = {}\n        for var in init_var:\n            self.set_model_var(var.name, var.numpy())\n\n    def set_model_var(self, name, value):\n        \"\"\"Add or set model variable. Value should be a float32 ndarray\"\"\"\n        if value.dtype != np.float32:\n            raise ValueError(\"Value should be a float32 numpy array\")\n        self._model[name] = tf.Variable(\n            value, name=MasterServicer.var_name_encode(name)\n        )\n\n    @staticmethod\n    def var_name_encode(name):\n        return name.replace(\":\", \"-\")\n\n    def GetTask(self, request, _):\n        res = elasticdl_pb2.Task()\n        res.model_version = self._version\n        res.minibatch_size = self._minibatch_size\n        task_id, task = self._task_q.get(request.worker_id)\n        if task:\n            res.task_id = task_id\n            res.shard_file_name = task.file_name\n            res.start = task.start\n            res.end = task.end\n        return res\n\n    def GetModel(self, request, _):\n        _ = self._validate_model_version(request.min_version)\n\n        res = elasticdl_pb2.Model()\n        with self._lock:\n            res.version = self._version\n            for k, v in self._model.items():\n                res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))\n        return res\n\n    def _update_model(self):\n        assert self._lock.locked()\n        grad_var = []\n        for k in self._gradient_sum:\n            self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait\n            grad_var.append((self._gradient_sum[k], self._model[k]))\n        self._opt.apply_gradients(grad_var)\n        self._version += 1\n        self._gradient_sum.clear()\n        self._grad_n = 0\n\n    def _validate_model_version(self, request_model_version):\n        if request_model_version > self._version:\n            err_msg = \"Model version %d not available yet, current version: %d\" % (\n                request_model_version,\n                self._version,\n            )\n            self._logger.warning(err_msg)\n            raise ValueError(err_msg)\n\n        invalid_model_version = request_model_version < self._version\n        if invalid_model_version:\n            self._logger.warning(\n                \"Task result for outdated version %d dropped\",\n                request_model_version,\n            )\n        return invalid_model_version\n\n    def ReportGradient(self, request, _):\n        invalid_model_version = self._validate_model_version(request.model_version)\n\n        res = elasticdl_pb2.ReportGradientResponse()\n        if invalid_model_version:\n            res.accepted = False\n            res.model_version = self._version\n            return res\n\n        # TODO: Update task queue with task_id\n        with self._lock:\n            tmp = {}\n            # Do sanity check before accumulating gradients.\n            for k, v in request.gradient.items():\n                if k not in self._model:\n                    raise ValueError(\n                        \"Gradient key: %s is not part of model\", k\n                    )\n                arr = tensor_to_ndarray(v)\n                if arr.shape != self._model[k].numpy().shape:\n                    raise ValueError(\n                        \"Gradient key: %s has incompatible dimension\", k\n                    )\n                tmp[k] = arr\n\n            for k, v in tmp.items():\n                if k in self._gradient_sum:\n                    self._gradient_sum[k] = self._gradient_sum[k] + v\n                else:\n                    self._gradient_sum[k] = v\n\n            self._grad_n += 1\n            if self._grad_n >= self._grad_to_wait:\n                self._update_model()\n        res.accepted = True\n        res.model_version = self._version\n        return res\n\n    def ReportTaskResult(self, request, _):\n        if request.err_message:\n            self._logger.warning(\n                \"Worker reported error: \" + request.err_message\n            )\n            self._task_q.report(request.task_id, False)\n        else:\n            self._task_q.report(request.task_id, True)\n        return empty_pb2.Empty()\n\n    def ReportEvaluationMetrics(self, request, _):\n        invalid_model_version = self._validate_model_version(request.model_version)\n\n        res = elasticdl_pb2.ReportEvaluationMetricsResponse()\n        if invalid_model_version:\n            res.accepted = False\n            res.model_version = self._version\n            return res\n\n        with self._lock:\n            for k, v in request.evaluation_metrics.items():\n                arr = tensor_to_ndarray(v)\n                self._evaluation_metrics[k] = arr\n\n            self._update_model()\n        res.accepted = True\n        res.model_version = self._version\n        return res\n", "path": "elasticdl/python/elasticdl/master/servicer.py"}], "after_files": [{"content": "import logging\nimport threading\nimport numpy as np\n\nimport tensorflow as tf\n\nassert tf.executing_eagerly()\n\nfrom google.protobuf import empty_pb2\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.proto import elasticdl_pb2_grpc\nfrom elasticdl.python.elasticdl.common.ndarray import ndarray_to_tensor, tensor_to_ndarray\n\n\nclass MasterServicer(elasticdl_pb2_grpc.MasterServicer):\n    \"\"\"Master service implementation\"\"\"\n\n    def __init__(\n        self,\n        grads_to_wait,\n        minibatch_size,\n        optimizer,\n        task_q,\n        *,\n        init_var=[]\n    ):\n        # TODO: group params together into a single object.\n        self._logger = logging.getLogger(__name__)\n        self._opt = optimizer\n        self._task_q = task_q\n        self._lock = threading.Lock()\n        # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable\n        # instead ndarray to avoid copying and conversion when calling\n        # optimizer's apply_gradients() function.\n        self._model = {}\n        self._version = 0\n        self._gradient_sum = {}\n        self._grad_to_wait = grads_to_wait\n        self._grad_n = 0\n        self._minibatch_size = minibatch_size\n        self._evaluation_metrics = {}\n        for var in init_var:\n            self.set_model_var(var.name, var.numpy())\n\n    def set_model_var(self, name, value):\n        \"\"\"Add or set model variable. Value should be a float32 ndarray\"\"\"\n        if value.dtype != np.float32:\n            raise ValueError(\"Value should be a float32 numpy array\")\n        self._model[name] = tf.Variable(\n            value, name=MasterServicer.var_name_encode(name)\n        )\n\n    @staticmethod\n    def var_name_encode(name):\n        return name.replace(\":\", \"-\")\n\n    def GetTask(self, request, _):\n        res = elasticdl_pb2.Task()\n        res.model_version = self._version\n        res.minibatch_size = self._minibatch_size\n        task_id, task = self._task_q.get(request.worker_id)\n        if task:\n            res.task_id = task_id\n            res.shard_file_name = task.file_name\n            res.start = task.start\n            res.end = task.end\n        return res\n\n    def GetModel(self, request, _):\n        _ = self._validate_model_version(request.min_version)\n\n        res = elasticdl_pb2.Model()\n        with self._lock:\n            res.version = self._version\n            for k, v in self._model.items():\n                res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))\n        return res\n\n    def _update_model_version(self):\n        assert self._lock.locked()\n        self._version += 1\n\n    def _update_model(self):\n        assert self._lock.locked()\n        grad_var = []\n        for k in self._gradient_sum:\n            self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait\n            grad_var.append((self._gradient_sum[k], self._model[k]))\n        self._opt.apply_gradients(grad_var)\n        self._update_model_version()\n        self._gradient_sum.clear()\n        self._grad_n = 0\n\n    def _validate_model_version(self, request_model_version):\n        if request_model_version > self._version:\n            err_msg = \"Model version %d not available yet, current version: %d\" % (\n                request_model_version,\n                self._version,\n            )\n            self._logger.warning(err_msg)\n            raise ValueError(err_msg)\n\n        invalid_model_version = request_model_version < self._version\n        if invalid_model_version:\n            self._logger.warning(\n                \"Task result for outdated version %d dropped\",\n                request_model_version,\n            )\n        return invalid_model_version\n\n    def ReportGradient(self, request, _):\n        invalid_model_version = self._validate_model_version(request.model_version)\n\n        res = elasticdl_pb2.ReportGradientResponse()\n        if invalid_model_version:\n            res.accepted = False\n            res.model_version = self._version\n            return res\n\n        # TODO: Update task queue with task_id\n        with self._lock:\n            tmp = {}\n            # Do sanity check before accumulating gradients.\n            for k, v in request.gradient.items():\n                if k not in self._model:\n                    raise ValueError(\n                        \"Gradient key: %s is not part of model\", k\n                    )\n                arr = tensor_to_ndarray(v)\n                if arr.shape != self._model[k].numpy().shape:\n                    raise ValueError(\n                        \"Gradient key: %s has incompatible dimension\", k\n                    )\n                tmp[k] = arr\n\n            for k, v in tmp.items():\n                if k in self._gradient_sum:\n                    self._gradient_sum[k] = self._gradient_sum[k] + v\n                else:\n                    self._gradient_sum[k] = v\n\n            self._grad_n += 1\n            if self._grad_n >= self._grad_to_wait:\n                self._update_model()\n        res.accepted = True\n        res.model_version = self._version\n        return res\n\n    def ReportTaskResult(self, request, _):\n        if request.err_message:\n            self._logger.warning(\n                \"Worker reported error: \" + request.err_message\n            )\n            self._task_q.report(request.task_id, False)\n        else:\n            self._task_q.report(request.task_id, True)\n        return empty_pb2.Empty()\n\n    def ReportEvaluationMetrics(self, request, _):\n        invalid_model_version = self._validate_model_version(request.model_version)\n\n        res = elasticdl_pb2.ReportEvaluationMetricsResponse()\n        if invalid_model_version:\n            res.accepted = False\n            res.model_version = self._version\n            return res\n\n        with self._lock:\n            for k, v in request.evaluation_metrics.items():\n                arr = tensor_to_ndarray(v)\n                self._evaluation_metrics[k] = arr\n\n            self._update_model_version()\n        res.accepted = True\n        res.model_version = self._version\n        return res\n", "path": "elasticdl/python/elasticdl/master/servicer.py"}]}