Index: Lib/asynchat.py =================================================================== --- Lib/asynchat.py (revision 74793) +++ Lib/asynchat.py (working copy) @@ -273,47 +273,7 @@ del self.incoming[:] self.producer_fifo.clear() -class simple_producer: - def __init__ (self, data, buffer_size=512): - self.data = data - self.buffer_size = buffer_size - - def more (self): - if len (self.data) > self.buffer_size: - result = self.data[:self.buffer_size] - self.data = self.data[self.buffer_size:] - return result - else: - result = self.data - self.data = b'' - return result - -class fifo: - def __init__ (self, list=None): - if not list: - self.list = deque() - else: - self.list = deque(list) - - def __len__ (self): - return len(self.list) - - def is_empty (self): - return not self.list - - def first (self): - return self.list[0] - - def push (self, data): - self.list.append(data) - - def pop (self): - if self.list: - return (1, self.list.popleft()) - else: - return (0, None) - # Given 'haystack', see if any prefix of 'needle' is at its end. This # assumes an exact match has already been checked. Return the number of # characters matched. Index: Lib/test/test_asynchat.py =================================================================== --- Lib/test/test_asynchat.py (revision 74793) +++ Lib/test/test_asynchat.py (working copy) @@ -168,17 +168,6 @@ self.assertEqual(c.contents, []) self.assertEqual(c.buffer, data) - def test_simple_producer(self): - s, event = start_echo_server() - c = echo_client(b'\n', s.port) - data = b"hello world\nI'm not dead yet!\n" - p = asynchat.simple_producer(data+SERVER_QUIT, buffer_size=8) - c.push_with_producer(p) - asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01) - s.join() - - self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"]) - def test_string_producer(self): s, event = start_echo_server() c = echo_client(b'\n', s.port) @@ -225,34 +214,9 @@ self.assertEqual(asynchat.find_prefix_at_end("qwerty\r", "\r\n"), 1) self.assertEqual(asynchat.find_prefix_at_end("qwertydkjf", "\r\n"), 0) -class TestFifo(unittest.TestCase): - def test_basic(self): - f = asynchat.fifo() - f.push(7) - f.push(b'a') - self.assertEqual(len(f), 2) - self.assertEqual(f.first(), 7) - self.assertEqual(f.pop(), (1, 7)) - self.assertEqual(len(f), 1) - self.assertEqual(f.first(), b'a') - self.assertEqual(f.is_empty(), False) - self.assertEqual(f.pop(), (1, b'a')) - self.assertEqual(len(f), 0) - self.assertEqual(f.is_empty(), True) - self.assertEqual(f.pop(), (0, None)) - - def test_given_list(self): - f = asynchat.fifo([b'x', 17, 3]) - self.assertEqual(len(f), 3) - self.assertEqual(f.pop(), (1, b'x')) - self.assertEqual(f.pop(), (1, 17)) - self.assertEqual(f.pop(), (1, 3)) - self.assertEqual(f.pop(), (0, None)) - - def test_main(verbose=None): support.run_unittest(TestAsynchat, TestAsynchat_WithPoll, - TestHelperFunctions, TestFifo) + TestHelperFunctions) if __name__ == "__main__": test_main(verbose=True)