Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_match_iand():
t1 = Template()
t1.sender = "sender1@host"
t2 = Template()
t2.to = "recv1@host"
t2.metadata = {"performative": "query"}
m1 = Message()
m1.sender = "sender1@host"
t1 &= t2
assert not t1.match(m1)
m2 = Message()
m2.to = "recv1@host"
m2.metadata = {"performative": "query"}
assert not t1.match(m2)
m3 = Message()
m3.sender = "sender1@host"
m3.to = "recv1@host"
m3.metadata = {"performative": "query"}
assert t1.match(m3)
def test_match_or():
t1 = Template()
t1.sender = "sender1@host"
t2 = Template()
t2.to = "recv1@host"
t2.metadata = {"performative": "query"}
m1 = Message()
m1.sender = "sender1@host"
assert (t1 | t2).match(m1)
m2 = Message()
m2.to = "recv1@host"
m2.metadata = {"performative": "query"}
assert (t1 | t2).match(m2)
m3 = Message()
m3.sender = "sender2@host"
m3.to = "recv1@host"
m3.metadata = {"performative": "inform"}
assert not (t1 | t2).match(m3)
def test_match_and():
t1 = Template()
t1.sender = "sender1@host"
t2 = Template()
t2.to = "recv1@host"
t2.metadata = {"performative": "query"}
m1 = Message()
m1.sender = "sender1@host"
assert not (t1 & t2).match(m1)
m2 = Message()
m2.to = "recv1@host"
m2.metadata = {"performative": "query"}
assert not (t1 & t2).match(m2)
m3 = Message()
m3.sender = "sender1@host"
m3.to = "recv1@host"
m3.metadata = {"performative": "query"}
assert (t1 & t2).match(m3)
def test_filter_to():
cycle = itertools.cycle(range(3))
trace = TraceStore(10)
for i in range(5):
msg = Message(
body=str(i), to="{}@server".format(next(cycle)), sender="sender@server"
)
trace.append(msg)
result = [x[1].body for x in trace.filter(to="0@server")]
assert result == ["0", "3"]
def test_match_false_metadata_with_different_key():
template = Template()
template.metadata = {"performative": "query"}
message = Message()
message.set_metadata("language", "query")
assert not template.match(message)
m2.metadata = {"performative": "query"}
assert (t1 ^ t2).match(m2)
m3 = Message()
m3.sender = "sender2@host"
m3.to = "recv1@host"
m3.metadata = {"performative": "inform"}
assert not (t1 ^ t2).match(m3)
t1 = Template()
t1.sender = "sender1@host"
t2 = Template()
t2.sender = "sender1@host"
m4 = Message()
m4.sender = "sender1@host"
assert not (t1 ^ t2).match(m4)
def test_match_false_body():
template = Template()
template.body = "Hello World"
message = Message()
assert not template.match(message)
message.body = "Bye Bye Love"
assert not template.match(message)
def test_match():
template = Template()
template.sender = "sender1@host"
template.to = "recv1@host"
template.body = "Hello World"
template.thread = "thread-id"
template.metadata = {"performative": "query"}
message = Message()
message.sender = "sender1@host"
message.to = "recv1@host"
message.body = "Hello World"
message.thread = "thread-id"
message.set_metadata("performative", "query")
assert template.match(message)
t2.metadata = {"performative": "query"}
m1 = Message()
m1.sender = "sender1@host"
t1 |= t2
assert t1.match(m1)
m2 = Message()
m2.to = "recv1@host"
m2.metadata = {"performative": "query"}
assert t1.match(m2)
m3 = Message()
m3.sender = "sender2@host"
m3.to = "recv1@host"
m3.metadata = {"performative": "inform"}
assert not t1.match(m3)
async def run(self):
print("InformBehav running")
msg = Message(to=self.agent.recv_jid) # Instantiate the message
msg.set_metadata(
"performative", "inform"
) # Set the "inform" FIPA performative
msg.body = "Hello World {}".format(
self.agent.recv_jid
) # Set the message content
await self.send(msg)
print("Message sent!")
# stop agent from behaviour
await self.agent.stop()