prompt: 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()
  
  ```
  
golden_diff: 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
  
verification_info: {"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"}]}