def _Test(self, use_monitor):
    def cfg_fn(port, curl):
      curl.opts["__port__"] = port

    def _LockCheckReset(monitor, req):
      self.assertTrue(monitor._lock.is_owned(shared=0),
                      msg="Lock must be owned in exclusive mode")
      assert not hasattr(req, "lockcheck__")
      setattr(req, "lockcheck__", True)

    def _BuildNiceName(port, default=None):
      if port % 5 == 0:
        return "nicename%s" % port
      else:
        # Use standard name
        return default

    requests = \
      [http.client.HttpClientRequest("localhost", i, "POST", "/version%s" % i,
                                     curl_config_fn=compat.partial(cfg_fn, i),
                                     completion_cb=NotImplementedError,
                                     nicename=_BuildNiceName(i))
       for i in range(15176, 15501)]
    requests_count = len(requests)

    if use_monitor:
      lock_monitor_cb = self._MonitorChecker()
    else:
      lock_monitor_cb = None

    def _ProcessRequests(multi, handles):
      self.assertTrue(isinstance(multi, self._DummyCurlMulti))
      self.assertEqual(len(requests), len(handles))
      self.assertTrue(compat.all(isinstance(curl, _FakeCurl)
                                 for curl in handles))

      # Prepare for lock check
      for req in requests:
        assert req.completion_cb is NotImplementedError
        if use_monitor:
          req.completion_cb = \
            compat.partial(_LockCheckReset, lock_monitor_cb.GetMonitor())

      for idx, curl in enumerate(handles):
        try:
          port = curl.opts["__port__"]
        except KeyError:
          self.fail("Per-request config function was not called")

        if use_monitor:
          # Check if lock information is correct
          lock_info = lock_monitor_cb.GetMonitor().GetLockInfo(None)
          expected = \
            [("rpc/%s" % (_BuildNiceName(handle.opts["__port__"],
                                         default=("localhost/version%s" %
                                                  handle.opts["__port__"]))),
              None,
              [threading.currentThread().getName()], None)
             for handle in handles[idx:]]
          self.assertEqual(sorted(lock_info), sorted(expected))

        if port % 3 == 0:
          response_code = http.HTTP_OK
          msg = None
        else:
          response_code = http.HttpNotFound.code
          msg = "test error"

        curl.info = {
          pycurl.RESPONSE_CODE: response_code,
        }
        if hasattr(pycurl, 'LOCAL_IP'):
          curl.info[pycurl.LOCAL_IP] = '127.0.0.1'
        if hasattr(pycurl, 'LOCAL_PORT'):
          curl.info[pycurl.LOCAL_PORT] = port

        # Prepare for reset
        self.assertFalse(curl.opts.pop(pycurl.POSTFIELDS))
        self.assertTrue(callable(curl.opts.pop(pycurl.WRITEFUNCTION)))

        yield (curl, msg)

      if use_monitor:
        self.assertTrue(compat.all(req.lockcheck__ for req in requests))

    if use_monitor:
      self.assertEqual(lock_monitor_cb.GetMonitor(), None)

    http.client.ProcessRequests(requests, lock_monitor_cb=lock_monitor_cb,
                                _curl=_FakeCurl,
                                _curl_multi=self._DummyCurlMulti,
                                _curl_process=_ProcessRequests)
    for req in requests:
      if req.port % 3 == 0:
        self.assertTrue(req.success)
        self.assertEqual(req.error, None)
      else:
        self.assertFalse(req.success)
        self.assertTrue("test error" in req.error)

    # See if monitor was disabled
    if use_monitor:
      monitor = lock_monitor_cb.GetMonitor()
      self.assertEqual(monitor._pending_fn, None)
      self.assertEqual(monitor.GetLockInfo(None), [])
    else:
      self.assertEqual(lock_monitor_cb, None)

    self.assertEqual(len(requests), requests_count)
dataset/ETHPy150Open ganeti/ganeti/test/py/ganeti.http_unittest.py/TestProcessRequests._Test/original