balkian/soil
1
0
Fork 0
mirror of https://github.com/gsi-upm/soil synced 2025-08-23 19:52:19 +00:00
Code Issues Releases Wiki Activity

black formatting

Browse Source
This commit is contained in:
J. Fernando Sánchez
2022-10-20 14:12:10 +02:00
parent b2d48cb4df
commit c09e480d37
14 changed files with 90 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
tests/test_agents.py
View File

@@ -14,31 +14,32 @@ class Dead(agents.FSM):
class TestAgents(TestCase):
def test_die_returns_infinity(self):
'''The last step of a dead agent should return time.INFINITY'''
"""The last step of a dead agent should return time.INFINITY"""
d = Dead(unique_id=0, model=environment.Environment())
ret = d.step()
assert ret == stime.NEVER
def test_die_raises_exception(self):
'''A dead agent should raise an exception if it is stepped after death'''
"""A dead agent should raise an exception if it is stepped after death"""
d = Dead(unique_id=0, model=environment.Environment())
d.step()
with pytest.raises(stime.DeadAgent):
d.step()
def test_agent_generator(self):
'''
"""
The step function of an agent could be a generator. In that case, the state of the
agent will be resumed after every call to step.
'''
"""
a = 0
class Gen(agents.BaseAgent):
def step(self):
nonlocal a
for i in range(5):
yield
a += 1
e = environment.Environment()
g = Gen(model=e, unique_id=e.next_id())
e.schedule.add(g)
@@ -50,8 +51,9 @@ class TestAgents(TestCase):
def test_state_decorator(self):
class MyAgent(agents.FSM):
run = 0
@agents.default_state
@agents.state('original')
@agents.state("original")
def root(self):
self.run += 1
return self.other
@@ -66,19 +68,19 @@ class TestAgents(TestCase):
assert a.run == 1
a.step()
def test_broadcast(self):
'''
"""
An agent should be able to broadcast messages to every other agent, AND each receiver should be able
to process it
'''
"""
class BCast(agents.Evented):
pings_received = 0
def step(self):
print(self.model.broadcast)
try:
self.model.broadcast('PING')
self.model.broadcast("PING")
except Exception as ex:
print(ex)
while True:
@@ -87,7 +89,7 @@ class TestAgents(TestCase):
def on_receive(self, msg, sender=None):
self.pings_received += 1
e = environment.EventedEnvironment()
for i in range(10):
@@ -96,12 +98,12 @@ class TestAgents(TestCase):
pings_received = lambda: [a.pings_received for a in e.agents]
assert sorted(pings_received()) == list(range(1, 11))
e.step()
assert all(x==10 for x in pings_received())
assert all(x == 10 for x in pings_received())
def test_ask_messages(self):
'''
"""
An agent should be able to ask another agent, and wait for a response.
'''
"""
# There are two agents, they try to send pings
# This is arguably a very contrived example. In practice, the or
@@ -124,24 +126,24 @@ class TestAgents(TestCase):
def step(self):
target_id = (self.unique_id + 1) % self.count_agents()
target = self.model.agents[target_id]
print('starting')
print("starting")
while True:
if pongs or not pings: #First agent, or anyone after that
if pongs or not pings: # First agent, or anyone after that
pings.append(self.now)
response = yield target.ask('PING')
response = yield target.ask("PING")
responses.append(response)
else:
print('NOT sending ping')
print('Checking msgs')
print("NOT sending ping")
print("Checking msgs")
# Do not block if we have already received a PING
if not self.check_messages():
yield self.received()
print('done')
yield self.received()
print("done")
def on_receive(self, msg, sender=None):
if msg == 'PING':
if msg == "PING":
pongs.append(self.now)
return 'PONG'
return "PONG"
raise Exception("This should never happen")
e = environment.EventedEnvironment(schedule_class=stime.OrderedTimedActivation)
@@ -149,7 +151,6 @@ class TestAgents(TestCase):
e.add_agent(agent_class=Ping)
assert e.now == 0
for i in range(5):
e.step()
time = i + 1

2
tests/test_examples.py
View File

@@ -44,7 +44,7 @@ def add_example_tests():
for cfg, path in serialization.load_files(
join(EXAMPLES, "**", "*.yml"),
):
if 'soil_output' in path:
if "soil_output" in path:
continue
p = make_example_test(path=path, cfg=config.Config.from_raw(cfg))
fname = os.path.basename(path)

7
tests/test_main.py
View File

@@ -182,8 +182,11 @@ class TestMain(TestCase):
n_trials = 50
max_time = 2
s = simulation.Simulation(model_params={'agents': [{'agent_class': CheckRun}]},
num_trials=n_trials, max_time=max_time)
s = simulation.Simulation(
model_params={"agents": [{"agent_class": CheckRun}]},
num_trials=n_trials,
max_time=max_time,
)
runs = list(s.run_simulation(dry_run=True))
over = list(x.now for x in runs if x.now > 2)
assert len(runs) == n_trials

19
tests/test_time.py
View File

@@ -2,11 +2,12 @@ from unittest import TestCase
from soil import time, agents, environment
class TestMain(TestCase):
def test_cond(self):
'''
"""
A condition should match a When if the concition is True
'''
"""
t = time.Cond(lambda t: True)
f = time.Cond(lambda t: False)
@@ -16,17 +17,16 @@ class TestMain(TestCase):
assert w is not f
def test_cond(self):
'''
"""
Comparing a Cond to a Delta should always return False
'''
"""
c = time.Cond(lambda t: False)
d = time.Delta(1)
assert c is not d
def test_cond_env(self):
'''
'''
""" """
times_started = []
times_awakened = []
@@ -35,21 +35,18 @@ class TestMain(TestCase):
done = []
class CondAgent(agents.BaseAgent):
def step(self):
nonlocal done
times_started.append(self.now)
while True:
times_asleep.append(self.now)
yield time.Cond(lambda agent: agent.now >= 10,
delta=2)
yield time.Cond(lambda agent: agent.now >= 10, delta=2)
times_awakened.append(self.now)
if self.now >= 10:
break
done.append(self.now)
env = environment.Environment(agents=[{'agent_class': CondAgent}])
env = environment.Environment(agents=[{"agent_class": CondAgent}])
while env.schedule.time < 11:
times.append(env.now)