File: test_pickle.py

package info (click to toggle)
celery 5.5.3-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 8,008 kB
  • sloc: python: 64,346; sh: 795; makefile: 378
file content (47 lines) | stat: -rw-r--r-- 1,386 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
from celery.utils.serialization import pickle


class RegularException(Exception):
    pass


class ArgOverrideException(Exception):

    def __init__(self, message, status_code=10):
        self.status_code = status_code
        super().__init__(message, status_code)


class test_Pickle:

    def test_pickle_regular_exception(self):
        exc = None
        try:
            raise RegularException('RegularException raised')
        except RegularException as exc_:
            exc = exc_

        pickled = pickle.dumps({'exception': exc})
        unpickled = pickle.loads(pickled)
        exception = unpickled.get('exception')
        assert exception
        assert isinstance(exception, RegularException)
        assert exception.args == ('RegularException raised',)

    def test_pickle_arg_override_exception(self):

        exc = None
        try:
            raise ArgOverrideException(
                'ArgOverrideException raised', status_code=100,
            )
        except ArgOverrideException as exc_:
            exc = exc_

        pickled = pickle.dumps({'exception': exc})
        unpickled = pickle.loads(pickled)
        exception = unpickled.get('exception')
        assert exception
        assert isinstance(exception, ArgOverrideException)
        assert exception.args == ('ArgOverrideException raised', 100)
        assert exception.status_code == 100