class TestSocketError:

    def _main_loop(self):
        return time.time() + tc.TASK_INTERVAL*10000, [DATAGRAM1]

    def _on_datagram_received(self):
        return
    
    def setup(self):
        self.main_loop_call_counter = 0
        self.callback_values = []
        self.datagrams_received = []
        
        self.reactor = ThreadedReactor(self._main_loop,
                                       tc.CLIENT_PORT,
                                       self._on_datagram_received,
                                       task_interval=tc.TASK_INTERVAL)
        self.reactor.s = _SocketMock()

    def test_socket_error(self):
        self.reactor.s.raise_error_on_next_sendto()
        self.reactor.run_one_step()
        self.reactor.s.raise_error_on_next_recvfrom()
        self.reactor.run_one_step()

    def teardown(self):
        return

 def setup(self):
     self.lock = threading.Lock()
     self.datagrams_received = []
     self.callback_order = []
     self.client_r = ThreadedReactor(task_interval=tc.TASK_INTERVAL)
     self.server_r = ThreadedReactor(task_interval=tc.TASK_INTERVAL)
     self.client_r.listen_udp(tc.CLIENT_ADDR[1], self.on_datagram_received)
     self.server_r.listen_udp(tc.SERVER_ADDR[1], self.on_datagram_received)
     self.client_r.start()
     self.server_r.start()

 def test_recvfrom(self):
     self.r.start()
     r2 = ThreadedReactor()
     r2.listen_udp(tc.SERVER_ADDR[1], lambda x,y:None)
     logger.critical('TESTING: IGNORE CRITICAL MESSAGE')
     r2.sendto('z', tc.CLIENT_ADDR)
     # self.r will call recvfrom (which raises socket.error)
     time.sleep(tc.TASK_INTERVAL)
     ok_(not self.callback_fired)
     self.r.stop()

    def _test_recvfrom(self):
        #self.r.start()
        r2 = ThreadedReactor(self._main_loop, tc.CLIENT_PORT,
                             self._on_datagram_received,
                             task_interval=tc.TASK_INTERVAL)
        r2.s = _SocketErrorMock()
        assert not r2.running
#        r2.start()
        assert r2.running
        logger.critical('TESTING: IGNORE CRITICAL MESSAGE')
        # self.r will call recvfrom (which raises socket.error)
        while not r2.s.error_raised:
            time.sleep(tc.TASK_INTERVAL)
        assert r2.running # the error is ignored
        ok_(not self.callback_fired)

class Controller:
    
    def __init__(self, dht_addr):
        my_addr = dht_addr
        my_id = identifier.RandomId()
        my_node = Node(my_addr, my_id)
        tracker_ = tracker.Tracker()
        token_m = token_manager.TokenManager()

        self.reactor = ThreadedReactor()
        rpc_m = RPCManager(self.reactor, my_addr[1])
        querier_ = Querier(rpc_m, my_id)
        routing_m = RoutingManager(my_node, querier_,
                                   bootstrap_nodes)
        responder_ = Responder(my_id, routing_m,
                              tracker_, token_m)

        responder_.set_on_query_received_callback(
            routing_m.on_query_received)
        querier_.set_on_response_received_callback(
            routing_m.on_response_received)
        querier_.set_on_error_received_callback(
            routing_m.on_error_received)
        querier_.set_on_timeout_callback(routing_m.on_timeout)
        querier_.set_on_nodes_found_callback(routing_m.on_nodes_found)

        routing_m.do_bootstrap()

        rpc_m.add_msg_callback(QUERY,
                               responder_.on_query_received)

        self.lookup_m = LookupManager(my_id, querier_,
                                      routing_m)
        self._routing_m = routing_m
        

    def start(self):
        self.reactor.start()

    def stop(self):
        #TODO2: stop each manager
        self.reactor.stop()

    def get_peers(self, info_hash, callback_f, bt_port=None):
        return self.lookup_m.get_peers(info_hash, callback_f, bt_port)

    def print_routing_table_stats(self):
        self._routing_m.print_stats()

    def __init__(self, dht_addr):
        self.my_addr = dht_addr
        self.my_id = identifier.RandomId()
        self.my_node = Node(self.my_addr, self.my_id)
        self.tracker = tracker.Tracker()
        self.token_m = token_manager.TokenManager()

        self.reactor = ThreadedReactor()
        self.rpc_m = RPCManager(self.reactor, self.my_addr[1])
        self.querier = Querier(self.rpc_m, self.my_id)
        self.routing_m = RoutingManager(self.my_node, self.querier,
                                        bootstrap_nodes)
        self.responder = Responder(self.my_id, self.routing_m,
                                   self.tracker, self.token_m)

        self.responder.set_on_query_received_callback(
            self.routing_m.on_query_received)
        self.querier.set_on_response_received_callback(
            self.routing_m.on_response_received)
        self.querier.set_on_error_received_callback(
            self.routing_m.on_error_received)
        self.querier.set_on_timeout_callback(self.routing_m.on_timeout)
        self.querier.set_on_nodes_found_callback(self.routing_m.on_nodes_found)

        self.routing_m.do_bootstrap()

        self.rpc_m.add_msg_callback(QUERY,
                                    self.responder.on_query_received)

        self.lookup_m = LookupManager(self.my_id, self.querier,
                                      self.routing_m)

    def __init__(self, dht_addr, state_path,
                 routing_m_mod, lookup_m_mod,
                 private_dht_name):
        #TODO: don't do this evil stuff!!!
        message.private_dht_name = private_dht_name
        
        self.state_filename = os.path.join(state_path, STATE_FILENAME)
        self.load_state()
        if not self._my_id:
            self._my_id = identifier.RandomId()
        self._my_node = Node(dht_addr, self._my_id)
        self._tracker = tracker.Tracker()
        self._token_m = token_manager.TokenManager()

        self._reactor = ThreadedReactor()
        self._reactor.listen_udp(self._my_node.addr[1],
                                 self._on_datagram_received)
        #self._rpc_m = RPCManager(self._reactor)
        self._querier = Querier(self._my_id)
        bootstrap_nodes = self.loaded_nodes or BOOTSTRAP_NODES
        del self.loaded_nodes
        self._routing_m = routing_m_mod.RoutingManager(self._my_node, 
                                                       bootstrap_nodes)
