diff --git a/ably/http/http.py b/ably/http/http.py
index 0792df99..d21a9386 100644
--- a/ably/http/http.py
+++ b/ably/http/http.py
@@ -140,9 +140,9 @@ def dump_body(self, body):
         else:
             return json.dumps(body, separators=(',', ':'))
 
-    def get_rest_hosts(self):
-        hosts = self.options.get_rest_hosts()
-        host = self.__host or self.options.fallback_realtime_host
+    def get_hosts(self):
+        hosts = self.options.get_hosts()
+        host = self.__host or self.options.fallback_host
         if host is None:
             return hosts
 
@@ -186,7 +186,7 @@ async def make_request(self, method, path, version=None, headers=None, body=None
         http_max_retry_duration = self.http_max_retry_duration
         requested_at = time.time()
 
-        hosts = self.get_rest_hosts()
+        hosts = self.get_hosts()
         for retry_count, host in enumerate(hosts):
             def should_stop_retrying(retry_count=retry_count):
                 time_passed = time.time() - requested_at
@@ -229,7 +229,7 @@ def should_stop_retrying(retry_count=retry_count):
                         continue
 
                     # Keep fallback host for later (RSC15f)
-                    if retry_count > 0 and host != self.options.get_rest_host():
+                    if retry_count > 0 and host != self.options.get_host():
                         self.__host = host
                         self.__host_expires = time.time() + (self.options.fallback_retry_timeout / 1000.0)
 
@@ -277,7 +277,7 @@ def options(self):
 
     @property
     def preferred_host(self):
-        return self.options.get_rest_host()
+        return self.options.get_host()
 
     @property
     def preferred_port(self):
diff --git a/ably/realtime/connectionmanager.py b/ably/realtime/connectionmanager.py
index 01a0735b..dedb78f5 100644
--- a/ably/realtime/connectionmanager.py
+++ b/ably/realtime/connectionmanager.py
@@ -122,7 +122,7 @@ def __init__(self, realtime: AblyRealtime, initial_state):
         self.retry_timer: Timer | None = None
         self.connect_base_task: asyncio.Task | None = None
         self.disconnect_transport_task: asyncio.Task | None = None
-        self.__fallback_hosts: list[str] = self.options.get_fallback_realtime_hosts()
+        self.__fallback_hosts: list[str] = self.options.get_fallback_hosts()
         self.queued_messages: deque[PendingMessage] = deque()
         self.__error_reason: AblyException | None = None
         self.msg_serial: int = 0
@@ -532,7 +532,7 @@ async def connect_with_fallback_hosts(self, fallback_hosts: list) -> Exception |
 
     async def connect_base(self) -> None:
         fallback_hosts = self.__fallback_hosts
-        primary_host = self.options.get_realtime_host()
+        primary_host = self.options.get_host()
         try:
             await self.try_host(primary_host)
             return
diff --git a/ably/realtime/realtime.py b/ably/realtime/realtime.py
index 9b9c4016..632236cd 100644
--- a/ably/realtime/realtime.py
+++ b/ably/realtime/realtime.py
@@ -48,10 +48,14 @@ def __init__(self, key: Optional[str] = None, loop: Optional[asyncio.AbstractEve
             You can set this to false and explicitly connect to Ably using the
             connect() method. The default is true.
         **kwargs: client options
+            endpoint: str
+                Endpoint specifies either a routing policy name or fully qualified domain name to connect to Ably.
             realtime_host: str
+                Deprecated: this property is deprecated and will be removed in a future version.
                 Enables a non-default Ably host to be specified for realtime connections.
                 For development environments only. The default value is realtime.ably.io.
             environment: str
+                Deprecated: this property is deprecated and will be removed in a future version.
                 Enables a custom environment to be used with the Ably service. Defaults to `production`
             realtime_request_timeout: float
                 Timeout (in milliseconds) for the wait of acknowledgement for operations performed via a realtime
diff --git a/ably/rest/rest.py b/ably/rest/rest.py
index a77fcd90..bc84e638 100644
--- a/ably/rest/rest.py
+++ b/ably/rest/rest.py
@@ -32,8 +32,14 @@ def __init__(self, key: Optional[str] = None, token: Optional[str] = None,
 
           **Optional Parameters**
           - `client_id`: Undocumented
-          - `rest_host`: The host to connect to. Defaults to rest.ably.io
-          - `environment`: The environment to use. Defaults to 'production'
+          - `endpoint`: Endpoint specifies either a routing policy name or
+            fully qualified domain name to connect to Ably.
+          - `rest_host`: Deprecated: this property is deprecated and will
+            be removed in a future version. The host to connect to.
+            Defaults to rest.ably.io
+          - `environment`: Deprecated: this property is deprecated and
+            will be removed in a future version. The environment to use.
+            Defaults to 'production'
           - `port`: The port to connect to. Defaults to 80
           - `tls_port`: The tls_port to connect to. Defaults to 443
           - `tls`: Specifies whether the client should use TLS. Defaults
diff --git a/ably/transport/defaults.py b/ably/transport/defaults.py
index 7a732d9a..4d785387 100644
--- a/ably/transport/defaults.py
+++ b/ably/transport/defaults.py
@@ -1,17 +1,8 @@
 class Defaults:
     protocol_version = "2"
-    fallback_hosts = [
-        "a.ably-realtime.com",
-        "b.ably-realtime.com",
-        "c.ably-realtime.com",
-        "d.ably-realtime.com",
-        "e.ably-realtime.com",
-    ]
-
-    rest_host = "rest.ably.io"
-    realtime_host = "realtime.ably.io"  # RTN2
+
     connectivity_check_url = "https://internet-up.ably-realtime.com/is-the-internet-up.txt"
-    environment = 'production'
+    endpoint = 'main'
 
     port = 80
     tls_port = 443
@@ -53,11 +44,34 @@ def get_scheme(options):
             return "http"
 
     @staticmethod
-    def get_environment_fallback_hosts(environment):
+    def get_hostname(endpoint):
+        if "." in endpoint or "::" in endpoint or "localhost" in endpoint:
+            return endpoint
+
+        if endpoint.startswith("nonprod:"):
+            return endpoint[len("nonprod:"):] + ".realtime.ably-nonprod.net"
+
+        return endpoint + ".realtime.ably.net"
+
+    @staticmethod
+    def get_fallback_hosts(endpoint="main"):
+        if "." in endpoint or "::" in endpoint or "localhost" in endpoint:
+            return []
+
+        if endpoint.startswith("nonprod:"):
+            root = endpoint.replace("nonprod:", "")
+            return [
+                root + ".a.fallback.ably-realtime-nonprod.com",
+                root + ".b.fallback.ably-realtime-nonprod.com",
+                root + ".c.fallback.ably-realtime-nonprod.com",
+                root + ".d.fallback.ably-realtime-nonprod.com",
+                root + ".e.fallback.ably-realtime-nonprod.com",
+            ]
+
         return [
-            environment + "-a-fallback.ably-realtime.com",
-            environment + "-b-fallback.ably-realtime.com",
-            environment + "-c-fallback.ably-realtime.com",
-            environment + "-d-fallback.ably-realtime.com",
-            environment + "-e-fallback.ably-realtime.com",
+            endpoint + ".a.fallback.ably-realtime.com",
+            endpoint + ".b.fallback.ably-realtime.com",
+            endpoint + ".c.fallback.ably-realtime.com",
+            endpoint + ".d.fallback.ably-realtime.com",
+            endpoint + ".e.fallback.ably-realtime.com",
         ]
diff --git a/ably/transport/websockettransport.py b/ably/transport/websockettransport.py
index 325685b7..6c57f3f5 100644
--- a/ably/transport/websockettransport.py
+++ b/ably/transport/websockettransport.py
@@ -142,8 +142,8 @@ async def on_protocol_message(self, msg):
                 self.max_idle_interval = max_idle_interval + self.options.realtime_request_timeout
                 self.on_activity()
             self.is_connected = True
-            if self.host != self.options.get_realtime_host():  # RTN17e
-                self.options.fallback_realtime_host = self.host
+            if self.host != self.options.get_host():  # RTN17e
+                self.options.fallback_host = self.host
             self.connection_manager.on_connected(connection_details, connection_id, reason=exception)
         elif action == ProtocolMessageAction.DISCONNECTED:
             error = msg.get('error')
diff --git a/ably/types/options.py b/ably/types/options.py
index 8804b3b9..16e2279d 100644
--- a/ably/types/options.py
+++ b/ably/types/options.py
@@ -26,16 +26,22 @@ def decode(self, delta: bytes, base: bytes) -> bytes:
 
 class Options(AuthOptions):
     def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realtime_host=None, port=0,
-                 tls_port=0, use_binary_protocol=True, queue_messages=True, recover=False, environment=None,
-                 http_open_timeout=None, http_request_timeout=None, realtime_request_timeout=None,
-                 http_max_retry_count=None, http_max_retry_duration=None, fallback_hosts=None,
-                 fallback_retry_timeout=None, disconnected_retry_timeout=None, idempotent_rest_publishing=None,
-                 loop=None, auto_connect=True, suspended_retry_timeout=None, connectivity_check_url=None,
+                 tls_port=0, use_binary_protocol=True, queue_messages=True, recover=False, endpoint=None,
+                 environment=None, http_open_timeout=None, http_request_timeout=None,
+                 realtime_request_timeout=None, http_max_retry_count=None, http_max_retry_duration=None,
+                 fallback_hosts=None, fallback_retry_timeout=None,
+                 disconnected_retry_timeout=None, idempotent_rest_publishing=None, loop=None, auto_connect=True,
+                 suspended_retry_timeout=None, connectivity_check_url=None,
                  channel_retry_timeout=Defaults.channel_retry_timeout, add_request_ids=False,
                  vcdiff_decoder: VCDiffDecoder = None, transport_params=None, **kwargs):
 
         super().__init__(**kwargs)
 
+        # REC1b1: endpoint is incompatible with deprecated options
+        if endpoint is not None:
+            if environment is not None or rest_host is not None or realtime_host is not None:
+                raise ValueError('endpoint is incompatible with any of environment, rest_host or realtime_host')
+
         # TODO check these defaults
         if fallback_retry_timeout is None:
             fallback_retry_timeout = Defaults.fallback_retry_timeout
@@ -64,25 +70,39 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti
             from ably import api_version
             idempotent_rest_publishing = api_version >= '1.2'
 
-        if environment is None:
-            environment = Defaults.environment
+        if environment is not None and endpoint is None:
+            log.warning("environment client option is deprecated, please use endpoint instead")
+            endpoint = environment
+
+        # REC1d: restHost or realtimeHost option
+        # REC1d1: restHost takes precedence over realtimeHost
+        if rest_host is not None and endpoint is None:
+            log.warning("rest_host client option is deprecated, please use endpoint instead")
+            endpoint = rest_host
+        elif realtime_host is not None and endpoint is None:
+            # REC1d2: realtimeHost if restHost not specified
+            log.warning("realtime_host client option is deprecated, please use endpoint instead")
+            endpoint = realtime_host
+
+        if endpoint is None:
+            endpoint = Defaults.endpoint
 
         self.__client_id = client_id
         self.__log_level = log_level
         self.__tls = tls
-        self.__rest_host = rest_host
-        self.__realtime_host = realtime_host
         self.__port = port
         self.__tls_port = tls_port
         self.__use_binary_protocol = use_binary_protocol
         self.__queue_messages = queue_messages
         self.__recover = recover
-        self.__environment = environment
+        self.__endpoint = endpoint
         self.__http_open_timeout = http_open_timeout
         self.__http_request_timeout = http_request_timeout
         self.__realtime_request_timeout = realtime_request_timeout
         self.__http_max_retry_count = http_max_retry_count
         self.__http_max_retry_duration = http_max_retry_duration
+        # Field for internal use only
+        self.__fallback_host = None
         self.__fallback_hosts = fallback_hosts
         self.__fallback_retry_timeout = fallback_retry_timeout
         self.__disconnected_retry_timeout = disconnected_retry_timeout
@@ -93,13 +113,10 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti
         self.__connection_state_ttl = connection_state_ttl
         self.__suspended_retry_timeout = suspended_retry_timeout
         self.__connectivity_check_url = connectivity_check_url
-        self.__fallback_realtime_host = None
         self.__add_request_ids = add_request_ids
         self.__vcdiff_decoder = vcdiff_decoder
         self.__transport_params = transport_params or {}
-
-        self.__rest_hosts = self.__get_rest_hosts()
-        self.__realtime_hosts = self.__get_realtime_hosts()
+        self.__hosts = self.__get_hosts()
 
     @property
     def client_id(self):
@@ -125,23 +142,6 @@ def tls(self):
     def tls(self, value):
         self.__tls = value
 
-    @property
-    def rest_host(self):
-        return self.__rest_host
-
-    @rest_host.setter
-    def rest_host(self, value):
-        self.__rest_host = value
-
-    # RTC1d
-    @property
-    def realtime_host(self):
-        return self.__realtime_host
-
-    @realtime_host.setter
-    def realtime_host(self, value):
-        self.__realtime_host = value
-
     @property
     def port(self):
         return self.__port
@@ -183,8 +183,8 @@ def recover(self, value):
         self.__recover = value
 
     @property
-    def environment(self):
-        return self.__environment
+    def endpoint(self):
+        return self.__endpoint
 
     @property
     def http_open_timeout(self):
@@ -268,12 +268,18 @@ def connectivity_check_url(self):
         return self.__connectivity_check_url
 
     @property
-    def fallback_realtime_host(self):
-        return self.__fallback_realtime_host
+    def fallback_host(self):
+        """
+        For internal use only, can be deleted in future
+        """
+        return self.__fallback_host
 
-    @fallback_realtime_host.setter
-    def fallback_realtime_host(self, value):
-        self.__fallback_realtime_host = value
+    @fallback_host.setter
+    def fallback_host(self, value):
+        """
+        For internal use only, can be deleted in future
+        """
+        self.__fallback_host = value
 
     @property
     def add_request_ids(self):
@@ -287,37 +293,20 @@ def vcdiff_decoder(self):
     def transport_params(self):
         return self.__transport_params
 
-    def __get_rest_hosts(self):
+    def __get_hosts(self):
         """
         Return the list of hosts as they should be tried. First comes the main
         host. Then the fallback hosts in random order.
         The returned list will have a length of up to http_max_retry_count.
         """
-        # Defaults
-        host = self.rest_host
-        if host is None:
-            host = Defaults.rest_host
-
-        environment = self.environment
+        host = Defaults.get_hostname(self.endpoint)
+        # REC2: Determine fallback hosts
+        fallback_hosts = self.get_fallback_hosts()
 
         http_max_retry_count = self.http_max_retry_count
         if http_max_retry_count is None:
             http_max_retry_count = Defaults.http_max_retry_count
 
-        # Prepend environment
-        if environment != 'production':
-            host = f'{environment}-{host}'
-
-        # Fallback hosts
-        fallback_hosts = self.fallback_hosts
-        if fallback_hosts is None:
-            if host == Defaults.rest_host:
-                fallback_hosts = Defaults.fallback_hosts
-            elif environment != 'production':
-                fallback_hosts = Defaults.get_environment_fallback_hosts(environment)
-            else:
-                fallback_hosts = []
-
         # Shuffle
         fallback_hosts = list(fallback_hosts)
         random.shuffle(fallback_hosts)
@@ -328,31 +317,19 @@ def __get_rest_hosts(self):
         hosts = hosts[:http_max_retry_count]
         return hosts
 
-    def __get_realtime_hosts(self):
-        if self.realtime_host is not None:
-            host = self.realtime_host
-            return [host]
-        elif self.environment != "production":
-            host = f'{self.environment}-{Defaults.realtime_host}'
-        else:
-            host = Defaults.realtime_host
-
-        return [host] + self.__fallback_hosts
-
-    def get_rest_hosts(self):
-        return self.__rest_hosts
-
-    def get_rest_host(self):
-        return self.__rest_hosts[0]
-
-    def get_realtime_hosts(self):
-        return self.__realtime_hosts
+    def get_hosts(self):
+        return self.__hosts
 
-    def get_realtime_host(self):
-        return self.__realtime_hosts[0]
+    def get_host(self):
+        return self.__hosts[0]
 
-    def get_fallback_rest_hosts(self):
-        return self.__rest_hosts[1:]
+    # REC2: Various client options collectively determine a set of fallback domains
+    def get_fallback_hosts(self):
+        # REC2a: If the fallbackHosts client option is specified
+        if self.__fallback_hosts is not None:
+            # REC2a2: the set of fallback domains is given by the value of the fallbackHosts option
+            return self.__fallback_hosts
 
-    def get_fallback_realtime_hosts(self):
-        return self.__realtime_hosts[1:]
+        # REC2c: Otherwise, the set of fallback domains is defined implicitly by the options
+        # used to define the primary domain as specified in (REC1)
+        return Defaults.get_fallback_hosts(self.endpoint)
diff --git a/test/ably/realtime/realtimeconnection_test.py b/test/ably/realtime/realtimeconnection_test.py
index 76e52e43..d3075e8e 100644
--- a/test/ably/realtime/realtimeconnection_test.py
+++ b/test/ably/realtime/realtimeconnection_test.py
@@ -187,7 +187,7 @@ async def test_connectivity_check_bad_status(self):
         assert ably.connection.connection_manager.check_connection() is False
 
     async def test_unroutable_host(self):
-        ably = await TestApp.get_ably_realtime(realtime_host="10.255.255.1", realtime_request_timeout=3000)
+        ably = await TestApp.get_ably_realtime(endpoint="10.255.255.1", realtime_request_timeout=3000)
         state_change = await ably.connection.once_async()
         assert state_change.reason
         assert state_change.reason.code == 50003
@@ -197,7 +197,7 @@ async def test_unroutable_host(self):
         await ably.close()
 
     async def test_invalid_host(self):
-        ably = await TestApp.get_ably_realtime(realtime_host="iamnotahost")
+        ably = await TestApp.get_ably_realtime(endpoint="iamnotahost")
         state_change = await ably.connection.once_async()
         assert state_change.reason
         assert state_change.reason.code == 40000
@@ -299,8 +299,8 @@ async def test_fallback_host(self):
 
         await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
 
-        assert ably.connection.connection_manager.transport.host != self.test_vars["realtime_host"]
-        assert ably.options.fallback_realtime_host != self.test_vars["realtime_host"]
+        assert ably.connection.connection_manager.transport.host != self.test_vars["endpoint"]
+        assert ably.options.fallback_host != self.test_vars["endpoint"]
         await ably.close()
 
     async def test_fallback_host_no_connection(self):
@@ -325,7 +325,7 @@ def check_connection():
 
         await ably.connection.once_async(ConnectionState.DISCONNECTED)
 
-        assert ably.options.fallback_realtime_host is None
+        assert ably.options.fallback_host is None
         await ably.close()
 
     async def test_fallback_host_disconnected_protocol_msg(self):
@@ -344,8 +344,8 @@ async def test_fallback_host_disconnected_protocol_msg(self):
 
         await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
 
-        assert ably.connection.connection_manager.transport.host != self.test_vars["realtime_host"]
-        assert ably.options.fallback_realtime_host != self.test_vars["realtime_host"]
+        assert ably.connection.connection_manager.transport.host != self.test_vars["endpoint"]
+        assert ably.options.fallback_host != self.test_vars["endpoint"]
         await ably.close()
 
     #  RTN2d
diff --git a/test/ably/rest/restauth_test.py b/test/ably/rest/restauth_test.py
index 9c0495ba..185021e1 100644
--- a/test/ably/rest/restauth_test.py
+++ b/test/ably/rest/restauth_test.py
@@ -486,7 +486,7 @@ class TestRenewToken(BaseAsyncTestCase):
     async def setup(self):
         self.test_vars = await TestApp.get_test_vars()
         self.host = 'fake-host.ably.io'
-        self.ably = await TestApp.get_ably_rest(use_binary_protocol=False, rest_host=self.host)
+        self.ably = await TestApp.get_ably_rest(use_binary_protocol=False, endpoint=self.host)
         # with headers
         self.publish_attempts = 0
         self.channel = uuid.uuid4().hex
@@ -549,7 +549,7 @@ async def test_when_not_renewable(self):
 
         self.ably = await TestApp.get_ably_rest(
             key=None,
-            rest_host=self.host,
+            endpoint=self.host,
             token='token ID cannot be used to create a new token',
             use_binary_protocol=False)
         await self.ably.channels[self.channel].publish('evt', 'msg')
@@ -568,7 +568,7 @@ async def test_when_not_renewable_with_token_details(self):
         token_details = TokenDetails(token='a_dummy_token')
         self.ably = await TestApp.get_ably_rest(
             key=None,
-            rest_host=self.host,
+            endpoint=self.host,
             token_details=token_details,
             use_binary_protocol=False)
         await self.ably.channels[self.channel].publish('evt', 'msg')
@@ -638,7 +638,7 @@ def cb_publish(request):
 
     # RSA4b1
     async def test_query_time_false(self):
-        ably = await TestApp.get_ably_rest(rest_host=self.host)
+        ably = await TestApp.get_ably_rest(endpoint=self.host)
         await ably.auth.authorize()
         self.publish_fail = True
         await ably.channels[self.channel].publish('evt', 'msg')
@@ -647,7 +647,7 @@ async def test_query_time_false(self):
 
     # RSA4b1
     async def test_query_time_true(self):
-        ably = await TestApp.get_ably_rest(query_time=True, rest_host=self.host)
+        ably = await TestApp.get_ably_rest(query_time=True, endpoint=self.host)
         await ably.auth.authorize()
         self.publish_fail = False
         await ably.channels[self.channel].publish('evt', 'msg')
diff --git a/test/ably/rest/resthttp_test.py b/test/ably/rest/resthttp_test.py
index ba101c21..badbba1f 100644
--- a/test/ably/rest/resthttp_test.py
+++ b/test/ably/rest/resthttp_test.py
@@ -64,13 +64,13 @@ def make_url(host):
 
                 expected_urls_set = {
                     make_url(host)
-                    for host in Options(http_max_retry_count=10).get_rest_hosts()
+                    for host in Options(http_max_retry_count=10).get_hosts()
                 }
                 for ((_, url), _) in request_mock.call_args_list:
                     assert url in expected_urls_set
                     expected_urls_set.remove(url)
 
-                expected_hosts_set = set(Options(http_max_retry_count=10).get_rest_hosts())
+                expected_hosts_set = set(Options(http_max_retry_count=10).get_hosts())
                 for (prep_request_tuple, _) in send_mock.call_args_list:
                     assert prep_request_tuple[0].headers.get('host') in expected_hosts_set
                     expected_hosts_set.remove(prep_request_tuple[0].headers.get('host'))
@@ -79,7 +79,7 @@ def make_url(host):
     @respx.mock
     async def test_no_host_fallback_nor_retries_if_custom_host(self):
         custom_host = 'example.org'
-        ably = AblyRest(token="foo", rest_host=custom_host)
+        ably = AblyRest(token="foo", endpoint=custom_host)
 
         mock_route = respx.get("https://example.org").mock(side_effect=httpx.RequestError(''))
 
@@ -95,7 +95,7 @@ async def test_no_host_fallback_nor_retries_if_custom_host(self):
     async def test_cached_fallback(self):
         timeout = 2000
         ably = await TestApp.get_ably_rest(fallback_retry_timeout=timeout)
-        host = ably.options.get_rest_host()
+        host = ably.options.get_host()
 
         state = {'errors': 0}
         client = httpx.AsyncClient(http2=True)
@@ -128,7 +128,7 @@ async def side_effect(*args, **kwargs):
 
     @respx.mock
     async def test_no_retry_if_not_500_to_599_http_code(self):
-        default_host = Options().get_rest_host()
+        default_host = Options().get_host()
         ably = AblyRest(token="foo")
 
         default_url = f"{ably.http.preferred_scheme}://{default_host}:{ably.http.preferred_port}/"
@@ -215,7 +215,7 @@ async def test_request_over_http2(self):
         url = 'https://www.example.com'
         respx.get(url).mock(return_value=Response(status_code=200))
 
-        ably = await TestApp.get_ably_rest(rest_host=url)
+        ably = await TestApp.get_ably_rest(endpoint=url)
         r = await ably.http.make_request('GET', url, skip_auth=True)
         assert r.http_version == 'HTTP/2'
         await ably.close()
diff --git a/test/ably/rest/restinit_test.py b/test/ably/rest/restinit_test.py
index 8e8197d8..6b795961 100644
--- a/test/ably/rest/restinit_test.py
+++ b/test/ably/rest/restinit_test.py
@@ -71,22 +71,22 @@ def test_with_options_auth_url(self):
     def test_rest_host_and_environment(self):
         # rest host
         ably = AblyRest(token='foo', rest_host="some.other.host")
-        assert "some.other.host" == ably.options.rest_host, "Unexpected host mismatch"
+        assert "some.other.host" == ably.options.get_host(), "Unexpected host mismatch"
 
-        # environment: production
-        ably = AblyRest(token='foo', environment="production")
-        host = ably.options.get_rest_host()
-        assert "rest.ably.io" == host, f"Unexpected host mismatch {host}"
+        # environment: main
+        ably = AblyRest(token='foo', endpoint="main")
+        host = ably.options.get_host()
+        assert "main.realtime.ably.net" == host, f"Unexpected host mismatch {host}"
 
         # environment: other
-        ably = AblyRest(token='foo', environment="sandbox")
-        host = ably.options.get_rest_host()
-        assert "sandbox-rest.ably.io" == host, f"Unexpected host mismatch {host}"
+        ably = AblyRest(token='foo', endpoint="nonprod:sandbox")
+        host = ably.options.get_host()
+        assert "sandbox.realtime.ably-nonprod.net" == host, f"Unexpected host mismatch {host}"
 
         # both, as per #TO3k2
         with pytest.raises(ValueError):
             ably = AblyRest(token='foo', rest_host="some.other.host",
-                            environment="some.other.environment")
+                            endpoint="some.other.environment")
 
     # RSC15
     @dont_vary_protocol
@@ -100,16 +100,16 @@ def test_fallback_hosts(self):
         # Fallback hosts specified (RSC15g1)
         for aux in fallback_hosts:
             ably = AblyRest(token='foo', fallback_hosts=aux)
-            assert sorted(aux) == sorted(ably.options.get_fallback_rest_hosts())
+            assert sorted(aux) == sorted(ably.options.get_fallback_hosts())
 
-        # Specify environment (RSC15g2)
-        ably = AblyRest(token='foo', environment='sandbox', http_max_retry_count=10)
-        assert sorted(Defaults.get_environment_fallback_hosts('sandbox')) == sorted(
-            ably.options.get_fallback_rest_hosts())
+        # Specify endpoint (RSC15g2)
+        ably = AblyRest(token='foo', endpoint='nonprod:sandbox', http_max_retry_count=10)
+        assert sorted(Defaults.get_fallback_hosts('nonprod:sandbox')) == sorted(
+            ably.options.get_fallback_hosts())
 
-        # Fallback hosts and environment not specified (RSC15g3)
+        # Fallback hosts and endpoint not specified (RSC15g3)
         ably = AblyRest(token='foo', http_max_retry_count=10)
-        assert sorted(Defaults.fallback_hosts) == sorted(ably.options.get_fallback_rest_hosts())
+        assert sorted(Defaults.get_fallback_hosts()) == sorted(ably.options.get_fallback_hosts())
 
         # RSC15f
         ably = AblyRest(token='foo')
@@ -118,9 +118,9 @@ def test_fallback_hosts(self):
         assert 1000 == ably.options.fallback_retry_timeout
 
     @dont_vary_protocol
-    def test_specified_realtime_host(self):
-        ably = AblyRest(token='foo', realtime_host="some.other.host")
-        assert "some.other.host" == ably.options.realtime_host, "Unexpected host mismatch"
+    def test_specified_host(self):
+        ably = AblyRest(token='foo', endpoint="some.other.host")
+        assert "some.other.host" == ably.options.get_host(), "Unexpected host mismatch"
 
     @dont_vary_protocol
     def test_specified_port(self):
@@ -182,13 +182,13 @@ async def test_query_time_param(self):
     @dont_vary_protocol
     def test_requests_over_https_production(self):
         ably = AblyRest(token='token')
-        assert 'https://rest.ably.io' == f'{ably.http.preferred_scheme}://{ably.http.preferred_host}'
+        assert 'https://main.realtime.ably.net' == f'{ably.http.preferred_scheme}://{ably.http.preferred_host}'
         assert ably.http.preferred_port == 443
 
     @dont_vary_protocol
     def test_requests_over_http_production(self):
         ably = AblyRest(token='token', tls=False)
-        assert 'http://rest.ably.io' == f'{ably.http.preferred_scheme}://{ably.http.preferred_host}'
+        assert 'http://main.realtime.ably.net' == f'{ably.http.preferred_scheme}://{ ably.http.preferred_host}'
         assert ably.http.preferred_port == 80
 
     @dont_vary_protocol
@@ -204,14 +204,14 @@ async def test_request_basic_auth_over_http_fails(self):
 
     @dont_vary_protocol
     async def test_environment(self):
-        ably = AblyRest(token='token', environment='custom')
+        ably = AblyRest(token='token', endpoint='custom')
         with patch.object(AsyncClient, 'send', wraps=ably.http._Http__client.send) as get_mock:
             try:
                 await ably.time()
             except AblyException:
                 pass
             request = get_mock.call_args_list[0][0][0]
-            assert request.url == 'https://custom-rest.ably.io:443/time'
+            assert request.url == 'https://custom.realtime.ably.net:443/time'
 
         await ably.close()
 
diff --git a/test/ably/rest/restpaginatedresult_test.py b/test/ably/rest/restpaginatedresult_test.py
index 0ec6bb95..9aa85689 100644
--- a/test/ably/rest/restpaginatedresult_test.py
+++ b/test/ably/rest/restpaginatedresult_test.py
@@ -32,7 +32,7 @@ async def setup(self):
         self.ably = await TestApp.get_ably_rest(use_binary_protocol=False)
         # Mocked responses
         # without specific headers
-        self.mocked_api = respx.mock(base_url='http://rest.ably.io')
+        self.mocked_api = respx.mock(base_url='http://main.realtime.ably.net')
         self.ch1_route = self.mocked_api.get('/channels/channel_name/ch1')
         self.ch1_route.return_value = Response(
             headers={'content-type': 'application/json'},
@@ -45,8 +45,8 @@ async def setup(self):
             headers={
                 'content-type': 'application/json',
                 'link':
-                    '<http://rest.ably.io/channels/channel_name/ch2?page=1>; rel="first",'
-                    ' <http://rest.ably.io/channels/channel_name/ch2?page=2>; rel="next"'
+                    '<http://main.realtime.ably.net/channels/channel_name/ch2?page=1>; rel="first",'
+                    ' <http://main.realtime.ably.net/channels/channel_name/ch2?page=2>; rel="next"'
             },
             body='[{"id": 0}, {"id": 1}]',
             status=200
@@ -56,11 +56,11 @@ async def setup(self):
 
         self.paginated_result = await PaginatedResult.paginated_query(
             self.ably.http,
-            url='http://rest.ably.io/channels/channel_name/ch1',
+            url='http://main.realtime.ably.net/channels/channel_name/ch1',
             response_processor=lambda response: response.to_native())
         self.paginated_result_with_headers = await PaginatedResult.paginated_query(
             self.ably.http,
-            url='http://rest.ably.io/channels/channel_name/ch2',
+            url='http://main.realtime.ably.net/channels/channel_name/ch2',
             response_processor=lambda response: response.to_native())
         yield
         self.mocked_api.stop()
diff --git a/test/ably/rest/restrequest_test.py b/test/ably/rest/restrequest_test.py
index 7380ea07..4a08e1c5 100644
--- a/test/ably/rest/restrequest_test.py
+++ b/test/ably/rest/restrequest_test.py
@@ -100,8 +100,8 @@ async def test_timeout(self):
             await ably.request('GET', '/time', version=Defaults.protocol_version)
         await ably.close()
 
-        default_endpoint = 'https://sandbox-rest.ably.io/time'
-        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        default_endpoint = 'https://sandbox.realtime.ably-nonprod.net/time'
+        fallback_host = 'sandbox.a.fallback.ably-realtime-nonprod.com'
         fallback_endpoint = f'https://{fallback_host}/time'
         ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
         with respx.mock:
@@ -117,7 +117,7 @@ async def test_timeout(self):
 
         # Bad host, no Fallback
         ably = AblyRest(key=self.test_vars["keys"][0]["key_str"],
-                        rest_host='some.other.host',
+                        endpoint='some.other.host',
                         port=self.test_vars["port"],
                         tls_port=self.test_vars["tls_port"],
                         tls=self.test_vars["tls"])
@@ -128,8 +128,8 @@ async def test_timeout(self):
     # RSC15l3
     @dont_vary_protocol
     async def test_503_status_fallback(self):
-        default_endpoint = 'https://sandbox-rest.ably.io/time'
-        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        default_endpoint = 'https://sandbox.realtime.ably-nonprod.net/time'
+        fallback_host = 'sandbox.a.fallback.ably-realtime-nonprod.com'
         fallback_endpoint = f'https://{fallback_host}/time'
         ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
         with respx.mock:
@@ -149,8 +149,8 @@ async def test_503_status_fallback(self):
     # RSC15l2
     @dont_vary_protocol
     async def test_httpx_timeout_fallback(self):
-        default_endpoint = 'https://sandbox-rest.ably.io/time'
-        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        default_endpoint = 'https://sandbox.realtime.ably-nonprod.net/time'
+        fallback_host = 'sandbox.a.fallback.ably-realtime-nonprod.com'
         fallback_endpoint = f'https://{fallback_host}/time'
         ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
         with respx.mock:
@@ -170,8 +170,8 @@ async def test_httpx_timeout_fallback(self):
     # RSC15l3
     @dont_vary_protocol
     async def test_503_status_fallback_on_publish(self):
-        default_endpoint = 'https://sandbox-rest.ably.io/channels/test/messages'
-        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        default_endpoint = 'https://sandbox.realtime.ably-nonprod.net/channels/test/messages'
+        fallback_host = 'sandbox.a.fallback.ably-realtime-nonprod.com'
         fallback_endpoint = f'https://{fallback_host}/channels/test/messages'
 
         fallback_response_text = (
@@ -201,8 +201,8 @@ async def test_503_status_fallback_on_publish(self):
     # RSC15l4
     @dont_vary_protocol
     async def test_400_cloudfront_fallback(self):
-        default_endpoint = 'https://sandbox-rest.ably.io/time'
-        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        default_endpoint = 'https://sandbox.realtime.ably-nonprod.net/time'
+        fallback_host = 'sandbox.a.fallback.ably-realtime-nonprod.com'
         fallback_endpoint = f'https://{fallback_host}/time'
         ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
         with respx.mock:
diff --git a/test/ably/rest/resttime_test.py b/test/ably/rest/resttime_test.py
index a0e962fd..4b78620a 100644
--- a/test/ably/rest/resttime_test.py
+++ b/test/ably/rest/resttime_test.py
@@ -35,7 +35,7 @@ async def test_time_without_key_or_token(self):
 
     @dont_vary_protocol
     async def test_time_fails_without_valid_host(self):
-        ably = await TestApp.get_ably_rest(key=None, token='foo', rest_host="this.host.does.not.exist")
+        ably = await TestApp.get_ably_rest(key=None, token='foo', endpoint="this.host.does.not.exist")
         with pytest.raises(AblyException):
             await ably.time()
 
diff --git a/test/ably/testapp.py b/test/ably/testapp.py
index a5efb06c..f657fdd4 100644
--- a/test/ably/testapp.py
+++ b/test/ably/testapp.py
@@ -4,6 +4,7 @@
 
 from ably.realtime.realtime import AblyRealtime
 from ably.rest.rest import AblyRest
+from ably.transport.defaults import Defaults
 from ably.types.capability import Capability
 from ably.types.options import Options
 from ably.util.exceptions import AblyException
@@ -14,23 +15,14 @@
     app_spec_local = json.loads(f.read())
 
 tls = (os.environ.get('ABLY_TLS') or "true").lower() == "true"
-rest_host = os.environ.get('ABLY_REST_HOST', 'sandbox-rest.ably.io')
-realtime_host = os.environ.get('ABLY_REALTIME_HOST', 'sandbox-realtime.ably.io')
-
-environment = os.environ.get('ABLY_ENV', 'sandbox')
+endpoint = os.environ.get('ABLY_ENDPOINT', 'nonprod:sandbox')
 
 port = 80
 tls_port = 443
 
-if rest_host and not rest_host.endswith("rest.ably.io"):
-    tls = tls and rest_host != "localhost"
-    port = 8080
-    tls_port = 8081
-
-
 ably = AblyRest(token='not_a_real_token',
                 port=port, tls_port=tls_port, tls=tls,
-                environment=environment,
+                endpoint=endpoint,
                 use_binary_protocol=False)
 
 
@@ -49,12 +41,11 @@ async def get_test_vars():
 
             test_vars = {
                 "app_id": app_id,
-                "host": rest_host,
                 "port": port,
                 "tls_port": tls_port,
                 "tls": tls,
-                "environment": environment,
-                "realtime_host": realtime_host,
+                "endpoint": endpoint,
+                "host": Defaults.get_hostname(endpoint),
                 "keys": [{
                     "key_name": "{}.{}".format(app_id, k.get("id", "")),
                     "key_secret": k.get("value", ""),
@@ -88,15 +79,12 @@ def get_options(test_vars, **kwargs):
             'port': test_vars["port"],
             'tls_port': test_vars["tls_port"],
             'tls': test_vars["tls"],
-            'environment': test_vars["environment"],
+            'endpoint': test_vars["endpoint"],
         }
         auth_methods = ["auth_url", "auth_callback", "token", "token_details", "key"]
         if not any(x in kwargs for x in auth_methods):
             options["key"] = test_vars["keys"][0]["key_str"]
 
-        if any(x in kwargs for x in ["rest_host", "realtime_host"]):
-            options["environment"] = None
-
         options.update(kwargs)
 
         return options
@@ -105,7 +93,6 @@ def get_options(test_vars, **kwargs):
     async def clear_test_vars():
         test_vars = TestApp.__test_vars
         options = Options(key=test_vars["keys"][0]["key_str"])
-        options.rest_host = test_vars["host"]
         options.port = test_vars["port"]
         options.tls_port = test_vars["tls_port"]
         options.tls = test_vars["tls"]
diff --git a/test/unit/http_test.py b/test/unit/http_test.py
index 45f362ed..61e0d35e 100644
--- a/test/unit/http_test.py
+++ b/test/unit/http_test.py
@@ -3,17 +3,17 @@
 
 def test_http_get_rest_hosts_works_when_fallback_realtime_host_is_set():
     ably = AblyRest(token="foo")
-    ably.options.fallback_realtime_host = ably.options.get_rest_hosts()[0]
+    ably.options.fallback_host = ably.options.get_hosts()[0]
     # Should not raise TypeError
-    hosts = ably.http.get_rest_hosts()
+    hosts = ably.http.get_hosts()
     assert isinstance(hosts, list)
     assert all(isinstance(host, str) for host in hosts)
 
 
 def test_http_get_rest_hosts_works_when_fallback_realtime_host_is_not_set():
     ably = AblyRest(token="foo")
-    ably.options.fallback_realtime_host = None
+    ably.options.fallback_host = None
     # Should not raise TypeError
-    hosts = ably.http.get_rest_hosts()
+    hosts = ably.http.get_hosts()
     assert isinstance(hosts, list)
     assert all(isinstance(host, str) for host in hosts)
diff --git a/test/unit/options_test.py b/test/unit/options_test.py
new file mode 100644
index 00000000..266ed9ed
--- /dev/null
+++ b/test/unit/options_test.py
@@ -0,0 +1,186 @@
+import pytest
+
+from ably.types.options import Options
+
+
+# REC1b1: endpoint is incompatible with deprecated options
+def test_options_should_fail_early_with_incompatible_client_options():
+    # REC1b1: endpoint with environment
+    with pytest.raises(ValueError):
+        Options(endpoint="foo", environment="foo")
+
+    # REC1b1: endpoint with rest_host
+    with pytest.raises(ValueError):
+        Options(endpoint="foo", rest_host="foo")
+
+    # REC1b1: endpoint with realtime_host
+    with pytest.raises(ValueError):
+        Options(endpoint="foo", realtime_host="foo")
+
+
+# REC1a
+def test_options_should_return_the_default_hostnames():
+    opts = Options()
+    assert opts.get_host() == "main.realtime.ably.net"
+    assert "main.a.fallback.ably-realtime.com" in opts.get_fallback_hosts()
+
+
+# REC1b4
+def test_options_should_return_the_correct_routing_policy_hostnames():
+    opts = Options(endpoint="foo")
+    assert opts.get_host() == "foo.realtime.ably.net"
+    assert "foo.a.fallback.ably-realtime.com" in opts.get_fallback_hosts()
+
+
+# REC1b3
+def test_options_should_return_the_correct_nonprod_routing_policy_hostnames():
+    opts = Options(endpoint="nonprod:foo")
+    assert opts.get_host() == "foo.realtime.ably-nonprod.net"
+    assert "foo.a.fallback.ably-realtime-nonprod.com" in opts.get_fallback_hosts()
+
+
+# REC1b2
+def test_options_should_return_the_correct_fqdn_hostnames():
+    opts = Options(endpoint="foo.com")
+    assert opts.get_host() == "foo.com"
+    assert not opts.get_fallback_hosts()
+
+
+# REC1b2
+def test_options_should_return_an_ipv4_address():
+    opts = Options(endpoint="127.0.0.1")
+    assert opts.get_host() == "127.0.0.1"
+    assert not opts.get_fallback_hosts()
+
+
+# REC1b2
+def test_options_should_return_an_ipv6_address():
+    opts = Options(endpoint="::1")
+    assert opts.get_host() == "::1"
+
+
+# REC1b2
+def test_options_should_return_localhost():
+    opts = Options(endpoint="localhost")
+    assert opts.get_host() == "localhost"
+    assert not opts.get_fallback_hosts()
+
+
+# REC1c1: environment with rest_host or realtime_host is invalid
+def test_options_should_fail_with_environment_and_rest_or_realtime_host():
+    # REC1c1: environment with rest_host
+    with pytest.raises(ValueError):
+        Options(environment="foo", rest_host="bar")
+
+    # REC1c1: environment with realtime_host
+    with pytest.raises(ValueError):
+        Options(environment="foo", realtime_host="bar")
+
+
+# REC1c2: environment defines production routing policy ID
+def test_options_with_environment_should_return_routing_policy_hostnames():
+    opts = Options(environment="foo")
+    # REC1c2: primary domain is [id].realtime.ably.net
+    assert opts.get_host() == "foo.realtime.ably.net"
+    # REC2c5: fallback domains for production routing policy ID via environment
+    assert "foo.a.fallback.ably-realtime.com" in opts.get_fallback_hosts()
+    assert "foo.e.fallback.ably-realtime.com" in opts.get_fallback_hosts()
+
+
+# REC1d1: rest_host takes precedence for primary domain
+def test_options_with_rest_host_should_return_rest_host():
+    opts = Options(rest_host="custom.example.com")
+    # REC1d1: primary domain is the value of the restHost option
+    assert opts.get_host() == "custom.example.com"
+    # REC2c6: fallback domains for restHost is empty
+    assert not opts.get_fallback_hosts()
+
+
+# REC1d2: realtime_host if rest_host not specified
+def test_options_with_realtime_host_should_return_realtime_host():
+    opts = Options(realtime_host="custom.example.com")
+    # REC1d2: primary domain is the value of the realtimeHost option
+    assert opts.get_host() == "custom.example.com"
+    # REC2c6: fallback domains for realtimeHost is empty
+    assert not opts.get_fallback_hosts()
+
+
+# REC1d1: rest_host takes precedence over realtime_host
+def test_options_with_rest_host_takes_precedence_over_realtime_host():
+    opts = Options(rest_host="rest.example.com", realtime_host="realtime.example.com")
+    # REC1d1: restHost takes precedence
+    assert opts.get_host() == "rest.example.com"
+    # REC2c6: fallback domains is empty
+    assert not opts.get_fallback_hosts()
+
+
+# REC2a2: fallback_hosts value is used when specified
+def test_options_with_fallback_hosts_should_use_specified_hosts():
+    custom_fallbacks = ["fallback1.example.com", "fallback2.example.com"]
+    opts = Options(fallback_hosts=custom_fallbacks)
+    # REC2a2: the set of fallback domains is given by the value of the fallbackHosts option
+    assert opts.get_fallback_hosts() == custom_fallbacks
+
+
+# REC2a2: empty fallback_hosts array is respected
+def test_options_with_empty_fallback_hosts_should_have_no_fallbacks():
+    opts = Options(fallback_hosts=[])
+    # REC2a2: empty array means no fallbacks
+    assert opts.get_fallback_hosts() == []
+
+
+# REC2c1: Default fallback hosts for main endpoint
+def test_options_default_fallback_hosts():
+    opts = Options()
+    fallbacks = opts.get_fallback_hosts()
+    # REC2c1: default fallback hosts
+    assert "main.a.fallback.ably-realtime.com" in fallbacks
+    assert "main.b.fallback.ably-realtime.com" in fallbacks
+    assert "main.c.fallback.ably-realtime.com" in fallbacks
+    assert "main.d.fallback.ably-realtime.com" in fallbacks
+    assert "main.e.fallback.ably-realtime.com" in fallbacks
+
+
+# REC2c3: Non-production routing policy fallback hosts
+def test_options_nonprod_fallback_hosts():
+    opts = Options(endpoint="nonprod:test")
+    fallbacks = opts.get_fallback_hosts()
+    # REC2c3: nonprod fallback hosts
+    assert "test.a.fallback.ably-realtime-nonprod.com" in fallbacks
+    assert "test.b.fallback.ably-realtime-nonprod.com" in fallbacks
+    assert "test.c.fallback.ably-realtime-nonprod.com" in fallbacks
+    assert "test.d.fallback.ably-realtime-nonprod.com" in fallbacks
+    assert "test.e.fallback.ably-realtime-nonprod.com" in fallbacks
+
+
+# REC2c4: Production routing policy fallback hosts
+def test_options_prod_routing_policy_fallback_hosts():
+    opts = Options(endpoint="custom")
+    fallbacks = opts.get_fallback_hosts()
+    # REC2c4: production routing policy fallback hosts
+    assert "custom.a.fallback.ably-realtime.com" in fallbacks
+    assert "custom.b.fallback.ably-realtime.com" in fallbacks
+    assert "custom.c.fallback.ably-realtime.com" in fallbacks
+    assert "custom.d.fallback.ably-realtime.com" in fallbacks
+    assert "custom.e.fallback.ably-realtime.com" in fallbacks
+
+
+# REC2c2: Explicit hostname (FQDN) has empty fallback hosts
+def test_options_fqdn_no_fallback_hosts():
+    opts = Options(endpoint="custom.example.com")
+    # REC2c2: explicit hostname has empty fallback
+    assert opts.get_fallback_hosts() == []
+
+
+# REC2c2: IPv6 address has empty fallback hosts
+def test_options_ipv6_no_fallback_hosts():
+    opts = Options(endpoint="::1")
+    # REC2c2: explicit hostname has empty fallback
+    assert opts.get_fallback_hosts() == []
+
+
+# REC2c2: localhost has empty fallback hosts
+def test_options_localhost_no_fallback_hosts():
+    opts = Options(endpoint="localhost")
+    # REC2c2: explicit hostname has empty fallback
+    assert opts.get_fallback_hosts() == []