(self, num, sequential_ips=True)
| 92 | self.rate_limit_tests() |
| 93 | |
| 94 | def setup_addr_msg(self, num, sequential_ips=True): |
| 95 | addrs = [] |
| 96 | for i in range(num): |
| 97 | addr = CAddress() |
| 98 | addr.time = self.mocktime + random.randrange(-100, 100) |
| 99 | addr.nServices = P2P_SERVICES |
| 100 | if sequential_ips: |
| 101 | assert self.counter < 256 ** 2 # Don't allow the returned ip addresses to wrap. |
| 102 | addr.ip = f"123.123.{self.counter // 256}.{self.counter % 256}" |
| 103 | self.counter += 1 |
| 104 | else: |
| 105 | addr.ip = f"{random.randrange(128,169)}.{random.randrange(1,255)}.{random.randrange(1,255)}.{random.randrange(1,255)}" |
| 106 | addr.port = 8333 + i |
| 107 | addrs.append(addr) |
| 108 | |
| 109 | msg = msg_addr() |
| 110 | msg.addrs = addrs |
| 111 | return msg |
| 112 | |
| 113 | def send_addr_msg(self, source, msg, receivers): |
| 114 | source.send_and_ping(msg) |
no test coverage detected