#        self._responder = Responder(self._my_id, self._routing_m,
#                                    self._tracker, self._token_m)

        self._lookup_m = lookup_m_mod.LookupManager(self._my_id)
        current_time = time.time()
        self._next_maintenance_ts = current_time
        self._next_save_state_ts = current_time + SAVE_STATE_DELAY
        
        self._running = False

 def setup(self):
     global time
     #TODO: mock time and socket
     #time = minitwisted.time = MockTime()
     #minitwisted.socket = MockSocket()
     
     self.lock = threading.Lock()
     self.datagrams_received = []
     self.callback_order = []
     self.client_r = ThreadedReactor(task_interval=tc.TASK_INTERVAL)
     self.server_r = ThreadedReactor(task_interval=tc.TASK_INTERVAL)
     self.client_s = self.client_r.listen_udp(tc.CLIENT_ADDR[1],
                                              self.on_datagram_received)
     self.server_s = self.server_r.listen_udp(tc.SERVER_ADDR[1],
                                              self.on_datagram_received)
     self.client_r.start()
     self.server_r.start()

 def test_listen_upd(self):
     r = ThreadedReactor()
     r.start()
     logger.warning(''.join(
         ('TESTING LOGS ** IGNORE EXPECTED WARNING ** ',
          '(udp_listen has not been called)')))
     self.client_r.sendto(DATA, tc.SERVER_ADDR)
     while 1: #waiting for data
         with self.lock:
             if self.datagrams_received:
                 break
         time.sleep(tc.TASK_INTERVAL)
     with self.lock:
         first_datagram = self.datagrams_received.pop(0)
         logger.debug('first_datagram: %s, %s' % (
                 first_datagram,
                 (DATA, tc.CLIENT_ADDR)))
         assert first_datagram, (DATA, tc.CLIENT_ADDR)
     r.stop()

 def setup(self):
     self.main_loop_call_counter = 0
     self.callback_values = []
     self.datagrams_received = []
     
     self.reactor = ThreadedReactor(self._main_loop,
                                    tc.CLIENT_PORT,
                                    self._on_datagram_received,
                                    task_interval=tc.TASK_INTERVAL)
     self.reactor.s = _SocketMock()

    def test_failed_join(self):
        self.lock = threading.RLock()
        self.reactor = ThreadedReactor(self._main_loop,
                                       tc.CLIENT_PORT,
                                       self._on_datagram_received,
                                       task_interval=tc.TASK_INTERVAL)
        self.reactor.s = _SocketMock(tc.TASK_INTERVAL)
#        self.reactor.start()
        self.reactor.call_asap(self._very_long_callback)
        time.sleep(tc.TASK_INTERVAL*2)
        assert_raises(Exception, self.reactor.stop)

