Add test for listen_elastic_ip

python/xoscar/backends/indigen/tests/test_pool.py

@@ -542,6 +542,46 @@ async def test_create_actor_pool():
     assert len(global_router._mapping) == 0
 
 
+@pytest.mark.asyncio
+async def test_create_actor_pool_elastic_ip():
+    start_method = (
+        os.environ.get("POOL_START_METHOD", "forkserver")
+        if sys.platform != "win32"
+        else None
+    )
+    pool = await create_actor_pool(
+        "111.111.111.111",
+        pool_cls=MainActorPool,
+        n_process=0,
+        subprocess_start_method=start_method,
+        extra_conf={"listen_elastic_ip": True},
+    )
+    async with pool:
+        # test global router
+        global_router = Router.get_instance()
+        # global router should not be the identical one with pool's router
+        assert global_router is not pool.router
+        assert pool.external_address in global_router._curr_external_addresses
+        assert pool.external_address in global_router._mapping
+        assert pool.external_address == "111.111.111.111"
+
+        ctx = get_context()
+
+        # actor on main pool
+        actor_ref = await ctx.create_actor(
+            TestActor, uid="test-1", address=pool.external_address
+        )
+        assert await actor_ref.add(3) == 3
+        await ctx.destroy_actor(actor_ref)
+        assert (await ctx.has_actor(actor_ref)) is False
+
+    assert pool.stopped
+    # after pool shutdown, global router must has been cleaned
+    global_router = Router.get_instance()
+    assert len(global_router._curr_external_addresses) == 0
+    assert len(global_router._mapping) == 0
+
+
 @pytest.mark.asyncio
 async def test_errors():
     with pytest.raises(ValueError):