Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/astroid/brain/brain_multiprocessing.py: 38%

1# Licensed under the LGPL: https://www.gnu.org/licenses/old-licenses/lgpl-2.1.en.html 

2# For details: https://github.com/pylint-dev/astroid/blob/main/LICENSE 

3# Copyright (c) https://github.com/pylint-dev/astroid/blob/main/CONTRIBUTORS.txt 

4 

5from astroid.bases import BoundMethod 

6from astroid.brain.helpers import register_module_extender 

7from astroid.builder import parse 

8from astroid.exceptions import InferenceError 

9from astroid.manager import AstroidManager 

10from astroid.nodes.scoped_nodes import FunctionDef 

11from astroid.util import UninferableBase 

12 

13 

14def _multiprocessing_transform(): 

15 module = parse(""" 

16 from multiprocessing.managers import SyncManager 

17 def Manager(): 

18 return SyncManager() 

19 """) 

20 # Multiprocessing uses a getattr lookup inside contexts, 

21 # in order to get the attributes they need. Since it's extremely 

22 # dynamic, we use this approach to fake it. 

23 node = parse(""" 

24 from multiprocessing.context import DefaultContext, BaseContext 

25 default = DefaultContext() 

26 base = BaseContext() 

27 """) 

28 try: 

29 context = next(node["default"].infer()) 

30 base = next(node["base"].infer()) 

31 except (InferenceError, StopIteration): 

32 return module 

33 

34 if isinstance(context, UninferableBase) or isinstance(base, UninferableBase): 

35 return module 

36 

37 for node in (context, base): 

38 for key, value in node.locals.items(): 

39 if key.startswith("_"): 

40 continue 

41 

42 value = value[0] 

43 if isinstance(value, FunctionDef): 

44 # We need to rebound this, since otherwise 

45 # it will have an extra argument (self). 

46 value = BoundMethod(value, node) 

47 module[key] = value 

48 return module 

49 

50 

51def _multiprocessing_managers_transform(): 

52 return parse(""" 

53 import array 

54 import threading 

55 import multiprocessing.pool as pool 

56 import queue 

57 

58 class Namespace(object): 

59 pass 

60 

61 class Value(object): 

62 def __init__(self, typecode, value, lock=True): 

63 self._typecode = typecode 

64 self._value = value 

65 def get(self): 

66 return self._value 

67 def set(self, value): 

68 self._value = value 

69 def __repr__(self): 

70 return '%s(%r, %r)'%(type(self).__name__, self._typecode, self._value) 

71 value = property(get, set) 

72 

73 def Array(typecode, sequence, lock=True): 

74 return array.array(typecode, sequence) 

75 

76 class SyncManager(object): 

77 Queue = JoinableQueue = queue.Queue 

78 Event = threading.Event 

79 RLock = threading.RLock 

80 Lock = threading.Lock 

81 BoundedSemaphore = threading.BoundedSemaphore 

82 Condition = threading.Condition 

83 Barrier = threading.Barrier 

84 Pool = pool.Pool 

85 list = list 

86 dict = dict 

87 Value = Value 

88 Array = Array 

89 Namespace = Namespace 

90 __enter__ = lambda self: self 

91 __exit__ = lambda *args: args 

92 

93 def start(self, initializer=None, initargs=None): 

94 pass 

95 def shutdown(self): 

96 pass 

97 """) 

98 

99 

100def register(manager: AstroidManager) -> None: 

101 register_module_extender( 

102 manager, "multiprocessing.managers", _multiprocessing_managers_transform 

103 ) 

104 register_module_extender(manager, "multiprocessing", _multiprocessing_transform)