encrypter.py

BitTorrentABC-Linux-V.2.4.3源码

    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20)
    assert c1.pop() == ''
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed
    
    e.connection_flushed(c1)
    assert c1.pop() == ''
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, 'b' * 20)
    assert c1.pop() == ''
    assert len(c.log) == 1
    assert c.log[0][0] == 'made'
    ch = c.log[0][1]
    del c.log[:]
    assert rs.connects == []
    assert not c1.closed
    assert ch.is_flushed()
    
    e.connection_flushed(c1)
    assert c1.pop() == ''
    assert c.log == [('flushed', ch)]
    assert rs.connects == []
    assert not c1.closed
    
    c1.flushed = false
    assert not ch.is_flushed()
    
def test_wrong_header_length():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(5) * 30)
    assert c.log == []
    assert c1.closed

def test_wrong_header():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + 'a' * len(protocol_name))
    assert c.log == []
    assert c1.closed
    
def test_wrong_download_id():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'e' * 20)
    assert c.log == []
    assert c1.closed

def test_wrong_other_id():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    e.start_connection('dns', 'o' * 20)
    assert c.log == []
    assert len(rs.connects) == 1
    assert rs.connects[0][0] == 'dns'
    c1 = rs.connects[0][1]
    del rs.connects[:]
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'b' * 20)
    assert c.log == []
    assert c1.closed

def test_over_max_len():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'o' * 20)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    ch = c.log[0][1]
    del c.log[:]
    assert not c1.closed

    e.data_came_in(c1, chr(1) + chr(0) * 3)
    assert c.log == [('lost', ch)]
    assert c1.closed

def test_keepalive():
    s = []
    def sched(interval, thing, s = s):
        s.append((interval, thing))
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, sched, 30, 'd' * 20)
    assert len(s) == 1
    assert s[0][1] == 30
    kfunc = s[0][0]
    del s[:]
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert not c1.closed

    kfunc()
    assert c1.pop() == ''
    assert c.log == []
    assert not c1.closed
    assert s == [(kfunc, 30)]
    del s[:]

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'o' * 20)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    del c.log[:]
    assert c1.pop() == ''
    assert not c1.closed

    kfunc()
    assert c1.pop() == chr(0) * 4
    assert c.log == []
    assert not c1.closed

def test_swallow_keepalive():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'o' * 20)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    del c.log[:]
    assert not c1.closed

    e.data_came_in(c1, chr(0) * 4)
    assert c.log == []
    assert not c1.closed

def test_local_close():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'o' * 20)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    ch = c.log[0][1]
    del c.log[:]
    assert not c1.closed

    ch.close()
    assert c.log == [('lost', ch)]
    del c.log[:]
    assert c1.closed

def test_local_close_in_message_receive():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'o' * 20)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    ch = c.log[0][1]
    del c.log[:]
    assert not c1.closed

    c.close_next = true
    e.data_came_in(c1, chr(0) * 3 + chr(4) + 'abcd')
    assert c.log == [('got', ch, 'abcd'), ('lost', ch)]
    assert c1.closed

def test_remote_close():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'o' * 20)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    ch = c.log[0][1]
    del c.log[:]
    assert not c1.closed

    e.connection_lost(c1)
    assert c.log == [('lost', ch)]
    assert not c1.closed

def test_partial_data_in():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 4)
    e.data_came_in(c1, chr(0) * 4 + 'd' * 20 + 'c' * 10)
    e.data_came_in(c1, 'c' * 10)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    del c.log[:]
    assert not c1.closed
    
def test_ignore_connect_of_extant():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()
    e.external_connection_made(c1)
    assert c1.pop() == chr(len(protocol_name)) + protocol_name + \
        chr(0) * 8 + 'd' * 20 + 'a' * 20
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

    e.data_came_in(c1, chr(len(protocol_name)) + protocol_name + 
        chr(0) * 8 + 'd' * 20 + 'o' * 20)
    assert len(c.log) == 1 and c.log[0][0] == 'made'
    del c.log[:]
    assert not c1.closed

    e.start_connection('dns', 'o' * 20)
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

def test_ignore_connect_to_self():
    c = DummyConnecter()
    rs = DummyRawServer()
    e = Encoder(c, rs, 'a' * 20, 500, dummyschedule, 30, 'd' * 20)
    c1 = DummyRawConnection()

    e.start_connection('dns', 'a' * 20)
    assert c.log == []
    assert rs.connects == []
    assert not c1.closed

def test_conversion():
    assert toint(tobinary(50000)) == 50000