mirror of
https://github.com/kernelkit/infix.git
synced 2026-08-02 13:53:01 +02:00
test: netns: Add must{,_not}_receive
In test cases, it is common that we want to verify that a certain type of traffic reaches a namespace. This has, up to this point, been solved by the generic sniffer.Sniffer. Try to improve the two most common use-cases... 1. Test if a flow reaches a namespace 2. Ensure that a flow does not reach a namespace ...by always exiting as soon as the first packet is received. This saves time on positive tests, and the synchronous API allows it to compose well with infamy.parallel().
This commit is contained in:
@@ -182,3 +182,22 @@ class IsolatedMacVlan:
|
||||
return
|
||||
|
||||
raise Exception(res)
|
||||
|
||||
def must_receive(self, expr, timeout=None, ifname=None, must=True):
|
||||
ifname = ifname if ifname else self.ifname
|
||||
timeout = timeout if timeout else self.ping_timeout
|
||||
|
||||
tshark = self.run(["tshark", "-nl", f"-i{ifname}",
|
||||
f"-aduration:{timeout}", "-c1", expr],
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True, check=True)
|
||||
|
||||
needle = "1 packet captured" if must else "0 packets captured"
|
||||
|
||||
if needle not in tshark.stdout:
|
||||
raise Exception(tshark)
|
||||
|
||||
def must_not_receive(self, *args, **kwargs):
|
||||
self.must_receive(*args, **kwargs, must=False)
|
||||
|
||||
Reference in New Issue
Block a user