diff --git a/test/actor_test.py b/test/actor_test.py index a0fe6d410..17c413f43 100644 --- a/test/actor_test.py +++ b/test/actor_test.py @@ -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 diff --git a/test/array_test.py b/test/array_test.py index feace8c55..f3c2400af 100644 --- a/test/array_test.py +++ b/test/array_test.py @@ -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)) diff --git a/test/autoscaler_test.py b/test/autoscaler_test.py index 845a1c870..52b6fde7b 100644 --- a/test/autoscaler_test.py +++ b/test/autoscaler_test.py @@ -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__": diff --git a/test/component_failures_test.py b/test/component_failures_test.py index 864033498..f604a23b5 100644 --- a/test/component_failures_test.py +++ b/test/component_failures_test.py @@ -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.") diff --git a/test/cython_test.py b/test/cython_test.py index 027dbafd8..f0241a16a 100644 --- a/test/cython_test.py +++ b/test/cython_test.py @@ -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]) diff --git a/test/failure_test.py b/test/failure_test.py index 00a1edd4d..d15b8526c 100644 --- a/test/failure_test.py +++ b/test/failure_test.py @@ -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( diff --git a/test/monitor_test.py b/test/monitor_test.py index e497b4347..b4680a575 100644 --- a/test/monitor_test.py +++ b/test/monitor_test.py @@ -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() diff --git a/test/multi_node_test.py b/test/multi_node_test.py index 520f94e39..f660103f5 100644 --- a/test/multi_node_test.py +++ b/test/multi_node_test.py @@ -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__": diff --git a/test/runtest.py b/test/runtest.py index 8e11adb4c..e5654e922 100644 --- a/test/runtest.py +++ b/test/runtest.py @@ -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", diff --git a/test/tensorflow_test.py b/test/tensorflow_test.py index 04261673f..7fc215d43 100644 --- a/test/tensorflow_test.py +++ b/test/tensorflow_test.py @@ -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__":