wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-18 12:40:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Convert asserts in unittest to pytest (#2529)

Browse Source
This commit is contained in:
Philipp Moritz
2018-08-01 22:32:10 -07:00
committed by Robert Nishihara
parent 9ea57c2a93
commit d8ba667175
10 changed files with 531 additions and 583 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/actor_test.py
+155 -179
View File
@@ -34,46 +34,40 @@ class ActorAPI(unittest.TestCase):
return self.arg0 + arg0, self.arg1 + arg1, self.arg2 + arg2
actor = Actor.remote(0)
self.assertEqual(ray.get(actor.get_values.remote(1)), (1, 3, "ab"))
assert ray.get(actor.get_values.remote(1)) == (1, 3, "ab")
actor = Actor.remote(1, 2)
self.assertEqual(ray.get(actor.get_values.remote(2, 3)), (3, 5, "ab"))
assert ray.get(actor.get_values.remote(2, 3)) == (3, 5, "ab")
actor = Actor.remote(1, 2, "c")
self.assertEqual(
ray.get(actor.get_values.remote(2, 3, "d")), (3, 5, "cd"))
assert ray.get(actor.get_values.remote(2, 3, "d")) == (3, 5, "cd")
actor = Actor.remote(1, arg2="c")
self.assertEqual(
ray.get(actor.get_values.remote(0, arg2="d")), (1, 3, "cd"))
self.assertEqual(
ray.get(actor.get_values.remote(0, arg2="d", arg1=0)),
(1, 1, "cd"))
assert ray.get(actor.get_values.remote(0, arg2="d")) == (1, 3, "cd")
assert ray.get(actor.get_values.remote(0, arg2="d", arg1=0)) == (1, 1,
"cd")
actor = Actor.remote(1, arg2="c", arg1=2)
self.assertEqual(
ray.get(actor.get_values.remote(0, arg2="d")), (1, 4, "cd"))
self.assertEqual(
ray.get(actor.get_values.remote(0, arg2="d", arg1=0)),
(1, 2, "cd"))
self.assertEqual(
ray.get(actor.get_values.remote(arg2="d", arg1=0, arg0=2)),
(3, 2, "cd"))
assert ray.get(actor.get_values.remote(0, arg2="d")) == (1, 4, "cd")
assert ray.get(actor.get_values.remote(0, arg2="d", arg1=0)) == (1, 2,
"cd")
assert ray.get(actor.get_values.remote(arg2="d", arg1=0,
arg0=2)) == (3, 2, "cd")
# Make sure we get an exception if the constructor is called
# incorrectly.
with self.assertRaises(Exception):
with pytest.raises(Exception):
actor = Actor.remote()
with self.assertRaises(Exception):
with pytest.raises(Exception):
actor = Actor.remote(0, 1, 2, arg3=3)
with self.assertRaises(Exception):
with pytest.raises(Exception):
actor = Actor.remote(0, arg0=1)
# Make sure we get an exception if the method is called incorrectly.
actor = Actor.remote(1)
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.get(actor.get_values.remote())
def testVariableNumberOfArgs(self):
@@ -90,21 +84,18 @@ class ActorAPI(unittest.TestCase):
return self.arg0 + arg0, self.arg1 + arg1, self.args, args
actor = Actor.remote(0)
self.assertEqual(ray.get(actor.get_values.remote(1)), (1, 3, (), ()))
assert ray.get(actor.get_values.remote(1)) == (1, 3, (), ())
actor = Actor.remote(1, 2)
self.assertEqual(
ray.get(actor.get_values.remote(2, 3)), (3, 5, (), ()))
assert ray.get(actor.get_values.remote(2, 3)) == (3, 5, (), ())
actor = Actor.remote(1, 2, "c")
self.assertEqual(
ray.get(actor.get_values.remote(2, 3, "d")), (3, 5, ("c", ),
("d", )))
assert ray.get(actor.get_values.remote(2, 3, "d")) == (3, 5, ("c", ),
("d", ))
actor = Actor.remote(1, 2, "a", "b", "c", "d")
self.assertEqual(
ray.get(actor.get_values.remote(2, 3, 1, 2, 3, 4)),
(3, 5, ("a", "b", "c", "d"), (1, 2, 3, 4)))
assert ray.get(actor.get_values.remote(
2, 3, 1, 2, 3, 4)) == (3, 5, ("a", "b", "c", "d"), (1, 2, 3, 4))
@ray.remote
class Actor(object):
@@ -115,11 +106,11 @@ class ActorAPI(unittest.TestCase):
return self.args, args
a = Actor.remote()
self.assertEqual(ray.get(a.get_values.remote()), ((), ()))
assert ray.get(a.get_values.remote()) == ((), ())
a = Actor.remote(1)
self.assertEqual(ray.get(a.get_values.remote(2)), ((1, ), (2, )))
assert ray.get(a.get_values.remote(2)) == ((1, ), (2, ))
a = Actor.remote(1, 2)
self.assertEqual(ray.get(a.get_values.remote(3, 4)), ((1, 2), (3, 4)))
assert ray.get(a.get_values.remote(3, 4)) == ((1, 2), (3, 4))
def testNoArgs(self):
ray.init(num_workers=0)
@@ -133,7 +124,7 @@ class ActorAPI(unittest.TestCase):
pass
actor = Actor.remote()
self.assertEqual(ray.get(actor.get_values.remote()), None)
assert ray.get(actor.get_values.remote()) is None
def testNoConstructor(self):
# If no __init__ method is provided, that should not be a problem.
@@ -145,7 +136,7 @@ class ActorAPI(unittest.TestCase):
pass
actor = Actor.remote()
self.assertEqual(ray.get(actor.get_values.remote()), None)
assert ray.get(actor.get_values.remote()) is None
def testCustomClasses(self):
ray.init(num_workers=0)
@@ -168,12 +159,12 @@ class ActorAPI(unittest.TestCase):
actor = Actor.remote(Foo(2))
results1 = ray.get(actor.get_values1.remote())
self.assertEqual(results1[0].x, 1)
self.assertEqual(results1[1].x, 2)
assert results1[0].x == 1
assert results1[1].x == 2
results2 = ray.get(actor.get_values2.remote(Foo(3)))
self.assertEqual(results2[0].x, 1)
self.assertEqual(results2[1].x, 2)
self.assertEqual(results2[2].x, 3)
assert results2[0].x == 1
assert results2[1].x == 2
assert results2[2].x == 3
def testCachingActors(self):
# Test defining actors before ray.init() has been called.
@@ -188,20 +179,20 @@ class ActorAPI(unittest.TestCase):
# Check that we can't actually create actors before ray.init() has been
# called.
with self.assertRaises(Exception):
with pytest.raises(Exception):
f = Foo.remote()
ray.init(num_workers=0)
f = Foo.remote()
self.assertEqual(ray.get(f.get_val.remote()), 3)
assert ray.get(f.get_val.remote()) == 3
def testDecoratorArgs(self):
ray.init(num_workers=0, driver_mode=ray.SILENT_MODE)
# This is an invalid way of using the actor decorator.
with self.assertRaises(Exception):
with pytest.raises(Exception):
@ray.remote()
class Actor(object):
@@ -209,7 +200,7 @@ class ActorAPI(unittest.TestCase):
pass
# This is an invalid way of using the actor decorator.
with self.assertRaises(Exception):
with pytest.raises(Exception):
@ray.remote(invalid_kwarg=0) # noqa: F811
class Actor(object):
@@ -217,7 +208,7 @@ class ActorAPI(unittest.TestCase):
pass
# This is an invalid way of using the actor decorator.
with self.assertRaises(Exception):
with pytest.raises(Exception):
@ray.remote(num_cpus=0, invalid_kwarg=0) # noqa: F811
class Actor(object):
@@ -259,7 +250,7 @@ class ActorAPI(unittest.TestCase):
random.seed(1234)
f2 = Foo.remote()
self.assertNotEqual(f1._ray_actor_id.id(), f2._ray_actor_id.id())
assert f1._ray_actor_id.id() != f2._ray_actor_id.id()
def testActorClassName(self):
ray.init(num_workers=0)
@@ -273,10 +264,10 @@ class ActorAPI(unittest.TestCase):
r = ray.worker.global_worker.redis_client
actor_keys = r.keys("ActorClass*")
self.assertEqual(len(actor_keys), 1)
assert len(actor_keys) == 1
actor_class_info = r.hgetall(actor_keys[0])
self.assertEqual(actor_class_info[b"class_name"], b"Foo")
self.assertEqual(actor_class_info[b"module"], b"actor_test")
assert actor_class_info[b"class_name"] == b"Foo"
assert actor_class_info[b"module"] == b"actor_test"
def testMultipleReturnValues(self):
ray.init(num_workers=0)
@@ -301,16 +292,16 @@ class ActorAPI(unittest.TestCase):
f = Foo.remote()
id0 = f.method0.remote()
self.assertEqual(ray.get(id0), 1)
assert ray.get(id0) == 1
id1 = f.method1.remote()
self.assertEqual(ray.get(id1), 1)
assert ray.get(id1) == 1
id2a, id2b = f.method2.remote()
self.assertEqual(ray.get([id2a, id2b]), [1, 2])
assert ray.get([id2a, id2b]) == [1, 2]
id3a, id3b, id3c = f.method3.remote()
self.assertEqual(ray.get([id3a, id3b, id3c]), [1, 2, 3])
assert ray.get([id3a, id3b, id3c]) == [1, 2, 3]
class ActorMethods(unittest.TestCase):
@@ -329,10 +320,10 @@ class ActorMethods(unittest.TestCase):
return self.x + y
t = Test.remote(2)
self.assertEqual(ray.get(t.f.remote(1)), 3)
assert ray.get(t.f.remote(1)) == 3
# Make sure that calling an actor method directly raises an exception.
with self.assertRaises(Exception):
with pytest.raises(Exception):
t.f(1)
def testActorDeletion(self):
@@ -365,7 +356,7 @@ class ActorMethods(unittest.TestCase):
# Make sure that if we create an actor and call a method on it
# immediately, the actor doesn't get killed before the method is
# called.
self.assertEqual(ray.get(Actor.remote().method.remote()), 1)
assert ray.get(Actor.remote().method.remote()) == 1
def testActorDeletionWithGPUs(self):
ray.init(num_workers=0, num_gpus=1)
@@ -400,12 +391,12 @@ class ActorMethods(unittest.TestCase):
c1 = Counter.remote()
c1.increase.remote()
self.assertEqual(ray.get(c1.value.remote()), 1)
assert ray.get(c1.value.remote()) == 1
c2 = Counter.remote()
c2.increase.remote()
c2.increase.remote()
self.assertEqual(ray.get(c2.value.remote()), 2)
assert ray.get(c2.value.remote()) == 2
def testActorClassMethods(self):
ray.init()
@@ -429,9 +420,9 @@ class ActorMethods(unittest.TestCase):
return value
a = Foo.as_remote().remote()
self.assertEqual(ray.get(a.echo.remote(2)), 2)
self.assertEqual(ray.get(a.f.remote()), 2)
self.assertEqual(ray.get(a.g.remote(2)), 4)
assert ray.get(a.echo.remote(2)) == 2
assert ray.get(a.f.remote()) == 2
assert ray.get(a.g.remote(2)) == 4
def testMultipleActors(self):
# Create a bunch of actors and call a bunch of methods on all of them.
@@ -461,9 +452,8 @@ class ActorMethods(unittest.TestCase):
]
result_values = ray.get(results)
for i in range(num_actors):
self.assertEqual(
result_values[(num_increases * i):(num_increases * (i + 1))],
list(range(i + 1, num_increases + i + 1)))
v = result_values[(num_increases * i):(num_increases * (i + 1))]
assert v == list(range(i + 1, num_increases + i + 1))
# Reset the actor values.
[actor.reset.remote() for actor in actors]
@@ -474,9 +464,8 @@ class ActorMethods(unittest.TestCase):
results += [actor.increase.remote() for actor in actors]
result_values = ray.get(results)
for j in range(num_increases):
self.assertEqual(
result_values[(num_actors * j):(num_actors * (j + 1))],
num_actors * [j + 1])
v = result_values[(num_actors * j):(num_actors * (j + 1))]
assert v == num_actors * [j + 1]
class ActorNesting(unittest.TestCase):
@@ -521,16 +510,15 @@ class ActorNesting(unittest.TestCase):
actor = Actor.remote(1)
values = ray.get(actor.get_values.remote())
self.assertEqual(values[0], 1)
self.assertEqual(values[1], val2)
self.assertEqual(ray.get(values[2]), list(range(1, 6)))
self.assertEqual(values[3], list(range(1, 6)))
assert values[0] == 1
assert values[1] == val2
assert ray.get(values[2]) == list(range(1, 6))
assert values[3] == list(range(1, 6))
self.assertEqual(ray.get(ray.get(actor.f.remote())), list(range(1, 6)))
self.assertEqual(ray.get(actor.g.remote()), list(range(1, 6)))
self.assertEqual(
ray.get(actor.h.remote([f.remote(i) for i in range(5)])),
list(range(1, 6)))
assert ray.get(ray.get(actor.f.remote())) == list(range(1, 6))
assert ray.get(actor.g.remote()) == list(range(1, 6))
assert ray.get(actor.h.remote(
[f.remote(i) for i in range(5)])) == list(range(1, 6))
def testDefineActorWithinActor(self):
# Make sure we can use remote funtions within actors.
@@ -557,7 +545,7 @@ class ActorNesting(unittest.TestCase):
return self.x, ray.get(self.actor2.get_value.remote())
actor1 = Actor1.remote(3)
self.assertEqual(ray.get(actor1.get_values.remote(5)), (3, 5))
assert ray.get(actor1.get_values.remote(5)) == (3, 5)
def testUseActorWithinActor(self):
# Make sure we can use actors within actors.
@@ -581,7 +569,7 @@ class ActorNesting(unittest.TestCase):
return self.x, ray.get(self.actor1.get_val.remote())
actor2 = Actor2.remote(3, 4)
self.assertEqual(ray.get(actor2.get_values.remote(5)), (3, 4))
assert ray.get(actor2.get_values.remote(5)) == (3, 4)
def testDefineActorWithinRemoteFunction(self):
# Make sure we can define and actors within remote funtions.
@@ -600,10 +588,9 @@ class ActorNesting(unittest.TestCase):
actor = Actor1.remote(x)
return ray.get([actor.get_value.remote() for _ in range(n)])
self.assertEqual(ray.get(f.remote(3, 1)), [3])
self.assertEqual(
ray.get([f.remote(i, 20) for i in range(10)]),
[20 * [i] for i in range(10)])
assert ray.get(f.remote(3, 1)) == [3]
assert ray.get([f.remote(i, 20)
for i in range(10)]) == [20 * [i] for i in range(10)]
def testUseActorWithinRemoteFunction(self):
# Make sure we can create and use actors within remote funtions.
@@ -622,7 +609,7 @@ class ActorNesting(unittest.TestCase):
actor = Actor1.remote(x)
return ray.get(actor.get_values.remote())
self.assertEqual(ray.get(f.remote(3)), 3)
assert ray.get(f.remote(3)) == 3
def testActorImportCounter(self):
# This is mostly a test of the export counters to make sure that when
@@ -652,7 +639,7 @@ class ActorNesting(unittest.TestCase):
actor = Actor.remote()
return ray.get(actor.get_val.remote())
self.assertEqual(ray.get(g.remote()), num_remote_functions - 1)
assert ray.get(g.remote()) == num_remote_functions - 1
class ActorInheritance(unittest.TestCase):
@@ -683,8 +670,8 @@ class ActorInheritance(unittest.TestCase):
return self.f()
actor = Actor.remote(1)
self.assertEqual(ray.get(actor.get_value.remote()), 1)
self.assertEqual(ray.get(actor.g.remote(5)), 6)
assert ray.get(actor.get_value.remote()) == 1
assert ray.get(actor.g.remote(5)) == 6
class ActorSchedulingProperties(unittest.TestCase):
@@ -711,7 +698,7 @@ class ActorSchedulingProperties(unittest.TestCase):
return ray.worker.global_worker.worker_id
resulting_ids = ray.get([f.remote() for _ in range(100)])
self.assertNotIn(actor_id, resulting_ids)
assert actor_id not in resulting_ids
class ActorsOnMultipleNodes(unittest.TestCase):
@@ -728,7 +715,7 @@ class ActorsOnMultipleNodes(unittest.TestCase):
f = Foo.remote()
ready_ids, _ = ray.wait([f.method.remote()], timeout=100)
self.assertEquals(ready_ids, [])
assert ready_ids == []
def testActorLoadBalancing(self):
num_local_schedulers = 3
@@ -763,7 +750,7 @@ class ActorsOnMultipleNodes(unittest.TestCase):
and all(count >= minimum_count for count in counts)):
break
attempts += 1
self.assertLess(attempts, num_attempts)
assert attempts < num_attempts
# Make sure we can get the results of a bunch of tasks.
results = []
@@ -812,19 +799,18 @@ class ActorsWithGPUs(unittest.TestCase):
locations_and_ids = ray.get(
[actor.get_location_and_ids.remote() for actor in actors])
node_names = {location for location, gpu_id in locations_and_ids}
self.assertEqual(len(node_names), num_local_schedulers)
assert len(node_names) == num_local_schedulers
location_actor_combinations = []
for node_name in node_names:
for gpu_id in range(num_gpus_per_scheduler):
location_actor_combinations.append((node_name, (gpu_id, )))
self.assertEqual(
set(locations_and_ids), set(location_actor_combinations))
assert set(locations_and_ids) == set(location_actor_combinations)
# Creating a new actor should fail because all of the GPUs are being
# used.
a = Actor1.remote()
ready_ids, _ = ray.wait([a.get_location_and_ids.remote()], timeout=10)
self.assertEqual(ready_ids, [])
assert ready_ids == []
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
@@ -856,20 +842,20 @@ class ActorsWithGPUs(unittest.TestCase):
locations_and_ids = ray.get(
[actor.get_location_and_ids.remote() for actor in actors1])
node_names = {location for location, gpu_id in locations_and_ids}
self.assertEqual(len(node_names), num_local_schedulers)
assert len(node_names) == num_local_schedulers
# Keep track of which GPU IDs are being used for each location.
gpus_in_use = {node_name: [] for node_name in node_names}
for location, gpu_ids in locations_and_ids:
gpus_in_use[location].extend(gpu_ids)
for node_name in node_names:
self.assertEqual(len(set(gpus_in_use[node_name])), 4)
assert len(set(gpus_in_use[node_name])) == 4
# Creating a new actor should fail because all of the GPUs are being
# used.
a = Actor1.remote()
ready_ids, _ = ray.wait([a.get_location_and_ids.remote()], timeout=10)
self.assertEqual(ready_ids, [])
assert ready_ids == []
# We should be able to create more actors that use only a single GPU.
@ray.remote(num_gpus=1)
@@ -887,20 +873,19 @@ class ActorsWithGPUs(unittest.TestCase):
# Make sure that no two actors are assigned to the same GPU.
locations_and_ids = ray.get(
[actor.get_location_and_ids.remote() for actor in actors2])
self.assertEqual(node_names,
{location
for location, gpu_id in locations_and_ids})
names = {location for location, gpu_id in locations_and_ids}
assert node_names == names
for location, gpu_ids in locations_and_ids:
gpus_in_use[location].extend(gpu_ids)
for node_name in node_names:
self.assertEqual(len(gpus_in_use[node_name]), 5)
self.assertEqual(set(gpus_in_use[node_name]), set(range(5)))
assert len(gpus_in_use[node_name]) == 5
assert set(gpus_in_use[node_name]) == set(range(5))
# Creating a new actor should fail because all of the GPUs are being
# used.
a = Actor2.remote()
ready_ids, _ = ray.wait([a.get_location_and_ids.remote()], timeout=10)
self.assertEqual(ready_ids, [])
assert ready_ids == []
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
@@ -931,22 +916,21 @@ class ActorsWithGPUs(unittest.TestCase):
locations_and_ids = ray.get(
[actor.get_location_and_ids.remote() for actor in actors])
node_names = {location for location, gpu_id in locations_and_ids}
self.assertEqual(len(node_names), 2)
assert len(node_names) == 2
for node_name in node_names:
node_gpu_ids = [
gpu_id for location, gpu_id in locations_and_ids
if location == node_name
]
self.assertIn(len(node_gpu_ids), [5, 10])
self.assertEqual(
set(node_gpu_ids), {(i, )
for i in range(len(node_gpu_ids))})
assert len(node_gpu_ids) in [5, 10]
assert set(node_gpu_ids) == {(i, )
for i in range(len(node_gpu_ids))}
# Creating a new actor should fail because all of the GPUs are being
# used.
a = Actor1.remote()
ready_ids, _ = ray.wait([a.get_location_and_ids.remote()], timeout=10)
self.assertEqual(ready_ids, [])
assert ready_ids == []
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
@@ -995,7 +979,7 @@ class ActorsWithGPUs(unittest.TestCase):
# All the GPUs should be used up now.
a = Actor.remote()
ready_ids, _ = ray.wait([a.get_location_and_ids.remote()], timeout=10)
self.assertEqual(ready_ids, [])
assert ready_ids == []
@unittest.skipIf(sys.version_info < (3, 0), "This test requires Python 3.")
@unittest.skipIf(
@@ -1018,8 +1002,8 @@ class ActorsWithGPUs(unittest.TestCase):
second_interval = list_of_intervals[j]
# Check that list_of_intervals[i] and list_of_intervals[j]
# don't overlap.
self.assertLess(first_interval[0], first_interval[1])
self.assertLess(second_interval[0], second_interval[1])
assert first_interval[0] < first_interval[1]
assert second_interval[0] < second_interval[1]
intervals_nonoverlapping = (
first_interval[1] <= second_interval[0]
or second_interval[1] <= first_interval[0])
@@ -1085,9 +1069,8 @@ class ActorsWithGPUs(unittest.TestCase):
# Run a bunch of GPU tasks.
locations_to_intervals = locations_to_intervals_for_many_tasks()
# Make sure that all GPUs were used.
self.assertEqual(
len(locations_to_intervals),
num_local_schedulers * num_gpus_per_scheduler)
assert (len(locations_to_intervals) == num_local_schedulers *
num_gpus_per_scheduler)
# For each GPU, verify that the set of tasks that used this specific
# GPU did not overlap in time.
for locations in locations_to_intervals:
@@ -1099,20 +1082,19 @@ class ActorsWithGPUs(unittest.TestCase):
actor_location = (actor_location[0], actor_location[1][0])
# This check makes sure that actor_location is formatted the same way
# that the keys of locations_to_intervals are formatted.
self.assertIn(actor_location, locations_to_intervals)
assert actor_location in locations_to_intervals
# Run a bunch of GPU tasks.
locations_to_intervals = locations_to_intervals_for_many_tasks()
# Make sure that all but one of the GPUs were used.
self.assertEqual(
len(locations_to_intervals),
num_local_schedulers * num_gpus_per_scheduler - 1)
assert (len(locations_to_intervals) ==
num_local_schedulers * num_gpus_per_scheduler - 1)
# For each GPU, verify that the set of tasks that used this specific
# GPU did not overlap in time.
for locations in locations_to_intervals:
check_intervals_non_overlapping(locations_to_intervals[locations])
# Make sure that the actor's GPU was not used.
self.assertNotIn(actor_location, locations_to_intervals)
assert actor_location not in locations_to_intervals
# Create several more actors that use GPUs.
actors = [Actor1.remote() for _ in range(3)]
@@ -1122,17 +1104,16 @@ class ActorsWithGPUs(unittest.TestCase):
# Run a bunch of GPU tasks.
locations_to_intervals = locations_to_intervals_for_many_tasks()
# Make sure that all but 11 of the GPUs were used.
self.assertEqual(
len(locations_to_intervals),
num_local_schedulers * num_gpus_per_scheduler - 1 - 3)
assert (len(locations_to_intervals) ==
num_local_schedulers * num_gpus_per_scheduler - 1 - 3)
# For each GPU, verify that the set of tasks that used this specific
# GPU did not overlap in time.
for locations in locations_to_intervals:
check_intervals_non_overlapping(locations_to_intervals[locations])
# Make sure that the GPUs were not used.
self.assertNotIn(actor_location, locations_to_intervals)
assert actor_location not in locations_to_intervals
for location in actor_locations:
self.assertNotIn(location, locations_to_intervals)
assert location not in locations_to_intervals
# Create more actors to fill up all the GPUs.
more_actors = [
@@ -1146,7 +1127,7 @@ class ActorsWithGPUs(unittest.TestCase):
# Now if we run some GPU tasks, they should not be scheduled.
results = [f1.remote() for _ in range(30)]
ready_ids, remaining_ids = ray.wait(results, timeout=1000)
self.assertEqual(len(ready_ids), 0)
assert len(ready_ids) == 0
def testActorsAndTasksWithGPUsVersionTwo(self):
# Create tasks and actors that both use GPUs and make sure that they
@@ -1181,7 +1162,7 @@ class ActorsWithGPUs(unittest.TestCase):
actors.append(a)
gpu_ids = ray.get(results)
self.assertEqual(set(gpu_ids), set(range(10)))
assert set(gpu_ids) == set(range(10))
@unittest.skipIf(sys.version_info < (3, 0), "This test requires Python 3.")
def testActorsAndTaskResourceBookkeeping(self):
@@ -1216,9 +1197,9 @@ class ActorsWithGPUs(unittest.TestCase):
intervals = ray.get(interval_ids)
intervals.sort(key=lambda x: x[0])
for interval1, interval2 in zip(intervals[:-1], intervals[1:]):
self.assertLess(interval1[0], interval1[1])
self.assertLess(interval1[1], interval2[0])
self.assertLess(interval2[0], interval2[1])
assert interval1[0] < interval1[1]
assert interval1[1] < interval2[0]
assert interval2[0] < interval2[1]
def testBlockingActorTask(self):
ray.init(num_cpus=1, num_gpus=1)
@@ -1253,8 +1234,8 @@ class ActorsWithGPUs(unittest.TestCase):
actor = CPUFoo.remote()
x_id = actor.blocking_method.remote()
ready_ids, remaining_ids = ray.wait([x_id], timeout=1000)
self.assertEqual(ready_ids, [])
self.assertEqual(remaining_ids, [x_id])
assert ready_ids == []
assert remaining_ids == [x_id]
@ray.remote(num_gpus=1)
class GPUFoo(object):
@@ -1268,8 +1249,8 @@ class ActorsWithGPUs(unittest.TestCase):
actor = GPUFoo.remote()
x_id = actor.blocking_method.remote()
ready_ids, remaining_ids = ray.wait([x_id], timeout=1000)
self.assertEqual(ready_ids, [])
self.assertEqual(remaining_ids, [x_id])
assert ready_ids == []
assert remaining_ids == [x_id]
@unittest.skipIf(
@@ -1322,7 +1303,7 @@ class ActorReconstruction(unittest.TestCase):
# Get all of the results
results = ray.get(ids)
self.assertEqual(results, list(range(1, 1 + len(results))))
assert results == list(range(1, 1 + len(results)))
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False), "Hanging with new GCS API.")
@@ -1388,10 +1369,8 @@ class ActorReconstruction(unittest.TestCase):
# Get the results and check that they have the correct values.
for _, result_id_list in result_ids.items():
self.assertEqual(
ray.get(result_id_list), list(
range(1,
len(result_id_list) + 1)))
results = list(range(1, len(result_id_list) + 1))
assert ray.get(result_id_list) == results
def setup_counter_actor(self,
test_checkpoint=False,
@@ -1472,16 +1451,16 @@ class ActorReconstruction(unittest.TestCase):
process.wait()
# Check that the actor restored from a checkpoint.
self.assertTrue(ray.get(actor.test_restore.remote()))
assert ray.get(actor.test_restore.remote())
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(actor.inc.remote())
self.assertEqual(x, 101)
assert x == 101
# Check that the number of inc calls since actor initialization is less
# than the counter value, since the actor initialized from a
# checkpoint.
num_inc_calls = ray.get(actor.get_num_inc_calls.remote())
self.assertLess(num_inc_calls, x)
assert num_inc_calls < x
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False), "Hanging with new GCS API.")
@@ -1498,16 +1477,16 @@ class ActorReconstruction(unittest.TestCase):
process.wait()
# Check that the actor restored from a checkpoint.
self.assertTrue(ray.get(actor.test_restore.remote()))
assert ray.get(actor.test_restore.remote())
# Check that the number of inc calls since actor initialization is
# exactly zero, since there could not have been another inc call since
# the remote checkpoint.
num_inc_calls = ray.get(actor.get_num_inc_calls.remote())
self.assertEqual(num_inc_calls, 0)
assert num_inc_calls == 0
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(actor.inc.remote())
self.assertEqual(x, 101)
assert x == 101
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False), "Hanging with new GCS API.")
@@ -1523,17 +1502,17 @@ class ActorReconstruction(unittest.TestCase):
process.wait()
# Check that the actor restored from a checkpoint.
self.assertTrue(ray.get(actor.test_restore.remote()))
assert ray.get(actor.test_restore.remote())
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(actor.inc.remote())
self.assertEqual(x, 101)
assert x == 101
# Check that the number of inc calls since actor initialization is less
# than the counter value, since the actor initialized from a
# checkpoint.
num_inc_calls = ray.get(actor.get_num_inc_calls.remote())
self.assertLess(num_inc_calls, x)
self.assertLess(5, num_inc_calls)
assert num_inc_calls < x
assert 5 < num_inc_calls
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False), "Hanging with new GCS API.")
@@ -1552,18 +1531,17 @@ class ActorReconstruction(unittest.TestCase):
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(actor.inc.remote())
self.assertEqual(x, 101)
assert x == 101
# Check that the number of inc calls since actor initialization is
# equal to the counter value, since the actor did not initialize from a
# checkpoint.
num_inc_calls = ray.get(actor.get_num_inc_calls.remote())
self.assertEqual(num_inc_calls, x)
assert num_inc_calls == x
# Check that errors were raised when trying to save the checkpoint.
errors = ray.error_info()
self.assertLess(0, len(errors))
assert 0 < len(errors)
for error in errors:
self.assertEqual(error["type"],
ray_constants.CHECKPOINT_PUSH_ERROR)
assert error["type"] == ray_constants.CHECKPOINT_PUSH_ERROR
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False), "Hanging with new GCS API.")
@@ -1582,19 +1560,18 @@ class ActorReconstruction(unittest.TestCase):
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(actor.inc.remote())
self.assertEqual(x, 101)
assert x == 101
# Check that the number of inc calls since actor initialization is
# equal to the counter value, since the actor did not initialize from a
# checkpoint.
num_inc_calls = ray.get(actor.get_num_inc_calls.remote())
self.assertEqual(num_inc_calls, x)
assert num_inc_calls == x
# Check that an error was raised when trying to resume from the
# checkpoint.
errors = ray.error_info()
self.assertEqual(len(errors), 1)
assert len(errors) == 1
for error in errors:
self.assertEqual(error["type"],
ray_constants.CHECKPOINT_PUSH_ERROR)
assert error["type"] == ray_constants.CHECKPOINT_PUSH_ERROR
@unittest.skip("Fork/join consistency not yet implemented.")
def testDistributedHandle(self):
@@ -1626,11 +1603,11 @@ class ActorReconstruction(unittest.TestCase):
process.wait()
# Check that the actor did not restore from a checkpoint.
self.assertFalse(ray.get(counter.test_restore.remote()))
assert not ray.get(counter.test_restore.remote())
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(counter.inc.remote())
self.assertEqual(x, count + 1)
assert x == count + 1
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False), "Hanging with new GCS API.")
@@ -1664,16 +1641,16 @@ class ActorReconstruction(unittest.TestCase):
process.wait()
# Check that the actor restored from a checkpoint.
self.assertTrue(ray.get(counter.test_restore.remote()))
assert ray.get(counter.test_restore.remote())
# Check that the number of inc calls since actor initialization is
# exactly zero, since there could not have been another inc call since
# the remote checkpoint.
num_inc_calls = ray.get(counter.get_num_inc_calls.remote())
self.assertEqual(num_inc_calls, 0)
assert num_inc_calls == 0
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(counter.inc.remote())
self.assertEqual(x, count + 1)
assert x == count + 1
@unittest.skip("Fork/join consistency not yet implemented.")
def testCheckpointDistributedHandle(self):
@@ -1705,11 +1682,11 @@ class ActorReconstruction(unittest.TestCase):
process.wait()
# Check that the actor restored from a checkpoint.
self.assertTrue(ray.get(counter.test_restore.remote()))
assert ray.get(counter.test_restore.remote())
# Check that we can submit another call on the actor and get the
# correct counter result.
x = ray.get(counter.inc.remote())
self.assertEqual(x, count + 1)
assert x == count + 1
def _testNondeterministicReconstruction(
self, num_forks, num_items_per_fork, num_forks_to_wait):
@@ -1779,11 +1756,10 @@ class ActorReconstruction(unittest.TestCase):
ray.get(enqueue_tasks)
reconstructed_queue = ray.get(actor.read.remote())
# Make sure the final queue has all items from all forks.
self.assertEqual(
len(reconstructed_queue), num_forks * num_items_per_fork)
assert len(reconstructed_queue) == num_forks * num_items_per_fork
# Make sure that the prefix of the final queue matches the queue from
# the initial execution.
self.assertEqual(queue, reconstructed_queue[:len(queue)])
assert queue == reconstructed_queue[:len(queue)]
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False),
@@ -1834,7 +1810,7 @@ class DistributedActorHandles(unittest.TestCase):
items = ray.get(queue.read.remote())
for i in range(num_iters):
filtered_items = [item[1] for item in items if item[0] == i]
self.assertEqual(filtered_items, list(range(1)))
assert filtered_items == list(range(1))
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
@@ -1858,7 +1834,7 @@ class DistributedActorHandles(unittest.TestCase):
items = ray.get(queue.read.remote())
for i in range(num_forks):
filtered_items = [item[1] for item in items if item[0] == i]
self.assertEqual(filtered_items, list(range(num_items_per_fork)))
assert filtered_items == list(range(num_items_per_fork))
@unittest.skip("Garbage collection for distributed actor handles not "
"implemented.")
@@ -1907,7 +1883,7 @@ class DistributedActorHandles(unittest.TestCase):
assert ray.get(counter.inc.remote()) == 2
assert ray.get(new_counter.inc.remote()) == 3
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.get(f.remote())
# The below test works, but do we want to disallow this usage?
@@ -1945,25 +1921,25 @@ class DistributedActorHandles(unittest.TestCase):
ray.experimental.register_actor("f1", f1)
# Test getting f.
f2 = ray.experimental.get_actor("f1")
self.assertEqual(f1._actor_id, f2._actor_id)
assert f1._actor_id == f2._actor_id
# Test same name register shall raise error.
with self.assertRaises(ValueError):
with pytest.raises(ValueError):
ray.experimental.register_actor("f1", f2)
# Test register with wrong object type.
with self.assertRaises(TypeError):
with pytest.raises(TypeError):
ray.experimental.register_actor("f3", 1)
# Test getting a nonexistent actor.
with self.assertRaises(ValueError):
with pytest.raises(ValueError):
ray.experimental.get_actor("nonexistent")
# Test method
self.assertEqual(ray.get(f1.method.remote()), 1)
self.assertEqual(ray.get(f2.method.remote()), 2)
self.assertEqual(ray.get(f1.method.remote()), 3)
self.assertEqual(ray.get(f2.method.remote()), 4)
assert ray.get(f1.method.remote()) == 1
assert ray.get(f2.method.remote()) == 2
assert ray.get(f1.method.remote()) == 3
assert ray.get(f2.method.remote()) == 4
@pytest.fixture
test/array_test.py
+3 -3
View File
@@ -162,7 +162,7 @@ class DistributedArrayTest(unittest.TestCase):
x_val = ray.get(da.assemble.remote(x))
q_val = ray.get(da.assemble.remote(q))
r_val = ray.get(r)
self.assertTrue(r_val.shape == (K, shape[1]))
assert r_val.shape == (K, shape[1])
assert_equal(r_val, np.triu(r_val))
assert_almost_equal(x_val, np.dot(q_val, r_val))
assert_almost_equal(np.dot(q_val.T, q_val), np.eye(K))
@@ -226,8 +226,8 @@ class DistributedArrayTest(unittest.TestCase):
a_val = ray.get(da.assemble.remote(a))
q_val = ray.get(da.assemble.remote(q))
r_val = ray.get(da.assemble.remote(r))
self.assertEqual(q_val.shape, (d1, K))
self.assertEqual(r_val.shape, (K, d2))
assert q_val.shape == (d1, K)
assert r_val.shape == (K, d2)
assert_almost_equal(np.dot(q_val.T, q_val), np.eye(K))
assert_equal(r_val, np.triu(r_val))
assert_almost_equal(a_val, np.dot(q_val, r_val))
test/autoscaler_test.py
+60 -61
View File
@@ -17,6 +17,7 @@ from ray.autoscaler.autoscaler import StandardAutoscaler, LoadMetrics, \
from ray.autoscaler.tags import TAG_RAY_NODE_TYPE, TAG_RAY_NODE_STATUS
from ray.autoscaler.node_provider import NODE_PROVIDERS, NodeProvider
from ray.autoscaler.updater import NodeUpdaterThread
import pytest
class MockNode(object):
@@ -131,41 +132,41 @@ class LoadMetricsTest(unittest.TestCase):
def testUpdate(self):
lm = LoadMetrics()
lm.update("1.1.1.1", {"CPU": 2}, {"CPU": 1})
self.assertEqual(lm.approx_workers_used(), 0.5)
assert lm.approx_workers_used() == 0.5
lm.update("1.1.1.1", {"CPU": 2}, {"CPU": 0})
self.assertEqual(lm.approx_workers_used(), 1.0)
assert lm.approx_workers_used() == 1.0
lm.update("2.2.2.2", {"CPU": 2}, {"CPU": 0})
self.assertEqual(lm.approx_workers_used(), 2.0)
assert lm.approx_workers_used() == 2.0
def testPruneByNodeIp(self):
lm = LoadMetrics()
lm.update("1.1.1.1", {"CPU": 1}, {"CPU": 0})
lm.update("2.2.2.2", {"CPU": 1}, {"CPU": 0})
lm.prune_active_ips({"1.1.1.1", "4.4.4.4"})
self.assertEqual(lm.approx_workers_used(), 1.0)
assert lm.approx_workers_used() == 1.0
def testBottleneckResource(self):
lm = LoadMetrics()
lm.update("1.1.1.1", {"CPU": 2}, {"CPU": 0})
lm.update("2.2.2.2", {"CPU": 2, "GPU": 16}, {"CPU": 2, "GPU": 2})
self.assertEqual(lm.approx_workers_used(), 1.88)
assert lm.approx_workers_used() == 1.88
def testHeartbeat(self):
lm = LoadMetrics()
lm.update("1.1.1.1", {"CPU": 2}, {"CPU": 1})
lm.mark_active("2.2.2.2")
self.assertIn("1.1.1.1", lm.last_heartbeat_time_by_ip)
self.assertIn("2.2.2.2", lm.last_heartbeat_time_by_ip)
self.assertNotIn("3.3.3.3", lm.last_heartbeat_time_by_ip)
assert "1.1.1.1" in lm.last_heartbeat_time_by_ip
assert "2.2.2.2" in lm.last_heartbeat_time_by_ip
assert "3.3.3.3" not in lm.last_heartbeat_time_by_ip
def testDebugString(self):
lm = LoadMetrics()
lm.update("1.1.1.1", {"CPU": 2}, {"CPU": 0})
lm.update("2.2.2.2", {"CPU": 2, "GPU": 16}, {"CPU": 2, "GPU": 2})
debug = lm.debug_string()
self.assertIn("ResourceUsage: 2.0/4.0 CPU, 14.0/16.0 GPU", debug)
self.assertIn("NumNodesConnected: 2", debug)
self.assertIn("NumNodesUsed: 1.88", debug)
assert "ResourceUsage: 2.0/4.0 CPU, 14.0/16.0 GPU" in debug
assert "NumNodesConnected: 2" in debug
assert "NumNodesUsed: 1.88" in debug
class AutoscalingTest(unittest.TestCase):
@@ -213,10 +214,9 @@ class AutoscalingTest(unittest.TestCase):
def testInvalidConfig(self):
invalid_config = "/dev/null"
self.assertRaises(
ValueError,
lambda: StandardAutoscaler(
invalid_config, LoadMetrics(), update_interval_s=0))
with pytest.raises(ValueError):
StandardAutoscaler(
invalid_config, LoadMetrics(), update_interval_s=0)
def testValidation(self):
"""Ensures that schema validation is working."""
@@ -227,17 +227,17 @@ class AutoscalingTest(unittest.TestCase):
self.fail("Test config did not pass validation test!")
config["blah"] = "blah"
with self.assertRaises(ValueError):
with pytest.raises(ValueError):
validate_config(config)
del config["blah"]
config["provider"]["blah"] = "blah"
with self.assertRaises(ValueError):
with pytest.raises(ValueError):
validate_config(config)
del config["provider"]["blah"]
del config["provider"]
with self.assertRaises(ValueError):
with pytest.raises(ValueError):
validate_config(config)
def testValidateDefaultConfig(self):
@@ -258,7 +258,7 @@ class AutoscalingTest(unittest.TestCase):
self.provider = MockProvider()
autoscaler = StandardAutoscaler(
config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
self.assertEqual(len(self.provider.nodes({})), 0)
assert len(self.provider.nodes({})) == 0
autoscaler.update()
self.waitForNodes(2)
autoscaler.update()
@@ -324,25 +324,25 @@ class AutoscalingTest(unittest.TestCase):
max_concurrent_launches=5,
max_failures=0,
update_interval_s=0)
self.assertEqual(len(self.provider.nodes({})), 0)
assert len(self.provider.nodes({})) == 0
# Update will try to create, but will block until we set the flag
self.provider.ready_to_create.clear()
autoscaler.update()
self.assertEqual(autoscaler.num_launches_pending.value, 2)
self.assertEqual(len(self.provider.nodes({})), 0)
assert autoscaler.num_launches_pending.value == 2
assert len(self.provider.nodes({})) == 0
# Set the flag, check it updates
self.provider.ready_to_create.set()
self.waitForNodes(2)
self.assertEqual(autoscaler.num_launches_pending.value, 0)
assert autoscaler.num_launches_pending.value == 0
# Update the config to reduce the cluster size
new_config = SMALL_CLUSTER.copy()
new_config["max_workers"] = 1
self.write_config(new_config)
autoscaler.update()
self.assertEqual(len(self.provider.nodes({})), 1)
assert len(self.provider.nodes({})) == 1
def testDelayedLaunchWithFailure(self):
config = SMALL_CLUSTER.copy()
@@ -357,7 +357,7 @@ class AutoscalingTest(unittest.TestCase):
max_concurrent_launches=8,
max_failures=0,
update_interval_s=0)
self.assertEqual(len(self.provider.nodes({})), 0)
assert len(self.provider.nodes({})) == 0
# update() should launch a wave of 5 nodes (max_launch_batch)
# Force this first wave to block.
@@ -370,8 +370,8 @@ class AutoscalingTest(unittest.TestCase):
else: # Python 2.7
waiters = rtc1._Event__cond._Condition__waiters
self.waitFor(lambda: len(waiters) == 1)
self.assertEqual(autoscaler.num_launches_pending.value, 5)
self.assertEqual(len(self.provider.nodes({})), 0)
assert autoscaler.num_launches_pending.value == 5
assert len(self.provider.nodes({})) == 0
# Call update() to launch a second wave of 3 nodes,
# as 5 + 3 = 8 = max_concurrent_launches.
@@ -381,24 +381,24 @@ class AutoscalingTest(unittest.TestCase):
rtc2.set()
autoscaler.update()
self.waitForNodes(3)
self.assertEqual(autoscaler.num_launches_pending.value, 5)
assert autoscaler.num_launches_pending.value == 5
# The first wave of 5 will now tragically fail
self.provider.fail_creates = True
rtc1.set()
self.waitFor(lambda: autoscaler.num_launches_pending.value == 0)
self.assertEqual(len(self.provider.nodes({})), 3)
assert len(self.provider.nodes({})) == 3
# Retry the first wave, allowing it to succeed this time
self.provider.fail_creates = False
autoscaler.update()
self.waitForNodes(8)
self.assertEqual(autoscaler.num_launches_pending.value, 0)
assert autoscaler.num_launches_pending.value == 0
# Final wave of 2 nodes
autoscaler.update()
self.waitForNodes(10)
self.assertEqual(autoscaler.num_launches_pending.value, 0)
assert autoscaler.num_launches_pending.value == 0
def testUpdateThrottling(self):
config_path = self.write_config(SMALL_CLUSTER)
@@ -412,7 +412,7 @@ class AutoscalingTest(unittest.TestCase):
update_interval_s=10)
autoscaler.update()
self.waitForNodes(2)
self.assertEqual(autoscaler.num_launches_pending.value, 0)
assert autoscaler.num_launches_pending.value == 0
new_config = SMALL_CLUSTER.copy()
new_config["max_workers"] = 1
self.write_config(new_config)
@@ -420,8 +420,8 @@ class AutoscalingTest(unittest.TestCase):
# not updated yet
# note that node termination happens in the main thread, so
# we do not need to add any delay here before checking
self.assertEqual(len(self.provider.nodes({})), 2)
self.assertEqual(autoscaler.num_launches_pending.value, 0)
assert len(self.provider.nodes({})) == 2
assert autoscaler.num_launches_pending.value == 0
def testLaunchConfigChange(self):
config_path = self.write_config(SMALL_CLUSTER)
@@ -460,8 +460,8 @@ class AutoscalingTest(unittest.TestCase):
for _ in range(10):
autoscaler.update()
time.sleep(0.1)
self.assertEqual(autoscaler.num_launches_pending.value, 0)
self.assertEqual(len(self.provider.nodes({})), 2)
assert autoscaler.num_launches_pending.value == 0
assert len(self.provider.nodes({})) == 2
# New a good config again
new_config = SMALL_CLUSTER.copy()
@@ -479,7 +479,8 @@ class AutoscalingTest(unittest.TestCase):
config_path, LoadMetrics(), max_failures=2, update_interval_s=0)
autoscaler.update()
autoscaler.update()
self.assertRaises(Exception, autoscaler.update)
with pytest.raises(Exception):
autoscaler.update()
def testLaunchNewNodeOnOutOfBandTerminate(self):
config_path = self.write_config(SMALL_CLUSTER)
@@ -491,7 +492,7 @@ class AutoscalingTest(unittest.TestCase):
self.waitForNodes(2)
for node in self.provider.mock_nodes.values():
node.state = "terminated"
self.assertEqual(len(self.provider.nodes({})), 0)
assert len(self.provider.nodes({})) == 0
autoscaler.update()
self.waitForNodes(2)
@@ -581,12 +582,12 @@ class AutoscalingTest(unittest.TestCase):
lm = LoadMetrics()
autoscaler = StandardAutoscaler(
config_path, lm, max_failures=0, update_interval_s=0)
self.assertEqual(len(self.provider.nodes({})), 0)
assert len(self.provider.nodes({})) == 0
autoscaler.update()
self.waitForNodes(1)
autoscaler.update()
self.assertEqual(autoscaler.num_launches_pending.value, 0)
self.assertEqual(len(self.provider.nodes({})), 1)
assert autoscaler.num_launches_pending.value == 0
assert len(self.provider.nodes({})) == 1
# Scales up as nodes are reported as used
local_ip = services.get_node_ip_address()
@@ -602,20 +603,20 @@ class AutoscalingTest(unittest.TestCase):
lm.update("172.0.0.0", {"CPU": 2}, {"CPU": 2})
lm.update("172.0.0.1", {"CPU": 2}, {"CPU": 2})
autoscaler.update()
self.assertEqual(autoscaler.num_launches_pending.value, 0)
self.assertEqual(len(self.provider.nodes({})), 5)
assert autoscaler.num_launches_pending.value == 0
assert len(self.provider.nodes({})) == 5
# Scales down as nodes become unused
lm.last_used_time_by_ip["172.0.0.0"] = 0
lm.last_used_time_by_ip["172.0.0.1"] = 0
autoscaler.update()
self.assertEqual(autoscaler.num_launches_pending.value, 0)
self.assertEqual(len(self.provider.nodes({})), 3)
assert autoscaler.num_launches_pending.value == 0
assert len(self.provider.nodes({})) == 3
lm.last_used_time_by_ip["172.0.0.2"] = 0
lm.last_used_time_by_ip["172.0.0.3"] = 0
autoscaler.update()
self.assertEqual(autoscaler.num_launches_pending.value, 0)
self.assertEqual(len(self.provider.nodes({})), 1)
assert autoscaler.num_launches_pending.value == 0
assert len(self.provider.nodes({})) == 1
def testDontScaleBelowTarget(self):
config = SMALL_CLUSTER.copy()
@@ -627,10 +628,10 @@ class AutoscalingTest(unittest.TestCase):
lm = LoadMetrics()
autoscaler = StandardAutoscaler(
config_path, lm, max_failures=0, update_interval_s=0)
self.assertEqual(len(self.provider.nodes({})), 0)
assert len(self.provider.nodes({})) == 0
autoscaler.update()
self.assertEqual(autoscaler.num_launches_pending.value, 0)
self.assertEqual(len(self.provider.nodes({})), 0)
assert autoscaler.num_launches_pending.value == 0
assert len(self.provider.nodes({})) == 0
# Scales up as nodes are reported as used
local_ip = services.get_node_ip_address()
@@ -644,12 +645,12 @@ class AutoscalingTest(unittest.TestCase):
lm.update("172.0.0.0", {"CPU": 0}, {"CPU": 0})
lm.last_used_time_by_ip["172.0.0.0"] = 0
autoscaler.update()
self.assertEqual(len(self.provider.nodes({})), 1)
assert len(self.provider.nodes({})) == 1
# Reduce load on head => target nodes = 1 => target workers = 0
lm.update(local_ip, {"CPU": 2}, {"CPU": 1})
autoscaler.update()
self.assertEqual(len(self.provider.nodes({})), 0)
assert len(self.provider.nodes({})) == 0
def testRecoverUnhealthyWorkers(self):
config_path = self.write_config(SMALL_CLUSTER)
@@ -686,7 +687,7 @@ class AutoscalingTest(unittest.TestCase):
config_path = self.write_config(config)
autoscaler = StandardAutoscaler(
config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
self.assertIsInstance(autoscaler.provider, NodeProvider)
assert isinstance(autoscaler.provider, NodeProvider)
def testExternalNodeScalerWrongImport(self):
config = SMALL_CLUSTER.copy()
@@ -695,10 +696,9 @@ class AutoscalingTest(unittest.TestCase):
"module": "mymodule.provider_class",
}
invalid_provider = self.write_config(config)
self.assertRaises(
ImportError,
lambda: StandardAutoscaler(
invalid_provider, LoadMetrics(), update_interval_s=0))
with pytest.raises(ImportError):
StandardAutoscaler(
invalid_provider, LoadMetrics(), update_interval_s=0)
def testExternalNodeScalerWrongModuleFormat(self):
config = SMALL_CLUSTER.copy()
@@ -707,10 +707,9 @@ class AutoscalingTest(unittest.TestCase):
"module": "does-not-exist",
}
invalid_provider = self.write_config(config)
self.assertRaises(
ValueError,
lambda: StandardAutoscaler(
invalid_provider, LoadMetrics(), update_interval_s=0))
with pytest.raises(ValueError):
StandardAutoscaler(
invalid_provider, LoadMetrics(), update_interval_s=0)
if __name__ == "__main__":
test/component_failures_test.py
+8 -10
View File
@@ -50,9 +50,8 @@ class ComponentFailureTest(unittest.TestCase):
time.sleep(0.1)
# Make sure that nothing has died.
self.assertTrue(
ray.services.all_processes_alive(
exclude=[ray.services.PROCESS_TYPE_WORKER]))
assert ray.services.all_processes_alive(
exclude=[ray.services.PROCESS_TYPE_WORKER])
# This test checks that when a worker dies in the middle of a wait, the
# plasma store and manager will not die.
@@ -90,9 +89,8 @@ class ComponentFailureTest(unittest.TestCase):
time.sleep(0.1)
# Make sure that nothing has died.
self.assertTrue(
ray.services.all_processes_alive(
exclude=[ray.services.PROCESS_TYPE_WORKER]))
assert ray.services.all_processes_alive(
exclude=[ray.services.PROCESS_TYPE_WORKER])
def _testWorkerFailed(self, num_local_schedulers):
@ray.remote
@@ -170,14 +168,14 @@ class ComponentFailureTest(unittest.TestCase):
for process in components[1:]:
process.kill()
process.wait()
self.assertNotEqual(process.poll(), None)
assert not process.poll() is None
# Make sure that we can still get the objects after the executing tasks
# died.
results = ray.get(object_ids)
expected_results = 4 * list(
range(num_workers_per_scheduler * num_local_schedulers))
self.assertEqual(results, expected_results)
assert results == expected_results
def check_components_alive(self, component_type, check_component_alive):
"""Check that a given component type is alive on all worker nodes.
@@ -185,14 +183,14 @@ class ComponentFailureTest(unittest.TestCase):
components = ray.services.all_processes[component_type][1:]
for component in components:
if check_component_alive:
self.assertTrue(component.poll() is None)
assert component.poll() is None
else:
print("waiting for " + component_type + " with PID " +
str(component.pid) + "to terminate")
component.wait()
print("done waiting for " + component_type + " with PID " +
str(component.pid) + "to terminate")
self.assertTrue(not component.poll() is None)
assert not component.poll() is None
@unittest.skipIf(
os.environ.get('RAY_USE_NEW_GCS', False), "Hanging with new GCS API.")
test/cython_test.py
+4 -4
View File
@@ -22,7 +22,7 @@ class CythonTest(unittest.TestCase):
ray.shutdown()
def assertEqualHelper(self, cython_func, expected, *args):
self.assertEqual(get_ray_result(cython_func, *args), expected)
assert get_ray_result(cython_func, *args) == expected
def test_simple_func(self):
self.assertEqualHelper(cyth.simple_func, 6, 1, 2, 3)
@@ -40,9 +40,9 @@ class CythonTest(unittest.TestCase):
result2 = ray.get(a2.increment.remote())
result3 = ray.get(a2.increment.remote())
self.assertEqual(result1, 1)
self.assertEqual(result2, 1)
self.assertEqual(result3, 2)
assert result1 == 1
assert result2 == 1
assert result3 == 2
def test_numpy(self):
array = np.array([-1.0, 0.0, 1.0, 2.0])
test/failure_test.py
+47 -45
View File
@@ -11,6 +11,7 @@ import time
import unittest
import ray.ray_constants as ray_constants
import pytest
def relevant_errors(error_type):
@@ -48,30 +49,29 @@ class TaskStatusTest(unittest.TestCase):
throw_exception_fct1.remote()
throw_exception_fct1.remote()
wait_for_errors(ray_constants.TASK_PUSH_ERROR, 2)
self.assertEqual(
len(relevant_errors(ray_constants.TASK_PUSH_ERROR)), 2)
assert len(relevant_errors(ray_constants.TASK_PUSH_ERROR)) == 2
for task in relevant_errors(ray_constants.TASK_PUSH_ERROR):
self.assertIn("Test function 1 intentionally failed.",
task.get("message"))
msg = task.get("message")
assert "Test function 1 intentionally failed." in msg
x = throw_exception_fct2.remote()
try:
ray.get(x)
except Exception as e:
self.assertIn("Test function 2 intentionally failed.", str(e))
assert "Test function 2 intentionally failed." in str(e)
else:
# ray.get should throw an exception.
self.assertTrue(False)
assert False
x, y, z = throw_exception_fct3.remote(1.0)
for ref in [x, y, z]:
try:
ray.get(ref)
except Exception as e:
self.assertIn("Test function 3 intentionally failed.", str(e))
assert "Test function 3 intentionally failed." in str(e)
else:
# ray.get should throw an exception.
self.assertTrue(False)
assert False
@ray.remote
def f():
@@ -80,10 +80,10 @@ class TaskStatusTest(unittest.TestCase):
try:
ray.get(f.remote())
except Exception as e:
self.assertIn("This function failed.", str(e))
assert "This function failed." in str(e)
else:
# ray.get should throw an exception.
self.assertTrue(False)
assert False
def testFailImportingRemoteFunction(self):
ray.init(num_workers=2, driver_mode=ray.SILENT_MODE)
@@ -110,13 +110,14 @@ def temporary_helper_function():
return module.temporary_python_file()
wait_for_errors(ray_constants.REGISTER_REMOTE_FUNCTION_PUSH_ERROR, 2)
self.assertIn("No module named", ray.error_info()[0]["message"])
self.assertIn("No module named", ray.error_info()[1]["message"])
assert "No module named" in ray.error_info()[0]["message"]
assert "No module named" in ray.error_info()[1]["message"]
# Check that if we try to call the function it throws an exception and
# does not hang.
for _ in range(10):
self.assertRaises(Exception, lambda: ray.get(g.remote()))
with pytest.raises(Exception):
ray.get(g.remote())
f.close()
@@ -133,11 +134,10 @@ def temporary_helper_function():
ray.worker.global_worker.run_function_on_all_workers(f)
wait_for_errors(ray_constants.FUNCTION_TO_RUN_PUSH_ERROR, 2)
# Check that the error message is in the task info.
self.assertEqual(len(ray.error_info()), 2)
self.assertIn("Function to run failed.",
ray.error_info()[0]["message"])
self.assertIn("Function to run failed.",
ray.error_info()[1]["message"])
error_info = ray.error_info()
assert len(error_info) == 2
assert "Function to run failed." in error_info[0]["message"]
assert "Function to run failed." in error_info[1]["message"]
def testFailImportingActor(self):
ray.init(num_workers=2, driver_mode=ray.SILENT_MODE)
@@ -168,31 +168,29 @@ def temporary_helper_function():
return 1
# There should be no errors yet.
self.assertEqual(len(ray.error_info()), 0)
assert len(ray.error_info()) == 0
# Create an actor.
foo = Foo.remote()
# Wait for the error to arrive.
wait_for_errors(ray_constants.REGISTER_ACTOR_PUSH_ERROR, 1)
self.assertIn("No module named", ray.error_info()[0]["message"])
assert "No module named" in ray.error_info()[0]["message"]
# Wait for the error from when the __init__ tries to run.
wait_for_errors(ray_constants.TASK_PUSH_ERROR, 1)
self.assertIn(
"failed to be imported, and so cannot execute this method",
ray.error_info()[1]["message"])
assert ("failed to be imported, and so cannot execute this method" in
ray.error_info()[1]["message"])
# Check that if we try to get the function it throws an exception and
# does not hang.
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.get(foo.get_val.remote())
# Wait for the error from when the call to get_val.
wait_for_errors(ray_constants.TASK_PUSH_ERROR, 2)
self.assertIn(
"failed to be imported, and so cannot execute this method",
ray.error_info()[2]["message"])
assert ("failed to be imported, and so cannot execute this method" in
ray.error_info()[2]["message"])
f.close()
@@ -225,14 +223,14 @@ class ActorTest(unittest.TestCase):
# Make sure that we get errors from a failed constructor.
wait_for_errors(ray_constants.TASK_PUSH_ERROR, 1)
self.assertEqual(len(ray.error_info()), 1)
self.assertIn(error_message1, ray.error_info()[0]["message"])
assert len(ray.error_info()) == 1
assert error_message1 in ray.error_info()[0]["message"]
# Make sure that we get errors from a failed method.
a.fail_method.remote()
wait_for_errors(ray_constants.TASK_PUSH_ERROR, 2)
self.assertEqual(len(ray.error_info()), 2)
self.assertIn(error_message2, ray.error_info()[1]["message"])
assert len(ray.error_info()) == 2
assert error_message2 in ray.error_info()[1]["message"]
def testIncorrectMethodCalls(self):
ray.init(num_workers=0, driver_mode=ray.SILENT_MODE)
@@ -248,27 +246,27 @@ class ActorTest(unittest.TestCase):
# Make sure that we get errors if we call the constructor incorrectly.
# Create an actor with too few arguments.
with self.assertRaises(Exception):
with pytest.raises(Exception):
a = Actor.remote()
# Create an actor with too many arguments.
with self.assertRaises(Exception):
with pytest.raises(Exception):
a = Actor.remote(1, 2)
# Create an actor the correct number of arguments.
a = Actor.remote(1)
# Call a method with too few arguments.
with self.assertRaises(Exception):
with pytest.raises(Exception):
a.get_val.remote()
# Call a method with too many arguments.
with self.assertRaises(Exception):
with pytest.raises(Exception):
a.get_val.remote(1, 2)
# Call a method that doesn't exist.
with self.assertRaises(AttributeError):
with pytest.raises(AttributeError):
a.nonexistent_method()
with self.assertRaises(AttributeError):
with pytest.raises(AttributeError):
a.nonexistent_method.remote()
@@ -289,7 +287,7 @@ class WorkerDeath(unittest.TestCase):
wait_for_errors(ray_constants.WORKER_CRASH_PUSH_ERROR, 1)
wait_for_errors(ray_constants.WORKER_DIED_PUSH_ERROR, 1)
self.assertEqual(len(ray.error_info()), 2)
assert len(ray.error_info()) == 2
def testWorkerDying(self):
ray.init(num_workers=0, driver_mode=ray.SILENT_MODE)
@@ -303,9 +301,9 @@ class WorkerDeath(unittest.TestCase):
wait_for_errors(ray_constants.WORKER_DIED_PUSH_ERROR, 1)
self.assertEqual(len(ray.error_info()), 1)
self.assertIn("died or was killed while executing",
ray.error_info()[0]["message"])
error_info = ray.error_info()
assert len(error_info) == 1
assert "died or was killed while executing" in error_info[0]["message"]
def testActorWorkerDying(self):
ray.init(num_workers=0, driver_mode=ray.SILENT_MODE)
@@ -321,8 +319,10 @@ class WorkerDeath(unittest.TestCase):
a = Actor.remote()
[obj], _ = ray.wait([a.kill.remote()], timeout=5000)
self.assertRaises(Exception, lambda: ray.get(obj))
self.assertRaises(Exception, lambda: ray.get(consume.remote(obj)))
with pytest.raises(Exception):
ray.get(obj)
with pytest.raises(Exception):
ray.get(consume.remote(obj))
wait_for_errors(ray_constants.WORKER_DIED_PUSH_ERROR, 1)
def testActorWorkerDyingFutureTasks(self):
@@ -343,7 +343,8 @@ class WorkerDeath(unittest.TestCase):
time.sleep(0.1)
tasks2 = [a.sleep.remote() for _ in range(10)]
for obj in tasks1 + tasks2:
self.assertRaises(Exception, lambda: ray.get(obj))
with pytest.raises(Exception):
ray.get(obj)
wait_for_errors(ray_constants.WORKER_DIED_PUSH_ERROR, 1)
@@ -360,7 +361,8 @@ class WorkerDeath(unittest.TestCase):
os.kill(pid, 9)
time.sleep(0.1)
task2 = a.getpid.remote()
self.assertRaises(Exception, lambda: ray.get(task2))
with pytest.raises(Exception):
ray.get(task2)
@unittest.skipIf(
test/monitor_test.py
+3 -3
View File
@@ -24,7 +24,7 @@ class MonitorTest(unittest.TestCase):
])
lines = [m.strip() for m in stdout.split("\n")]
init_cmd = [m for m in lines if m.startswith("ray.init")]
self.assertEqual(1, len(init_cmd))
assert 1 == len(init_cmd)
redis_address = init_cmd[0].split("redis_address=\"")[-1][:-2]
def StateSummary():
@@ -74,12 +74,12 @@ class MonitorTest(unittest.TestCase):
# values computed in the Driver function are being updated slowly and
# so the call to StateSummary() is getting outdated values. This could
# be fixed by looping until StateSummary() returns the desired values.
self.assertTrue(success.value)
assert success.value
# Check that objects, tasks, and functions are cleaned up.
ray.init(redis_address=redis_address)
# The assertion below can fail if the monitor is too slow to clean up
# the global state.
self.assertEqual((0, 1), StateSummary()[:2])
assert (0, 1) == StateSummary()[:2]
ray.shutdown()
subprocess.Popen(["ray", "stop"]).wait()
test/multi_node_test.py
+17 -16
View File
@@ -9,6 +9,7 @@ import unittest
import ray
from ray.test.test_utils import run_and_get_output
import pytest
def run_string_as_driver(driver_script):
@@ -49,7 +50,7 @@ class MultiNodeTest(unittest.TestCase):
ray.init(redis_address=self.redis_address, driver_mode=ray.SILENT_MODE)
# There shouldn't be any errors yet.
self.assertEqual(len(ray.error_info()), 0)
assert len(ray.error_info()) == 0
error_string1 = "error_string1"
error_string2 = "error_string2"
@@ -59,7 +60,7 @@ class MultiNodeTest(unittest.TestCase):
raise Exception(error_string1)
# Run a remote function that throws an error.
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.get(f.remote())
# Wait for the error to appear in Redis.
@@ -68,8 +69,8 @@ class MultiNodeTest(unittest.TestCase):
print("Waiting for error to appear.")
# Make sure we got the error.
self.assertEqual(len(ray.error_info()), 1)
self.assertIn(error_string1, ray.error_info()[0]["message"])
assert len(ray.error_info()) == 1
assert error_string1 in ray.error_info()[0]["message"]
# Start another driver and make sure that it does not receive this
# error. Make the other driver throw an error, and make sure it
@@ -104,12 +105,12 @@ print("success")
out = run_string_as_driver(driver_script)
# Make sure the other driver succeeded.
self.assertIn("success", out)
assert "success" in out
# Make sure that the other error message doesn't show up for this
# driver.
self.assertEqual(len(ray.error_info()), 1)
self.assertIn(error_string1, ray.error_info()[0]["message"])
assert len(ray.error_info()) == 1
assert error_string1 in ray.error_info()[0]["message"]
def testRemoteFunctionIsolation(self):
# This test will run multiple remote functions with the same names in
@@ -146,10 +147,10 @@ print("success")
for _ in range(10000):
result = ray.get([f.remote(), g.remote(0)])
self.assertEqual(result, [1, 2])
assert result == [1, 2]
# Make sure the other driver succeeded.
self.assertIn("success", out)
assert "success" in out
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
@@ -187,11 +188,11 @@ print("success")
for _ in range(3):
out = run_string_as_driver(driver_script1)
# Make sure the first driver ran to completion.
self.assertIn("success", out)
assert "success" in out
out = run_string_as_driver(driver_script2)
# Make sure the first driver ran to completion.
self.assertIn("success", out)
self.assertTrue(ray.services.all_processes_alive())
assert "success" in out
assert ray.services.all_processes_alive()
class StartRayScriptTest(unittest.TestCase):
@@ -254,7 +255,7 @@ class StartRayScriptTest(unittest.TestCase):
subprocess.Popen(["ray", "stop"]).wait()
# Test starting Ray with invalid arguments.
with self.assertRaises(Exception):
with pytest.raises(Exception):
run_and_get_output([
"ray", "start", "--head", "--redis-address", "127.0.0.1:6379"
])
@@ -273,7 +274,7 @@ class StartRayScriptTest(unittest.TestCase):
def f():
return 1
self.assertEqual(ray.get(f.remote()), 1)
assert ray.get(f.remote()) == 1
# Kill the Ray cluster.
subprocess.Popen(["ray", "stop"]).wait()
@@ -295,7 +296,7 @@ print("success")
out = run_string_as_driver(driver_script)
# Make sure the other driver succeeded.
self.assertIn("success", out)
assert "success" in out
class RunDriverForMultipleTimesTest(unittest.TestCase):
@@ -342,7 +343,7 @@ print("success")
for i in range(2):
out = run_string_as_driver(driver_script)
self.assertIn("success", out)
assert "success" in out
if __name__ == "__main__":
test/runtest.py
+218 -246
View File
@@ -307,7 +307,7 @@ class WorkerTest(unittest.TestCase):
return x
values = ray.get([f.remote(1) for i in range(num_workers * 2)])
self.assertEqual(values, [1] * (num_workers * 2))
assert values == [1] * (num_workers * 2)
def testPutGet(self):
ray.init(num_workers=0)
@@ -316,25 +316,25 @@ class WorkerTest(unittest.TestCase):
value_before = i * 10**6
objectid = ray.put(value_before)
value_after = ray.get(objectid)
self.assertEqual(value_before, value_after)
assert value_before == value_after
for i in range(100):
value_before = i * 10**6 * 1.0
objectid = ray.put(value_before)
value_after = ray.get(objectid)
self.assertEqual(value_before, value_after)
assert value_before == value_after
for i in range(100):
value_before = "h" * i
objectid = ray.put(value_before)
value_after = ray.get(objectid)
self.assertEqual(value_before, value_after)
assert value_before == value_after
for i in range(100):
value_before = [1] * i
objectid = ray.put(value_before)
value_after = ray.get(objectid)
self.assertEqual(value_before, value_after)
assert value_before == value_after
class APITest(unittest.TestCase):
@@ -364,8 +364,8 @@ class APITest(unittest.TestCase):
serializer=custom_serializer,
deserializer=custom_deserializer)
self.assertEqual(
ray.get(ray.put(Foo())), ((3, "string1", Foo.__name__), "string2"))
assert ray.get(ray.put(Foo())) == ((3, "string1", Foo.__name__),
"string2")
class Bar(object):
def __init__(self):
@@ -380,8 +380,7 @@ class APITest(unittest.TestCase):
def f():
return Bar()
self.assertEqual(
ray.get(f.remote()), ((3, "string1", Bar.__name__), "string2"))
assert ray.get(f.remote()) == ((3, "string1", Bar.__name__), "string2")
def testRegisterClass(self):
self.init_ray(num_workers=2)
@@ -396,15 +395,15 @@ class APITest(unittest.TestCase):
# Test subtypes of dictionaries.
value_before = OrderedDict([("hello", 1), ("world", 2)])
object_id = ray.put(value_before)
self.assertEqual(value_before, ray.get(object_id))
assert value_before == ray.get(object_id)
value_before = defaultdict(lambda: 0, [("hello", 1), ("world", 2)])
object_id = ray.put(value_before)
self.assertEqual(value_before, ray.get(object_id))
assert value_before == ray.get(object_id)
value_before = defaultdict(lambda: [], [("hello", 1), ("world", 2)])
object_id = ray.put(value_before)
self.assertEqual(value_before, ray.get(object_id))
assert value_before == ray.get(object_id)
# Test passing custom classes into remote functions from the driver.
@ray.remote
@@ -412,7 +411,7 @@ class APITest(unittest.TestCase):
return x
foo = ray.get(f.remote(Foo(7)))
self.assertEqual(foo, Foo(7))
assert foo == Foo(7)
regex = re.compile(r"\d+\.\d*")
new_regex = ray.get(f.remote(regex))
@@ -420,7 +419,7 @@ class APITest(unittest.TestCase):
# Ubuntu, so it is commented out for now:
# self.assertEqual(regex, new_regex)
# Instead, we do this:
self.assertEqual(regex.pattern, new_regex.pattern)
assert regex.pattern == new_regex.pattern
# Test returning custom classes created on workers.
@ray.remote
@@ -428,7 +427,7 @@ class APITest(unittest.TestCase):
return SubQux(), Qux()
subqux, qux = ray.get(g.remote())
self.assertEqual(subqux.objs[2].foo.value, 0)
assert subqux.objs[2].foo.value == 0
# Test exporting custom class definitions from one worker to another
# when the worker is blocked in a get.
@@ -444,7 +443,7 @@ class APITest(unittest.TestCase):
def h2(x):
return ray.get(h1.remote(x))
self.assertEqual(ray.get(h2.remote(10)).value, 10)
assert ray.get(h2.remote(10)).value == 10
# Test registering multiple classes with the same name.
@ray.remote(num_return_vals=3)
@@ -479,12 +478,12 @@ class APITest(unittest.TestCase):
c1.method1()
c2.method2()
self.assertFalse(hasattr(c0, "method1"))
self.assertFalse(hasattr(c0, "method2"))
self.assertFalse(hasattr(c1, "method0"))
self.assertFalse(hasattr(c1, "method2"))
self.assertFalse(hasattr(c2, "method0"))
self.assertFalse(hasattr(c2, "method1"))
assert not hasattr(c0, "method1")
assert not hasattr(c0, "method2")
assert not hasattr(c1, "method0")
assert not hasattr(c1, "method2")
assert not hasattr(c2, "method0")
assert not hasattr(c2, "method1")
@ray.remote
def k():
@@ -514,12 +513,12 @@ class APITest(unittest.TestCase):
c1.method1()
c2.method2()
self.assertFalse(hasattr(c0, "method1"))
self.assertFalse(hasattr(c0, "method2"))
self.assertFalse(hasattr(c1, "method0"))
self.assertFalse(hasattr(c1, "method2"))
self.assertFalse(hasattr(c2, "method0"))
self.assertFalse(hasattr(c2, "method1"))
assert not hasattr(c0, "method1")
assert not hasattr(c0, "method2")
assert not hasattr(c1, "method0")
assert not hasattr(c1, "method2")
assert not hasattr(c2, "method0")
assert not hasattr(c2, "method1")
def testKeywordArgs(self):
@ray.remote
@@ -537,43 +536,43 @@ class APITest(unittest.TestCase):
self.init_ray()
x = keyword_fct1.remote(1)
self.assertEqual(ray.get(x), "1 hello")
assert ray.get(x) == "1 hello"
x = keyword_fct1.remote(1, "hi")
self.assertEqual(ray.get(x), "1 hi")
assert ray.get(x) == "1 hi"
x = keyword_fct1.remote(1, b="world")
self.assertEqual(ray.get(x), "1 world")
assert ray.get(x) == "1 world"
x = keyword_fct1.remote(a=1, b="world")
self.assertEqual(ray.get(x), "1 world")
assert ray.get(x) == "1 world"
x = keyword_fct2.remote(a="w", b="hi")
self.assertEqual(ray.get(x), "w hi")
assert ray.get(x) == "w hi"
x = keyword_fct2.remote(b="hi", a="w")
self.assertEqual(ray.get(x), "w hi")
assert ray.get(x) == "w hi"
x = keyword_fct2.remote(a="w")
self.assertEqual(ray.get(x), "w world")
assert ray.get(x) == "w world"
x = keyword_fct2.remote(b="hi")
self.assertEqual(ray.get(x), "hello hi")
assert ray.get(x) == "hello hi"
x = keyword_fct2.remote("w")
self.assertEqual(ray.get(x), "w world")
assert ray.get(x) == "w world"
x = keyword_fct2.remote("w", "hi")
self.assertEqual(ray.get(x), "w hi")
assert ray.get(x) == "w hi"
x = keyword_fct3.remote(0, 1, c="w", d="hi")
self.assertEqual(ray.get(x), "0 1 w hi")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(0, b=1, c="w", d="hi")
self.assertEqual(ray.get(x), "0 1 w hi")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(a=0, b=1, c="w", d="hi")
self.assertEqual(ray.get(x), "0 1 w hi")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(0, 1, d="hi", c="w")
self.assertEqual(ray.get(x), "0 1 w hi")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(0, 1, c="w")
self.assertEqual(ray.get(x), "0 1 w world")
assert ray.get(x) == "0 1 w world"
x = keyword_fct3.remote(0, 1, d="hi")
self.assertEqual(ray.get(x), "0 1 hello hi")
assert ray.get(x) == "0 1 hello hi"
x = keyword_fct3.remote(0, 1)
self.assertEqual(ray.get(x), "0 1 hello world")
assert ray.get(x) == "0 1 hello world"
x = keyword_fct3.remote(a=0, b=1)
self.assertEqual(ray.get(x), "0 1 hello world")
assert ray.get(x) == "0 1 hello world"
# Check that we cannot pass invalid keyword arguments to functions.
@ray.remote
@@ -585,27 +584,27 @@ class APITest(unittest.TestCase):
return
# Make sure we get an exception if too many arguments are passed in.
with self.assertRaises(Exception):
with pytest.raises(Exception):
f1.remote(3)
with self.assertRaises(Exception):
with pytest.raises(Exception):
f1.remote(x=3)
with self.assertRaises(Exception):
with pytest.raises(Exception):
f2.remote(0, w=0)
with self.assertRaises(Exception):
with pytest.raises(Exception):
f2.remote(3, x=3)
# Make sure we get an exception if too many arguments are passed in.
with self.assertRaises(Exception):
with pytest.raises(Exception):
f2.remote(1, 2, 3, 4)
@ray.remote
def f3(x):
return x
self.assertEqual(ray.get(f3.remote(4)), 4)
assert ray.get(f3.remote(4)) == 4
def testVariableNumberOfArgs(self):
@ray.remote
@@ -629,11 +628,11 @@ class APITest(unittest.TestCase):
self.init_ray()
x = varargs_fct1.remote(0, 1, 2)
self.assertEqual(ray.get(x), "0 1 2")
assert ray.get(x) == "0 1 2"
x = varargs_fct2.remote(0, 1, 2)
self.assertEqual(ray.get(x), "1 2")
assert ray.get(x) == "1 2"
self.assertTrue(kwargs_exception_thrown)
assert kwargs_exception_thrown
@ray.remote
def f1(*args):
@@ -643,16 +642,16 @@ class APITest(unittest.TestCase):
def f2(x, y, *args):
return x, y, args
self.assertEqual(ray.get(f1.remote()), ())
self.assertEqual(ray.get(f1.remote(1)), (1, ))
self.assertEqual(ray.get(f1.remote(1, 2, 3)), (1, 2, 3))
with self.assertRaises(Exception):
assert ray.get(f1.remote()) == ()
assert ray.get(f1.remote(1)) == (1, )
assert ray.get(f1.remote(1, 2, 3)) == (1, 2, 3)
with pytest.raises(Exception):
f2.remote()
with self.assertRaises(Exception):
with pytest.raises(Exception):
f2.remote(1)
self.assertEqual(ray.get(f2.remote(1, 2)), (1, 2, ()))
self.assertEqual(ray.get(f2.remote(1, 2, 3)), (1, 2, (3, )))
self.assertEqual(ray.get(f2.remote(1, 2, 3, 4)), (1, 2, (3, 4)))
assert ray.get(f2.remote(1, 2)) == (1, 2, ())
assert ray.get(f2.remote(1, 2, 3)) == (1, 2, (3, ))
assert ray.get(f2.remote(1, 2, 3, 4)) == (1, 2, (3, 4))
def testNoArgs(self):
@ray.remote
@@ -671,7 +670,7 @@ class APITest(unittest.TestCase):
def f(x):
return x + 1
self.assertEqual(ray.get(f.remote(0)), 1)
assert ray.get(f.remote(0)) == 1
# Test that we can redefine the remote function.
@ray.remote
@@ -680,7 +679,7 @@ class APITest(unittest.TestCase):
while True:
val = ray.get(f.remote(0))
self.assertTrue(val in [1, 10])
assert val in [1, 10]
if val == 10:
break
else:
@@ -726,9 +725,9 @@ class APITest(unittest.TestCase):
def m(x):
return ray.get(k2.remote(x))
self.assertEqual(ray.get(k.remote(1)), 2)
self.assertEqual(ray.get(k2.remote(1)), 2)
self.assertEqual(ray.get(m.remote(1)), 2)
assert ray.get(k.remote(1)) == 2
assert ray.get(k2.remote(1)) == 2
assert ray.get(m.remote(1)) == 2
def testSubmitAPI(self):
self.init_ray(num_gpus=1, resources={"Custom": 1}, num_workers=1)
@@ -779,25 +778,23 @@ class APITest(unittest.TestCase):
def testGetMultiple(self):
self.init_ray()
object_ids = [ray.put(i) for i in range(10)]
self.assertEqual(ray.get(object_ids), list(range(10)))
assert ray.get(object_ids) == list(range(10))
# Get a random choice of object IDs with duplicates.
indices = list(np.random.choice(range(10), 5))
indices += indices
results = ray.get([object_ids[i] for i in indices])
self.assertEqual(results, indices)
assert results == indices
def testGetMultipleExperimental(self):
self.init_ray()
object_ids = [ray.put(i) for i in range(10)]
object_ids_tuple = tuple(object_ids)
self.assertEqual(
ray.experimental.get(object_ids_tuple), list(range(10)))
assert ray.experimental.get(object_ids_tuple) == list(range(10))
object_ids_nparray = np.array(object_ids)
self.assertEqual(
ray.experimental.get(object_ids_nparray), list(range(10)))
assert ray.experimental.get(object_ids_nparray) == list(range(10))
def testGetDict(self):
self.init_ray()
@@ -806,7 +803,7 @@ class APITest(unittest.TestCase):
d[str(i)] = i
result = ray.experimental.get(d)
expected = {str(i): i for i in range(10)}
self.assertEqual(result, expected)
assert result == expected
def testWait(self):
self.init_ray(num_cpus=1)
@@ -823,11 +820,11 @@ class APITest(unittest.TestCase):
f.remote(0.5)
]
ready_ids, remaining_ids = ray.wait(objectids)
self.assertEqual(len(ready_ids), 1)
self.assertEqual(len(remaining_ids), 3)
assert len(ready_ids) == 1
assert len(remaining_ids) == 3
ready_ids, remaining_ids = ray.wait(objectids, num_returns=4)
self.assertEqual(set(ready_ids), set(objectids))
self.assertEqual(remaining_ids, [])
assert set(ready_ids) == set(objectids)
assert remaining_ids == []
objectids = [
f.remote(0.5),
@@ -838,9 +835,9 @@ class APITest(unittest.TestCase):
start_time = time.time()
ready_ids, remaining_ids = ray.wait(
objectids, timeout=1750, num_returns=4)
self.assertLess(time.time() - start_time, 2)
self.assertEqual(len(ready_ids), 3)
self.assertEqual(len(remaining_ids), 1)
assert time.time() - start_time < 2
assert len(ready_ids) == 3
assert len(remaining_ids) == 1
ray.wait(objectids)
objectids = [
f.remote(1.0),
@@ -850,33 +847,34 @@ class APITest(unittest.TestCase):
]
start_time = time.time()
ready_ids, remaining_ids = ray.wait(objectids, timeout=5000)
self.assertTrue(time.time() - start_time < 5)
self.assertEqual(len(ready_ids), 1)
self.assertEqual(len(remaining_ids), 3)
assert time.time() - start_time < 5
assert len(ready_ids) == 1
assert len(remaining_ids) == 3
# Verify that calling wait with duplicate object IDs throws an
# exception.
x = ray.put(1)
self.assertRaises(Exception, lambda: ray.wait([x, x]))
with pytest.raises(Exception):
ray.wait([x, x])
# Make sure it is possible to call wait with an empty list.
ready_ids, remaining_ids = ray.wait([])
self.assertEqual(ready_ids, [])
self.assertEqual(remaining_ids, [])
assert ready_ids == []
assert remaining_ids == []
# Test semantics of num_returns with no timeout.
oids = [ray.put(i) for i in range(10)]
(found, rest) = ray.wait(oids, num_returns=2)
self.assertEqual(len(found), 2)
self.assertEqual(len(rest), 8)
assert len(found) == 2
assert len(rest) == 8
# Verify that incorrect usage raises a TypeError.
x = ray.put(1)
with self.assertRaises(TypeError):
with pytest.raises(TypeError):
ray.wait(x)
with self.assertRaises(TypeError):
with pytest.raises(TypeError):
ray.wait(1)
with self.assertRaises(TypeError):
with pytest.raises(TypeError):
ray.wait([1])
def testWaitIterables(self):
@@ -890,8 +888,8 @@ class APITest(unittest.TestCase):
objectids = (f.remote(1.0), f.remote(0.5), f.remote(0.5),
f.remote(0.5))
ready_ids, remaining_ids = ray.experimental.wait(objectids)
self.assertEqual(len(ready_ids), 1)
self.assertEqual(len(remaining_ids), 3)
assert len(ready_ids) == 1
assert len(remaining_ids) == 3
objectids = np.array(
[f.remote(1.0),
@@ -899,8 +897,8 @@ class APITest(unittest.TestCase):
f.remote(0.5),
f.remote(0.5)])
ready_ids, remaining_ids = ray.experimental.wait(objectids)
self.assertEqual(len(ready_ids), 1)
self.assertEqual(len(remaining_ids), 3)
assert len(ready_ids) == 1
assert len(remaining_ids) == 3
def testMultipleWaitsAndGets(self):
# It is important to use three workers here, so that the three tasks
@@ -964,8 +962,8 @@ class APITest(unittest.TestCase):
res1 = get_state.remote()
res2 = get_state.remote()
self.assertEqual(ray.get(res1), (1, 2, 3, 4))
self.assertEqual(ray.get(res2), (1, 2, 3, 4))
assert ray.get(res1) == (1, 2, 3, 4)
assert ray.get(res2) == (1, 2, 3, 4)
# Clean up the path on the workers.
def f(worker_info):
@@ -988,7 +986,7 @@ class APITest(unittest.TestCase):
def get_path1():
return sys.path
self.assertEqual("fake_directory", ray.get(get_path1.remote())[-1])
assert "fake_directory" == ray.get(get_path1.remote())[-1]
def f(worker_info):
sys.path.pop(-1)
@@ -1002,7 +1000,7 @@ class APITest(unittest.TestCase):
def get_path2():
return sys.path
self.assertTrue("fake_directory" not in ray.get(get_path2.remote()))
assert "fake_directory" not in ray.get(get_path2.remote())
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
@@ -1038,7 +1036,7 @@ class APITest(unittest.TestCase):
# Wait for the events to appear in the event log.
wait_for_num_events(1)
self.assertEqual(len(events()), 1)
assert len(events()) == 1
@ray.remote
def test_log_span_exception():
@@ -1049,7 +1047,7 @@ class APITest(unittest.TestCase):
test_log_span_exception.remote()
# Wait for the events to appear in the event log.
wait_for_num_events(2)
self.assertEqual(len(events()), 2)
assert len(events()) == 2
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") != "1",
@@ -1136,11 +1134,11 @@ class APITest(unittest.TestCase):
results5 = [f.remote() for _ in range(num_calls)]
self.assertEqual(ray.get(results1), num_calls * [1])
self.assertEqual(ray.get(results2), num_calls * [2])
self.assertEqual(ray.get(results3), num_calls * [3])
self.assertEqual(ray.get(results4), num_calls * [4])
self.assertEqual(ray.get(results5), num_calls * [5])
assert ray.get(results1) == num_calls * [1]
assert ray.get(results2) == num_calls * [2]
assert ray.get(results3) == num_calls * [3]
assert ray.get(results4) == num_calls * [4]
assert ray.get(results5) == num_calls * [5]
@ray.remote
def g():
@@ -1163,17 +1161,17 @@ class APITest(unittest.TestCase):
return 5
result_values = ray.get([g.remote() for _ in range(num_calls)])
self.assertEqual(result_values, num_calls * [5])
assert result_values == num_calls * [5]
def testIllegalAPICalls(self):
self.init_ray()
# Verify that we cannot call put on an ObjectID.
x = ray.put(1)
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.put(x)
# Verify that we cannot call get on a regular value.
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.get(3)
def testMultithreading(self):
@@ -1345,14 +1343,14 @@ class ResourcesTest(unittest.TestCase):
start_time = time.time()
ray.get([f.remote(0.5) for _ in range(10)])
duration = time.time() - start_time
self.assertLess(duration, 0.5 + time_buffer)
self.assertGreater(duration, 0.5)
assert duration < 0.5 + time_buffer
assert duration > 0.5
start_time = time.time()
ray.get([f.remote(0.5) for _ in range(11)])
duration = time.time() - start_time
self.assertLess(duration, 1 + time_buffer)
self.assertGreater(duration, 1)
assert duration < 1 + time_buffer
assert duration > 1
@ray.remote(num_cpus=3)
def f(n):
@@ -1361,14 +1359,14 @@ class ResourcesTest(unittest.TestCase):
start_time = time.time()
ray.get([f.remote(0.5) for _ in range(3)])
duration = time.time() - start_time
self.assertLess(duration, 0.5 + time_buffer)
self.assertGreater(duration, 0.5)
assert duration < 0.5 + time_buffer
assert duration > 0.5
start_time = time.time()
ray.get([f.remote(0.5) for _ in range(4)])
duration = time.time() - start_time
self.assertLess(duration, 1 + time_buffer)
self.assertGreater(duration, 1)
assert duration < 1 + time_buffer
assert duration > 1
@ray.remote(num_gpus=1)
def f(n):
@@ -1377,20 +1375,20 @@ class ResourcesTest(unittest.TestCase):
start_time = time.time()
ray.get([f.remote(0.5) for _ in range(2)])
duration = time.time() - start_time
self.assertLess(duration, 0.5 + time_buffer)
self.assertGreater(duration, 0.5)
assert duration < 0.5 + time_buffer
assert duration > 0.5
start_time = time.time()
ray.get([f.remote(0.5) for _ in range(3)])
duration = time.time() - start_time
self.assertLess(duration, 1 + time_buffer)
self.assertGreater(duration, 1)
assert duration < 1 + time_buffer
assert duration > 1
start_time = time.time()
ray.get([f.remote(0.5) for _ in range(4)])
duration = time.time() - start_time
self.assertLess(duration, 1 + time_buffer)
self.assertGreater(duration, 1)
assert duration < 1 + time_buffer
assert duration > 1
def testMultiResourceConstraints(self):
num_workers = 20
@@ -1423,26 +1421,26 @@ class ResourcesTest(unittest.TestCase):
start_time = time.time()
ray.get([f.remote(0.5), g.remote(0.5)])
duration = time.time() - start_time
self.assertLess(duration, 0.5 + time_buffer)
self.assertGreater(duration, 0.5)
assert duration < 0.5 + time_buffer
assert duration > 0.5
start_time = time.time()
ray.get([f.remote(0.5), f.remote(0.5)])
duration = time.time() - start_time
self.assertLess(duration, 1 + time_buffer)
self.assertGreater(duration, 1)
assert duration < 1 + time_buffer
assert duration > 1
start_time = time.time()
ray.get([g.remote(0.5), g.remote(0.5)])
duration = time.time() - start_time
self.assertLess(duration, 1 + time_buffer)
self.assertGreater(duration, 1)
assert duration < 1 + time_buffer
assert duration > 1
start_time = time.time()
ray.get([f.remote(0.5), f.remote(0.5), g.remote(0.5), g.remote(0.5)])
duration = time.time() - start_time
self.assertLess(duration, 1 + time_buffer)
self.assertGreater(duration, 1)
assert duration < 1 + time_buffer
assert duration > 1
def testGPUIDs(self):
num_gpus = 10
@@ -1529,15 +1527,15 @@ class ResourcesTest(unittest.TestCase):
"up.")
list_of_ids = ray.get([f0.remote() for _ in range(10)])
self.assertEqual(list_of_ids, 10 * [[]])
assert list_of_ids == 10 * [[]]
list_of_ids = ray.get([f1.remote() for _ in range(10)])
set_of_ids = {tuple(gpu_ids) for gpu_ids in list_of_ids}
self.assertEqual(set_of_ids, {(i, ) for i in range(10)})
assert set_of_ids == {(i, ) for i in range(10)}
list_of_ids = ray.get([f2.remote(), f4.remote(), f4.remote()])
all_ids = [gpu_id for gpu_ids in list_of_ids for gpu_id in gpu_ids]
self.assertEqual(set(all_ids), set(range(10)))
assert set(all_ids) == set(range(10))
remaining = [f5.remote() for _ in range(20)]
for _ in range(10):
@@ -1548,7 +1546,7 @@ class ResourcesTest(unittest.TestCase):
# should only be 2 tasks scheduled at a given time, so if we wait
# for 2 tasks to finish, then it should take at least 0.1 seconds
# for each pair of tasks to finish.
self.assertGreater(t2 - t1, 0.09)
assert t2 - t1 > 0.09
list_of_ids = ray.get(ready)
all_ids = [gpu_id for gpu_ids in list_of_ids for gpu_id in gpu_ids]
# Commenting out the below assert because it seems to fail a lot.
@@ -1615,9 +1613,9 @@ class ResourcesTest(unittest.TestCase):
return ray.worker.global_worker.plasma_client.store_socket_name
# Make sure tasks and actors run on the remote local scheduler.
self.assertNotEqual(ray.get(f.remote()), local_plasma)
assert ray.get(f.remote()) != local_plasma
a = Foo.remote()
self.assertNotEqual(ray.get(a.method.remote()), local_plasma)
assert ray.get(a.method.remote()) != local_plasma
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") != "1",
@@ -1666,10 +1664,10 @@ class ResourcesTest(unittest.TestCase):
def test():
pass
with self.assertRaises(ValueError):
with pytest.raises(ValueError):
test.remote()
with self.assertRaises(ValueError):
with pytest.raises(ValueError):
Foo2._submit([], {}, resources={"Custom": 1.5})
def testMultipleLocalSchedulers(self):
@@ -1751,22 +1749,22 @@ class ResourcesTest(unittest.TestCase):
def validate_names_and_results(names, results):
for name, result in zip(names, ray.get(results)):
if name == "run_on_0":
self.assertIn(result, [store_names[0]])
assert result in [store_names[0]]
elif name == "run_on_1":
self.assertIn(result, [store_names[1]])
assert result in [store_names[1]]
elif name == "run_on_2":
self.assertIn(result, [store_names[2]])
assert result in [store_names[2]]
elif name == "run_on_0_1_2":
self.assertIn(
result,
[store_names[0], store_names[1], store_names[2]])
assert (result in [
store_names[0], store_names[1], store_names[2]
])
elif name == "run_on_1_2":
self.assertIn(result, [store_names[1], store_names[2]])
assert result in [store_names[1], store_names[2]]
elif name == "run_on_0_2":
self.assertIn(result, [store_names[0], store_names[2]])
assert result in [store_names[0], store_names[2]]
else:
raise Exception("This should be unreachable.")
self.assertEqual(set(ray.get(results)), set(store_names))
assert set(ray.get(results)) == set(store_names)
names, results = run_lots_of_tasks()
validate_names_and_results(names, results)
@@ -1813,14 +1811,14 @@ class ResourcesTest(unittest.TestCase):
return ray.worker.global_worker.plasma_client.store_socket_name
# The f tasks should be scheduled on both local schedulers.
self.assertEqual(len(set(ray.get([f.remote() for _ in range(50)]))), 2)
assert len(set(ray.get([f.remote() for _ in range(50)]))) == 2
local_plasma = ray.worker.global_worker.plasma_client.store_socket_name
# The g tasks should be scheduled only on the second local scheduler.
local_scheduler_ids = set(ray.get([g.remote() for _ in range(50)]))
self.assertEqual(len(local_scheduler_ids), 1)
self.assertNotEqual(list(local_scheduler_ids)[0], local_plasma)
assert len(local_scheduler_ids) == 1
assert list(local_scheduler_ids)[0] != local_plasma
# Make sure that resource bookkeeping works when a task that uses a
# custom resources gets blocked.
@@ -1865,21 +1863,21 @@ class ResourcesTest(unittest.TestCase):
return ray.worker.global_worker.plasma_client.store_socket_name
# The f and g tasks should be scheduled on both local schedulers.
self.assertEqual(len(set(ray.get([f.remote() for _ in range(50)]))), 2)
self.assertEqual(len(set(ray.get([g.remote() for _ in range(50)]))), 2)
assert len(set(ray.get([f.remote() for _ in range(50)]))) == 2
assert len(set(ray.get([g.remote() for _ in range(50)]))) == 2
local_plasma = ray.worker.global_worker.plasma_client.store_socket_name
# The h tasks should be scheduled only on the second local scheduler.
local_scheduler_ids = set(ray.get([h.remote() for _ in range(50)]))
self.assertEqual(len(local_scheduler_ids), 1)
self.assertNotEqual(list(local_scheduler_ids)[0], local_plasma)
assert len(local_scheduler_ids) == 1
assert list(local_scheduler_ids)[0] != local_plasma
# Make sure that tasks with unsatisfied custom resource requirements do
# not get scheduled.
ready_ids, remaining_ids = ray.wait(
[j.remote(), k.remote()], timeout=500)
self.assertEqual(ready_ids, [])
assert ready_ids == []
def testManyCustomResources(self):
num_custom_resources = 10000
@@ -2015,7 +2013,7 @@ class WorkerPoolTests(unittest.TestCase):
pid1 = ray.get(f.remote())
pid2 = ray.get(f.remote())
self.assertEqual(pid1, pid2)
assert pid1 == pid2
ray.test.test_utils.wait_for_pid_to_exit(pid1)
@@ -2041,7 +2039,7 @@ class SchedulingAlgorithm(unittest.TestCase):
and all(count >= minimum_count for count in counts)):
break
attempts += 1
self.assertLess(attempts, num_attempts)
assert attempts < num_attempts
def testLoadBalancing(self):
# This test ensures that tasks are being assigned to all local
@@ -2110,19 +2108,19 @@ class GlobalStateAPI(unittest.TestCase):
ray.shutdown()
def testGlobalStateAPI(self):
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.global_state.object_table()
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.global_state.task_table()
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.global_state.client_table()
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.global_state.function_table()
with self.assertRaises(Exception):
with pytest.raises(Exception):
ray.global_state.log_files()
ray.init(num_cpus=5, num_gpus=3, resources={"CustomResource": 1})
@@ -2130,7 +2128,7 @@ class GlobalStateAPI(unittest.TestCase):
resources = {"CPU": 5, "GPU": 3, "CustomResource": 1}
assert ray.global_state.cluster_resources() == resources
self.assertEqual(ray.global_state.object_table(), {})
assert ray.global_state.object_table() == {}
driver_id = ray.experimental.state.binary_to_hex(
ray.worker.global_worker.worker_id)
@@ -2140,52 +2138,29 @@ class GlobalStateAPI(unittest.TestCase):
# One task is put in the task table which corresponds to this driver.
wait_for_num_tasks(1)
task_table = ray.global_state.task_table()
self.assertEqual(len(task_table), 1)
self.assertEqual(driver_task_id, list(task_table.keys())[0])
assert len(task_table) == 1
assert driver_task_id == list(task_table.keys())[0]
if not ray.worker.global_worker.use_raylet:
self.assertEqual(task_table[driver_task_id]["State"],
ray.experimental.state.TASK_STATUS_RUNNING)
assert (task_table[driver_task_id]["State"] ==
ray.experimental.state.TASK_STATUS_RUNNING)
if not ray.worker.global_worker.use_raylet:
self.assertEqual(task_table[driver_task_id]["TaskSpec"]["TaskID"],
driver_task_id)
self.assertEqual(task_table[driver_task_id]["TaskSpec"]["ActorID"],
ray_constants.ID_SIZE * "ff")
self.assertEqual(task_table[driver_task_id]["TaskSpec"]["Args"],
[])
self.assertEqual(
task_table[driver_task_id]["TaskSpec"]["DriverID"], driver_id)
self.assertEqual(
task_table[driver_task_id]["TaskSpec"]["FunctionID"],
ray_constants.ID_SIZE * "ff")
self.assertEqual(
(task_table[driver_task_id]["TaskSpec"]["ReturnObjectIDs"]),
[])
task_spec = task_table[driver_task_id]["TaskSpec"]
else:
self.assertEqual(len(task_table[driver_task_id]), 1)
self.assertEqual(
task_table[driver_task_id][0]["TaskSpec"]["TaskID"],
driver_task_id)
self.assertEqual(
task_table[driver_task_id][0]["TaskSpec"]["ActorID"],
ray_constants.ID_SIZE * "ff")
self.assertEqual(task_table[driver_task_id][0]["TaskSpec"]["Args"],
[])
self.assertEqual(
task_table[driver_task_id][0]["TaskSpec"]["DriverID"],
driver_id)
self.assertEqual(
task_table[driver_task_id][0]["TaskSpec"]["FunctionID"],
ray_constants.ID_SIZE * "ff")
self.assertEqual(
(task_table[driver_task_id][0]["TaskSpec"]["ReturnObjectIDs"]),
[])
assert len(task_table[driver_task_id]) == 1
task_spec = task_table[driver_task_id][0]["TaskSpec"]
assert task_spec["TaskID"] == driver_task_id
assert task_spec["ActorID"] == ray_constants.ID_SIZE * "ff"
assert task_spec["Args"] == []
assert task_spec["DriverID"] == driver_id
assert task_spec["FunctionID"] == ray_constants.ID_SIZE * "ff"
assert task_spec["ReturnObjectIDs"] == []
client_table = ray.global_state.client_table()
node_ip_address = ray.worker.global_worker.node_ip_address
if not ray.worker.global_worker.use_raylet:
self.assertEqual(len(client_table[node_ip_address]), 3)
assert len(client_table[node_ip_address]) == 3
manager_client = [
c for c in client_table[node_ip_address]
if c["ClientType"] == "plasma_manager"
@@ -2206,7 +2181,7 @@ class GlobalStateAPI(unittest.TestCase):
while time.time() - start_time < 10:
wait_for_num_tasks(1 + 1)
task_table = ray.global_state.task_table()
self.assertEqual(len(task_table), 1 + 1)
assert len(task_table) == 1 + 1
task_id_set = set(task_table.keys())
task_id_set.remove(driver_task_id)
task_id = list(task_id_set)[0]
@@ -2221,17 +2196,16 @@ class GlobalStateAPI(unittest.TestCase):
task_spec = task_table[task_id]["TaskSpec"]
else:
task_spec = task_table[task_id][0]["TaskSpec"]
self.assertEqual(task_spec["ActorID"], ray_constants.ID_SIZE * "ff")
self.assertEqual(task_spec["Args"], [1, "hi", x_id])
self.assertEqual(task_spec["DriverID"], driver_id)
self.assertEqual(task_spec["ReturnObjectIDs"], [result_id])
assert task_spec["ActorID"] == ray_constants.ID_SIZE * "ff"
assert task_spec["Args"] == [1, "hi", x_id]
assert task_spec["DriverID"] == driver_id
assert task_spec["ReturnObjectIDs"] == [result_id]
function_table_entry = function_table[task_spec["FunctionID"]]
self.assertEqual(function_table_entry["Name"], "runtest.f")
self.assertEqual(function_table_entry["DriverID"], driver_id)
self.assertEqual(function_table_entry["Module"], "runtest")
assert function_table_entry["Name"] == "runtest.f"
assert function_table_entry["DriverID"] == driver_id
assert function_table_entry["Module"] == "runtest"
self.assertEqual(task_table[task_id],
ray.global_state.task_table(task_id))
assert task_table[task_id] == ray.global_state.task_table(task_id)
# Wait for two objects, one for the x_id and one for result_id.
wait_for_num_objects(2)
@@ -2255,32 +2229,30 @@ class GlobalStateAPI(unittest.TestCase):
wait_for_object_table()
object_table = ray.global_state.object_table()
self.assertEqual(len(object_table), 2)
assert len(object_table) == 2
if not ray.worker.global_worker.use_raylet:
self.assertEqual(object_table[x_id]["IsPut"], True)
self.assertEqual(object_table[x_id]["TaskID"], driver_task_id)
self.assertEqual(object_table[x_id]["ManagerIDs"],
[manager_client["DBClientID"]])
db_client_id = manager_client["DBClientID"]
assert object_table[x_id]["IsPut"] is True
assert object_table[x_id]["TaskID"] == driver_task_id
assert object_table[x_id]["ManagerIDs"] == [db_client_id]
self.assertEqual(object_table[result_id]["IsPut"], False)
self.assertEqual(object_table[result_id]["TaskID"], task_id)
self.assertEqual(object_table[result_id]["ManagerIDs"],
[manager_client["DBClientID"]])
assert object_table[result_id]["IsPut"] is False
assert object_table[result_id]["TaskID"] == task_id
assert object_table[result_id]["ManagerIDs"] == [db_client_id]
else:
assert len(object_table[x_id]) == 1
self.assertEqual(object_table[x_id][0]["IsEviction"], False)
self.assertEqual(object_table[x_id][0]["NumEvictions"], 0)
assert object_table[x_id][0]["IsEviction"] is False
assert object_table[x_id][0]["NumEvictions"] == 0
assert len(object_table[result_id]) == 1
self.assertEqual(object_table[result_id][0]["IsEviction"], False)
self.assertEqual(object_table[result_id][0]["NumEvictions"], 0)
assert object_table[result_id][0]["IsEviction"] is False
assert object_table[result_id][0]["NumEvictions"] == 0
self.assertEqual(object_table[x_id],
ray.global_state.object_table(x_id))
self.assertEqual(object_table[result_id],
ray.global_state.object_table(result_id))
assert object_table[x_id] == ray.global_state.object_table(x_id)
object_table_entry = ray.global_state.object_table(result_id)
assert object_table[result_id] == object_table_entry
def testLogFileAPI(self):
ray.init(redirect_worker_output=True)
@@ -2310,7 +2282,7 @@ class GlobalStateAPI(unittest.TestCase):
break
time.sleep(0.1)
self.assertEqual(found_message, True)
assert found_message is True
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
@@ -2335,17 +2307,17 @@ class GlobalStateAPI(unittest.TestCase):
if len(profiles) == num_calls and len(limited_profiles) == 1:
break
time.sleep(0.1)
self.assertEqual(len(profiles), num_calls)
self.assertEqual(len(limited_profiles), 1)
assert len(profiles) == num_calls
assert len(limited_profiles) == 1
# Make sure that each entry is properly formatted.
for task_id, data in profiles.items():
self.assertIn("execute_start", data)
self.assertIn("execute_end", data)
self.assertIn("get_arguments_start", data)
self.assertIn("get_arguments_end", data)
self.assertIn("store_outputs_start", data)
self.assertIn("store_outputs_end", data)
assert "execute_start" in data
assert "execute_end" in data
assert "get_arguments_start" in data
assert "get_arguments_end" in data
assert "store_outputs_start" in data
assert "store_outputs_end" in data
def testWorkers(self):
num_workers = 3
@@ -2366,12 +2338,12 @@ class GlobalStateAPI(unittest.TestCase):
worker_info = ray.global_state.workers()
assert len(worker_info) >= num_workers
for worker_id, info in worker_info.items():
self.assertIn("node_ip_address", info)
self.assertIn("local_scheduler_socket", info)
self.assertIn("plasma_manager_socket", info)
self.assertIn("plasma_store_socket", info)
self.assertIn("stderr_file", info)
self.assertIn("stdout_file", info)
assert "node_ip_address" in info
assert "local_scheduler_socket" in info
assert "plasma_manager_socket" in info
assert "plasma_store_socket" in info
assert "stderr_file" in info
assert "stdout_file" in info
@unittest.skipIf(
os.environ.get("RAY_USE_XRAY") == "1",
test/tensorflow_test.py
+16 -16
View File
@@ -111,7 +111,7 @@ class TensorFlowTest(unittest.TestCase):
weights[name] += 1.0
variables.set_weights(weights)
self.assertEqual(weights, variables.get_weights())
assert weights == variables.get_weights()
loss2, init2, _, _ = make_linear_network("w", "b")
sess.run(init2)
@@ -123,16 +123,16 @@ class TensorFlowTest(unittest.TestCase):
weights2[name] += 2.0
variables2.set_weights(weights2)
self.assertEqual(weights2, variables2.get_weights())
assert weights2 == variables2.get_weights()
flat_weights = variables2.get_flat() + 2.0
variables2.set_flat(flat_weights)
assert_almost_equal(flat_weights, variables2.get_flat())
variables3 = ray.experimental.TensorFlowVariables([loss2])
self.assertEqual(variables3.sess, None)
assert variables3.sess is None
sess = tf.Session()
variables3.set_session(sess)
self.assertEqual(variables3.sess, sess)
assert variables3.sess == sess
# Test that the variable names for the two different nets are not
# modified by TensorFlow to be unique (i.e., they should already
@@ -153,8 +153,8 @@ class TensorFlowTest(unittest.TestCase):
ray.init(num_workers=1)
net = LossActor(use_loss=False)
self.assertEqual(len(net.values[0].variables.items()), 1)
self.assertEqual(len(net.values[0].placeholders.items()), 1)
assert len(net.values[0].variables.items()) == 1
assert len(net.values[0].placeholders.items()) == 1
net.values[0].set_weights(net.values[0].get_weights())
@@ -164,8 +164,8 @@ class TensorFlowTest(unittest.TestCase):
ray.init(num_workers=1)
net = LossActor()
self.assertEqual(len(net.values[0].variables.items()), 3)
self.assertEqual(len(net.values[0].placeholders.items()), 3)
assert len(net.values[0].variables.items()) == 3
assert len(net.values[0].placeholders.items()) == 3
net.values[0].set_weights(net.values[0].get_weights())
@@ -183,20 +183,20 @@ class TensorFlowTest(unittest.TestCase):
# This only works because at the moment they have size 1.
weights1 = net1.get_weights()
weights2 = net2.get_weights()
self.assertNotEqual(weights1, weights2)
assert weights1 != weights2
# Set the weights and get the weights, and make sure they are
# unchanged.
new_weights1 = net1.set_and_get_weights(weights1)
new_weights2 = net2.set_and_get_weights(weights2)
self.assertEqual(weights1, new_weights1)
self.assertEqual(weights2, new_weights2)
assert weights1 == new_weights1
assert weights2 == new_weights2
# Swap the weights.
new_weights1 = net2.set_and_get_weights(weights1)
new_weights2 = net1.set_and_get_weights(weights2)
self.assertEqual(weights1, new_weights1)
self.assertEqual(weights2, new_weights2)
assert weights1 == new_weights1
assert weights2 == new_weights2
# This test creates an additional network on the driver so that the
# tensorflow variables on the driver and the worker differ.
@@ -214,7 +214,7 @@ class TensorFlowTest(unittest.TestCase):
new_weights2 = ray.get(
net2.set_and_get_weights.remote(net2.get_weights.remote()))
self.assertEqual(weights2, new_weights2)
assert weights2 == new_weights2
def testVariablesControlDependencies(self):
ray.init(num_workers=1)
@@ -228,7 +228,7 @@ class TensorFlowTest(unittest.TestCase):
# Tests if all variables are properly retrieved, 2 variables and 2
# momentum variables.
self.assertEqual(len(net_vars.variables.items()), 4)
assert len(net_vars.variables.items()) == 4
def testRemoteTrainingStep(self):
ray.init(num_workers=1)
@@ -263,7 +263,7 @@ class TensorFlowTest(unittest.TestCase):
sess.run(train, feed_dict=feed_dict)
after_acc = sess.run(
loss, feed_dict=dict(zip(placeholders, [[2] * 100, [4] * 100])))
self.assertTrue(before_acc < after_acc)
assert before_acc < after_acc
if __name__ == "__main__":