Lines Matching full:cfg
12 def require_devmem(cfg):
13 if not hasattr(cfg, "_devmem_probed"):
14 probe_command = f"{cfg.bin_local} -f {cfg.ifname}"
15 cfg._devmem_supported = cmd(probe_command, fail=False, shell=True).ret == 0
16 cfg._devmem_probed = True
18 if not cfg._devmem_supported:
23 def check_rx(cfg) -> None:
24 require_devmem(cfg)
27 socat = f"socat -u - TCP{cfg.addr_ipver}:{cfg.baddr}:{port},bind={cfg.remote_baddr}:{port}"
28 listen_cmd = f"{cfg.bin_local} -l -f {cfg.ifname} -s {cfg.addr} -p {port} -c {cfg.remote_addr} -v 7"
33 head -c 1K | {socat}", host=cfg.remote, shell=True)
39 def check_tx(cfg) -> None:
40 require_devmem(cfg)
43 listen_cmd = f"socat -U - TCP{cfg.addr_ipver}-LISTEN:{port}"
45 with bkg(listen_cmd, host=cfg.remote, exit_wait=True) as socat:
46 wait_port_listen(port, host=cfg.remote)
47 cmd(f"echo -e \"hello\\nworld\"| {cfg.bin_local} -f {cfg.ifname} -s {cfg.remote_addr} -p {port}", shell=True)
53 def check_tx_chunks(cfg) -> None:
54 require_devmem(cfg)
57 listen_cmd = f"socat -U - TCP{cfg.addr_ipver}-LISTEN:{port}"
59 with bkg(listen_cmd, host=cfg.remote, exit_wait=True) as socat:
60 wait_port_listen(port, host=cfg.remote)
61 cmd(f"echo -e \"hello\\nworld\"| {cfg.bin_local} -f {cfg.ifname} -s {cfg.remote_addr} -p {port} -z 3", shell=True)
67 with NetDrvEpEnv(__file__) as cfg:
68 cfg.bin_local = path.abspath(path.dirname(__file__) + "/ncdevmem")
69 cfg.bin_remote = cfg.remote.deploy(cfg.bin_local)
72 args=(cfg, ))