class TestSocketError:
    def _main_loop(self):
        return time.time() + tc.TASK_INTERVAL * 10000, [DATAGRAM1]

    def _on_datagram_received(self):
        return

    def setup(self):
        self.main_loop_call_counter = 0
        self.callback_values = []
        self.datagrams_received = []

        self.lock = threading.RLock()
        self.reactor = ThreadedReactor(
            self._main_loop, tc.CLIENT_PORT, self._on_datagram_received, task_interval=tc.TASK_INTERVAL
        )
        self.reactor.s = _SocketErrorMock()
        self.reactor.start()

    def test_sendto_socket_error(self):
        time.sleep(tc.TASK_INTERVAL / 5)

    def teardown(self):
        self.reactor.stop()

    def test_mock_threaded_reactor(self):
        '''
        Just making sure that the interface is the same

        '''
        r = ThreadedReactor(task_interval=.1)
        rm = ThreadedReactorMock(task_interval=.1)

        r.listen_udp(tc.CLIENT_ADDR[1], lambda x,y:None)
        rm.listen_udp(tc.CLIENT_ADDR[1], lambda x,y:None)

        r.start()
        rm.start()

        r.sendto(DATA, tc.CLIENT_ADDR)
        rm.sendto(DATA, tc.CLIENT_ADDR)
        
        r.call_later(.1, self._callback)
        rm.call_later(.1, self._callback)
#        time.sleep(.002)
        r.stop()
        rm.stop()

class Controller():

    def __init__(self, dht_addr, state_path, routing_m_mod, lookup_m_mod, private_dht_name):
        message.private_dht_name = private_dht_name
        self.state_filename = os.path.join(state_path, STATE_FILENAME)
        self.load_state()
        if not self._my_id:
            self._my_id = identifier.RandomId()
        self._my_node = Node(dht_addr, self._my_id)
        self._tracker = tracker.Tracker()
        self._token_m = token_manager.TokenManager()
        self._reactor = ThreadedReactor()
        self._reactor.listen_udp(self._my_node.addr[1], self._on_datagram_received)
        self._querier = Querier(self._my_id)
        bootstrap_nodes = self.loaded_nodes or BOOTSTRAP_NODES
        del self.loaded_nodes
        self._routing_m = routing_m_mod.RoutingManager(self._my_node, bootstrap_nodes)
        self._lookup_m = lookup_m_mod.LookupManager(self._my_id)
        current_time = time.time()
        self._next_maintenance_ts = current_time
        self._next_save_state_ts = current_time + SAVE_STATE_DELAY
        self._running = False

    def start(self):
        self._running = True
        self._reactor.start()
        self._main_loop()

    def stop(self):
        self._reactor.stop()

    def save_state(self):
        rnodes = self._routing_m.get_main_rnodes()
        f = open(self.state_filename, 'w')
        f.write('%r\n' % self._my_id)
        for rnode in rnodes:
            f.write('%d\t%r\t%s\t%d\t%f\n' % (self._my_id.log_distance(rnode.id),
             rnode.id,
             rnode.addr[0],
             rnode.addr[1],
             rnode.rtt * 1000))

        f.close()

    def load_state(self):
        self._my_id = None
        self.loaded_nodes = []
        try:
            f = open(self.state_filename)
        except IOError:
            return

        try:
            hex_id = f.readline().strip()
            self._my_id = Id(hex_id)
            for line in f:
                _, hex_id, ip, port, _ = line.split()
                addr = (ip, int(port))
                node_ = Node(addr, Id(hex_id))
                self.loaded_nodes.append(node_)

            f.close()
        except:
            self._my_id = None
            self.loaded_nodes = []
            logger.error('state.dat is corrupted')

    def get_peers(self, lookup_id, info_hash, callback_f, bt_port = 0):
        logger.critical('get_peers %d %r' % (bt_port, info_hash))
        if time.time() > self._next_maintenance_ts + 1:
            logger.critical('minitwisted crashed or stopped!')
            return
        peers = self._tracker.get(info_hash)
        if peers:
            callback_f(lookup_id, peers)
        log_distance = info_hash.log_distance(self._my_id)
        bootstrap_rnodes = self._routing_m.get_closest_rnodes(log_distance, None, True)
        lookup_obj = self._lookup_m.get_peers(lookup_id, info_hash, callback_f, bt_port)
        lookup_queries_to_send = lookup_obj.start(bootstrap_rnodes)
        self._send_queries(lookup_queries_to_send)
        return len(lookup_queries_to_send)

    def print_routing_table_stats(self):
        self._routing_m.print_stats()

    def _main_loop(self):
        current_time = time.time()
        if current_time > self._next_maintenance_ts:
            maintenance_delay, queries_to_send, maintenance_lookup_target = self._routing_m.do_maintenance()
            self._send_queries(queries_to_send)
            if maintenance_lookup_target:
                log_distance = maintenance_lookup_target.log_distance(self._my_id)
                bootstrap_nodes = self._routing_m.get_closest_rnodes(log_distance, None, True)
                lookup_obj = self._lookup_m.maintenance_lookup(maintenance_lookup_target)
                lookup_queries_to_send = lookup_obj.start(bootstrap_nodes)
                self._send_queries(lookup_queries_to_send)
            self._next_maintenance_ts = current_time + maintenance_delay
        if current_time > self._next_save_state_ts:
            self.save_state()
            self._next_save_state_ts = current_time + SAVE_STATE_DELAY
#.........这里部分代码省略.........