| 2295 | # |
| 2296 | |
| 2297 | class _TestValue(BaseTestCase): |
| 2298 | |
| 2299 | ALLOWED_TYPES = ('processes',) |
| 2300 | |
| 2301 | codes_values = [ |
| 2302 | ('i', 4343, 24234), |
| 2303 | ('d', 3.625, -4.25), |
| 2304 | ('h', -232, 234), |
| 2305 | ('q', 2 ** 33, 2 ** 34), |
| 2306 | ('c', latin('x'), latin('y')) |
| 2307 | ] |
| 2308 | |
| 2309 | def setUp(self): |
| 2310 | if not HAS_SHAREDCTYPES: |
| 2311 | self.skipTest("requires multiprocessing.sharedctypes") |
| 2312 | |
| 2313 | @classmethod |
| 2314 | def _test(cls, values): |
| 2315 | for sv, cv in zip(values, cls.codes_values): |
| 2316 | sv.value = cv[2] |
| 2317 | |
| 2318 | |
| 2319 | def test_value(self, raw=False): |
| 2320 | if raw: |
| 2321 | values = [self.RawValue(code, value) |
| 2322 | for code, value, _ in self.codes_values] |
| 2323 | else: |
| 2324 | values = [self.Value(code, value) |
| 2325 | for code, value, _ in self.codes_values] |
| 2326 | |
| 2327 | for sv, cv in zip(values, self.codes_values): |
| 2328 | self.assertEqual(sv.value, cv[1]) |
| 2329 | |
| 2330 | proc = self.Process(target=self._test, args=(values,)) |
| 2331 | proc.daemon = True |
| 2332 | proc.start() |
| 2333 | proc.join() |
| 2334 | |
| 2335 | for sv, cv in zip(values, self.codes_values): |
| 2336 | self.assertEqual(sv.value, cv[2]) |
| 2337 | |
| 2338 | def test_rawvalue(self): |
| 2339 | self.test_value(raw=True) |
| 2340 | |
| 2341 | def test_getobj_getlock(self): |
| 2342 | val1 = self.Value('i', 5) |
| 2343 | lock1 = val1.get_lock() |
| 2344 | obj1 = val1.get_obj() |
| 2345 | |
| 2346 | val2 = self.Value('i', 5, lock=None) |
| 2347 | lock2 = val2.get_lock() |
| 2348 | obj2 = val2.get_obj() |
| 2349 | |
| 2350 | lock = self.Lock() |
| 2351 | val3 = self.Value('i', 5, lock=lock) |
| 2352 | lock3 = val3.get_lock() |
| 2353 | obj3 = val3.get_obj() |
| 2354 | self.assertEqual(lock, lock3) |