mirror of https://github.com/oxen-io/lokinet
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
45 lines
963 B
Python
45 lines
963 B
Python
import pyllarp
|
|
from time import time
|
|
|
|
def test_peer_stats(HiveForPeerStats):
|
|
hive = HiveForPeerStats(n_relays=5, n_clients=0, netid="hive")
|
|
|
|
start_time = time()
|
|
cur_time = start_time
|
|
test_duration = 30 #seconds
|
|
|
|
paths = []
|
|
|
|
print("looking for events...")
|
|
|
|
numInbound = 0
|
|
numOutbound = 0
|
|
numAttempts = 0
|
|
|
|
while cur_time < start_time + test_duration:
|
|
|
|
hive.CollectAllEvents()
|
|
|
|
for event in hive.events:
|
|
event_name = event.__class__.__name__
|
|
|
|
if event_name == "LinkSessionEstablishedEvent":
|
|
if event.inbound:
|
|
numInbound += 1
|
|
else:
|
|
numOutbound += 1
|
|
|
|
if event_name == "ConnectionAttemptEvent":
|
|
numAttempts += 1
|
|
|
|
hive.events = []
|
|
cur_time = time()
|
|
|
|
print("test duration exceeded")
|
|
print("in: {} out: {} attempts: {}", numInbound, numOutbound, numAttempts);
|
|
assert(numInbound == numOutbound)
|
|
assert(numOutbound == numAttempts)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|