| 1036 | self._do_testqueuepool(useclose=True) |
| 1037 | |
| 1038 | def _do_testqueuepool(self, useclose=False): |
| 1039 | p = self._queuepool_fixture(pool_size=3, max_overflow=-1) |
| 1040 | |
| 1041 | def status(pool): |
| 1042 | return ( |
| 1043 | pool.size(), |
| 1044 | pool.checkedin(), |
| 1045 | pool.overflow(), |
| 1046 | pool.checkedout(), |
| 1047 | ) |
| 1048 | |
| 1049 | c1 = p.connect() |
| 1050 | self.assert_(status(p) == (3, 0, -2, 1)) |
| 1051 | c2 = p.connect() |
| 1052 | self.assert_(status(p) == (3, 0, -1, 2)) |
| 1053 | c3 = p.connect() |
| 1054 | self.assert_(status(p) == (3, 0, 0, 3)) |
| 1055 | c4 = p.connect() |
| 1056 | self.assert_(status(p) == (3, 0, 1, 4)) |
| 1057 | c5 = p.connect() |
| 1058 | self.assert_(status(p) == (3, 0, 2, 5)) |
| 1059 | c6 = p.connect() |
| 1060 | self.assert_(status(p) == (3, 0, 3, 6)) |
| 1061 | if useclose: |
| 1062 | c4.close() |
| 1063 | c3.close() |
| 1064 | c2.close() |
| 1065 | else: |
| 1066 | c4 = c3 = c2 = None |
| 1067 | lazy_gc() |
| 1068 | eq_(status(p), (3, 3, 3, 3)) |
| 1069 | if useclose: |
| 1070 | c1.close() |
| 1071 | c5.close() |
| 1072 | c6.close() |
| 1073 | else: |
| 1074 | c1 = c5 = c6 = None |
| 1075 | lazy_gc() |
| 1076 | self.assert_(status(p) == (3, 3, 0, 0)) |
| 1077 | c1 = p.connect() |
| 1078 | c2 = p.connect() |
| 1079 | self.assert_(status(p) == (3, 1, 0, 2), status(p)) |
| 1080 | if useclose: |
| 1081 | c2.close() |
| 1082 | else: |
| 1083 | c2 = None |
| 1084 | lazy_gc() |
| 1085 | self.assert_(status(p) == (3, 2, 0, 1)) |
| 1086 | c1.close() |
| 1087 | |
| 1088 | def test_timeout_accessor(self): |
| 1089 | expected_timeout = 123 |