(srfunc, # type: Callable[..., Tuple[SndRcvList, PacketList]]
pkts, # type: _PacketIterable
prn=lambda x: x[1].summary(), # type: Optional[Callable[[QueryAnswer], Any]] # noqa: E501
prnfail=lambda x: x.summary(), # type: Optional[Callable[[Packet], Any]]
inter=1, # type: int
timeout=None, # type: Optional[int]
count=None, # type: Optional[int]
verbose=None, # type: Optional[int]
store=1, # type: int
*args, # type: Any
**kargs # type: Any
)
| 791 | |
| 792 | |
| 793 | def __sr_loop(srfunc, # type: Callable[..., Tuple[SndRcvList, PacketList]] |
| 794 | pkts, # type: _PacketIterable |
| 795 | prn=lambda x: x[1].summary(), # type: Optional[Callable[[QueryAnswer], Any]] # noqa: E501 |
| 796 | prnfail=lambda x: x.summary(), # type: Optional[Callable[[Packet], Any]] |
| 797 | inter=1, # type: int |
| 798 | timeout=None, # type: Optional[int] |
| 799 | count=None, # type: Optional[int] |
| 800 | verbose=None, # type: Optional[int] |
| 801 | store=1, # type: int |
| 802 | *args, # type: Any |
| 803 | **kargs # type: Any |
| 804 | ): |
| 805 | # type: (...) -> Tuple[SndRcvList, PacketList] |
| 806 | n = 0 |
| 807 | r = 0 |
| 808 | ct = conf.color_theme |
| 809 | if verbose is None: |
| 810 | verbose = conf.verb |
| 811 | parity = 0 |
| 812 | ans = [] # type: List[QueryAnswer] |
| 813 | unans = [] # type: List[Packet] |
| 814 | if timeout is None: |
| 815 | timeout = min(2 * inter, 5) |
| 816 | try: |
| 817 | while True: |
| 818 | parity ^= 1 |
| 819 | col = [ct.even, ct.odd][parity] |
| 820 | if count is not None: |
| 821 | if count == 0: |
| 822 | break |
| 823 | count -= 1 |
| 824 | start = time.monotonic() |
| 825 | if verbose > 1: |
| 826 | print("\rsend...\r", end=' ') |
| 827 | res = srfunc(pkts, timeout=timeout, verbose=0, chainCC=True, *args, **kargs) # noqa: E501 |
| 828 | n += len(res[0]) + len(res[1]) |
| 829 | r += len(res[0]) |
| 830 | if verbose > 1 and prn and len(res[0]) > 0: |
| 831 | msg = "RECV %i:" % len(res[0]) |
| 832 | print("\r" + ct.success(msg), end=' ') |
| 833 | for rcv in res[0]: |
| 834 | print(col(prn(rcv))) |
| 835 | print(" " * len(msg), end=' ') |
| 836 | if verbose > 1 and prnfail and len(res[1]) > 0: |
| 837 | msg = "fail %i:" % len(res[1]) |
| 838 | print("\r" + ct.fail(msg), end=' ') |
| 839 | for fail in res[1]: |
| 840 | print(col(prnfail(fail))) |
| 841 | print(" " * len(msg), end=' ') |
| 842 | if verbose > 1 and not (prn or prnfail): |
| 843 | print("recv:%i fail:%i" % tuple( |
| 844 | map(len, res[:2]) # type: ignore |
| 845 | )) |
| 846 | if verbose == 1: |
| 847 | if res[0]: |
| 848 | os.write(1, b"*") |
| 849 | if res[1]: |
| 850 | os.write(1, b".") |
no test coverage detected