def testPermanentFailure(self):
    def _BuggyOpMethod(client, callback):
      self._method_count += 1
      raise Exception('some permanent failure')

    op = self._CreateTestOp(user_id=1, handler=_BuggyOpMethod)
    user_op_mgr = self._CreateUserOpManager(user_id=1, handlers=[_BuggyOpMethod], callback=self.stop)
    user_op_mgr.Execute()
    self.wait()
    self.assertEqual(self._method_count, 3)

    # Ensure that operation still exists in db with right values.
    Operation.Query(self._client, 1, op.operation_id, None, lambda op: self.stop(op))
    op = self.wait()
    self.assertIsNotNone(op.first_failure)
    self.assertIsNotNone(op.last_failure)
    self.assertEqual(op.attempts, 3)
    self.assertEqual(op.quarantine, 1)
    self.assertGreater(op.backoff, 0)
Call WSGIParty's wrapped application.
dataset/ETHPy150Open viewfinderco/viewfinder/backend/db/test/op_manager_test.py/OpManagerTestCase.testPermanentFailure--crossover with--dataset/ETHPy150Open rduplain/wsgi_party/wsgi_party.py/WSGIParty.__call__