Skip to content

Commit d57e34f

Browse files
[brain tests] Burst threading from the main file
1 parent e809036 commit d57e34f

File tree

3 files changed

+54
-43
lines changed

3 files changed

+54
-43
lines changed

tests/brain/test_brain.py

Lines changed: 0 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -100,49 +100,6 @@ def test_extension_modules(self) -> None:
100100
extender(n)
101101

102102

103-
class ThreadingBrainTest(unittest.TestCase):
104-
def test_lock(self) -> None:
105-
lock_instance = builder.extract_node(
106-
"""
107-
import threading
108-
threading.Lock()
109-
"""
110-
)
111-
inferred = next(lock_instance.infer())
112-
self.assert_is_valid_lock(inferred)
113-
114-
acquire_method = inferred.getattr("acquire")[0]
115-
parameters = [param.name for param in acquire_method.args.args[1:]]
116-
assert parameters == ["blocking", "timeout"]
117-
118-
assert inferred.getattr("locked")
119-
120-
def test_rlock(self) -> None:
121-
self._test_lock_object("RLock")
122-
123-
def test_semaphore(self) -> None:
124-
self._test_lock_object("Semaphore")
125-
126-
def test_boundedsemaphore(self) -> None:
127-
self._test_lock_object("BoundedSemaphore")
128-
129-
def _test_lock_object(self, object_name: str) -> None:
130-
lock_instance = builder.extract_node(
131-
f"""
132-
import threading
133-
threading.{object_name}()
134-
"""
135-
)
136-
inferred = next(lock_instance.infer())
137-
self.assert_is_valid_lock(inferred)
138-
139-
def assert_is_valid_lock(self, inferred: Instance) -> None:
140-
self.assertIsInstance(inferred, astroid.Instance)
141-
self.assertEqual(inferred.root().name, "threading")
142-
for method in ("acquire", "release", "__enter__", "__exit__"):
143-
self.assertIsInstance(next(inferred.igetattr(method)), astroid.BoundMethod)
144-
145-
146103
def streams_are_fine():
147104
"""Check if streams are being overwritten,
148105
for example, by pytest

tests/brain/test_exceptions.py

Whitespace-only changes.

tests/brain/test_threading.py

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
# Licensed under the LGPL: https://www.gnu.org/licenses/old-licenses/lgpl-2.1.en.html
2+
# For details: https://github.com/PyCQA/astroid/blob/main/LICENSE
3+
# Copyright (c) https://github.com/PyCQA/astroid/blob/main/CONTRIBUTORS.txt
4+
5+
from __future__ import annotations
6+
7+
import unittest
8+
9+
import astroid
10+
from astroid import builder
11+
from astroid.bases import Instance
12+
13+
14+
class ThreadingBrainTest(unittest.TestCase):
15+
def test_lock(self) -> None:
16+
lock_instance = builder.extract_node(
17+
"""
18+
import threading
19+
threading.Lock()
20+
"""
21+
)
22+
inferred = next(lock_instance.infer())
23+
self.assert_is_valid_lock(inferred)
24+
25+
acquire_method = inferred.getattr("acquire")[0]
26+
parameters = [param.name for param in acquire_method.args.args[1:]]
27+
assert parameters == ["blocking", "timeout"]
28+
29+
assert inferred.getattr("locked")
30+
31+
def test_rlock(self) -> None:
32+
self._test_lock_object("RLock")
33+
34+
def test_semaphore(self) -> None:
35+
self._test_lock_object("Semaphore")
36+
37+
def test_boundedsemaphore(self) -> None:
38+
self._test_lock_object("BoundedSemaphore")
39+
40+
def _test_lock_object(self, object_name: str) -> None:
41+
lock_instance = builder.extract_node(
42+
f"""
43+
import threading
44+
threading.{object_name}()
45+
"""
46+
)
47+
inferred = next(lock_instance.infer())
48+
self.assert_is_valid_lock(inferred)
49+
50+
def assert_is_valid_lock(self, inferred: Instance) -> None:
51+
self.assertIsInstance(inferred, astroid.Instance)
52+
self.assertEqual(inferred.root().name, "threading")
53+
for method in ("acquire", "release", "__enter__", "__exit__"):
54+
self.assertIsInstance(next(inferred.igetattr(method)), astroid.BoundMethod)

0 commit comments

Comments
 (0)