Test that connection count is properly tracked.
(self)
| 597 | worker.method_queue.close() |
| 598 | |
| 599 | def test_connection_tracking(self): |
| 600 | """Test that connection count is properly tracked.""" |
| 601 | worker = self.create_worker() |
| 602 | worker.poller = mock.Mock() |
| 603 | worker.tpool = mock.Mock() |
| 604 | worker.method_queue = mock.Mock() |
| 605 | |
| 606 | assert worker.nr_conns == 0 |
| 607 | |
| 608 | # Simulate accept |
| 609 | client_sock = FakeSocket() |
| 610 | listener = mock.Mock() |
| 611 | listener.accept.return_value = (client_sock, ('127.0.0.1', 12345)) |
| 612 | listener.getsockname.return_value = ('127.0.0.1', 8000) |
| 613 | |
| 614 | worker.accept(listener) |
| 615 | assert worker.nr_conns == 1 |
| 616 | |
| 617 | # Simulate finish_request with close |
| 618 | conn = mock.Mock() |
| 619 | fs = mock.Mock() |
| 620 | fs.cancelled.return_value = False |
| 621 | fs.result.return_value = False # Not keepalive |
| 622 | worker.finish_request(conn, fs) |
| 623 | assert worker.nr_conns == 0 |
| 624 | |
| 625 | |
| 626 | class TestKeepaliveManagement: |
nothing calls this directly
no test coverage detected