Project

General

Profile

Statistics
| Branch: | Tag: | Revision:

gdp-v1 / tests / test_bb.py @ master

History | View | Annotate | Download (1.28 KB)

1 cbf443e3 Nitesh Mor
#!/usr/bin/env python
2
3
4
import sys
5
sys.path.insert(0, ".")
6
sys.path.insert(0, "..")
7
8
import gdp
9
import pytest
10
import time
11
import random
12
13
@pytest.fixture
14
def chtNode(setup_logging):
15 7918ce18 Nitesh Mor
    from gdp.ds import chtnode_pb2 as ds
16 cbf443e3 Nitesh Mor
    node = ds.NodeID()
17
    node.ID = str(gdp.utils.GDPAddr(random.getrandbits(256)))
18
    node.addr.TCPAddr = "127.0.0.1:9007"
19
    node.addr.UDPAddr = "127.0.0.1:9007"
20 fcb19b8c Nitesh Mor
    return gdp.routing.blackbox.CHTNode(node, None, 10000)
21 cbf443e3 Nitesh Mor
22
@pytest.fixture
23
def queryNode(setup_logging):
24 7918ce18 Nitesh Mor
    from gdp.ds import chtnode_pb2 as ds
25 cbf443e3 Nitesh Mor
    node = ds.NodeID()
26
    node.addr.TCPAddr = "query node TCPAddr"
27
    node.addr.UDPAddr = "query node UDPAddr"
28
    node.ID = str(gdp.utils.GDPAddr(random.getrandbits(256)))
29 7918ce18 Nitesh Mor
    return gdp.routing.blackbox.BBClient(node, ("127.0.0.1", 10000))
30 cbf443e3 Nitesh Mor
31
@pytest.yield_fixture
32
def ev_loop(request):
33
    gdp.utils.run_event_loop(set_daemon=True)
34
    yield
35
    gdp.utils.stop_event_loop()
36
37
def test_chtnode(ev_loop, chtNode, queryNode):
38
    time.sleep(0.5)
39
    keys = []
40
41 2ed94520 Nitesh Mor
    # Publish keys
42 cbf443e3 Nitesh Mor
    for i in xrange(10):
43
        random_key = gdp.utils.GDPAddr(random.getrandbits(256))
44
        keys.append(random_key)
45
        queryNode.publishKey(random_key)
46
47 2ed94520 Nitesh Mor
    # Query for published keys
48 cbf443e3 Nitesh Mor
    for _key in keys:
49
        node = queryNode.lookupKey(_key)
50
        assert node == queryNode.nodeID