Repository: celery/celery
File: t/unit/tasks/test_result.py (Python)

Code before:
        pending_res = self.app.AsyncResult(uuid())
        with patch('celery.result.time') as _time:
            with pytest.raises(TimeoutError):
                pending_res.get(timeout=0.001, interval=0.001)
                _time.sleep.assert_called_with(0.001)

    def test_get_timeout_longer(self):
        res = self.app.AsyncResult(self.task4['id'])  # has RETRY state
        with patch('celery.result.time') as _time:
            with pytest.raises(TimeoutError):
                res.get(timeout=1, interval=1)
                _time.sleep.assert_called_with(1)

    def test_ready(self):
        oks = (self.app.AsyncResult(self.task1['id']),
               self.app.AsyncResult(self.task2['id']),
               self.app.AsyncResult(self.task3['id']))
        assert all(result.ready() for result in oks)
        assert not self.app.AsyncResult(self.task4['id']).ready()

        assert not self.app.AsyncResult(uuid()).ready()

    def test_del(self):
        with patch('celery.result.AsyncResult.backend') as backend:
            result = self.app.AsyncResult(self.task1['id'])
            result_clone = copy.copy(result)
            del result
            assert backend.remove_pending_result.called_once_with(
                result_clone
            )

        result = self.app.AsyncResult(self.task1['id'])
        result.backend = None
        del result

    def test_get_request_meta(self):

        x = self.app.AsyncResult('1')
        request = Context(
            task='foo',
            children=None,
            args=['one', 'two'],
            kwargs={'kwarg1': 'three'},
            hostname="foo",
            retries=1,
            delivery_info={'routing_key': 'celery'}
        )
        x.backend.store_result(task_id="1", result='foo', state=states.SUCCESS,
                               traceback=None, request=request)
        assert x.name == 'foo'
        assert x.args == ['one', 'two']

Review comment (bug):
TBH I'm not sure why it fails only with PyPy, but I skip it to allow us to move forward.
Not sure if it's acceptable @auvipy 

Code after:

        pending_res = self.app.AsyncResult(uuid())
        with patch('celery.result.time') as _time:
            with pytest.raises(TimeoutError):
                pending_res.get(timeout=0.001, interval=0.001)
                _time.sleep.assert_called_with(0.001)

    def test_get_timeout_longer(self):
        res = self.app.AsyncResult(self.task4['id'])  # has RETRY state
        with patch('celery.result.time') as _time:
            with pytest.raises(TimeoutError):
                res.get(timeout=1, interval=1)
                _time.sleep.assert_called_with(1)

    def test_ready(self):
        oks = (self.app.AsyncResult(self.task1['id']),
               self.app.AsyncResult(self.task2['id']),
               self.app.AsyncResult(self.task3['id']))
        assert all(result.ready() for result in oks)
        assert not self.app.AsyncResult(self.task4['id']).ready()

        assert not self.app.AsyncResult(uuid()).ready()

    @pytest.mark.skipif(
        platform.python_implementation() == "PyPy",
        reason="Mocking here doesn't play well with PyPy",
    )
    def test_del(self):
        with patch('celery.result.AsyncResult.backend') as backend:
            result = self.app.AsyncResult(self.task1['id'])
            result.backend = backend
            result_clone = copy.copy(result)
            del result
            backend.remove_pending_result.assert_called_once_with(
                result_clone
            )

        result = self.app.AsyncResult(self.task1['id'])
        result.backend = None
        del result

    def test_get_request_meta(self):

        x = self.app.AsyncResult('1')
        request = Context(
            task='foo',
            children=None,
            args=['one', 'two'],
            kwargs={'kwarg1': 'three'},
            hostname="foo",
            retries=1,