Skip to content

Use endpoint as default connection option (ADR-119) #590

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Use endpoint as default connection option (ADR-119) #590

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a6b8bd5
Add endpoint option
surminus Feb 17, 2025
ce562e7
fixup! Add endpoint option
surminus Mar 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions ably/realtime/realtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
10 changes: 8 additions & 2 deletions ably/rest/rest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,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
Expand Down
48 changes: 31 additions & 17 deletions ably/transport/defaults.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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",
]
49 changes: 23 additions & 26 deletions ably/types/options.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,15 +9,20 @@

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=False, 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=False, 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,
Comment on lines +12 to +17
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent

🧩 Analysis chain

Add deprecation warnings for legacy options.

Since environment, rest_host, and realtime_host are being deprecated in favor of endpoint, consider adding deprecation warnings when these parameters are used without endpoint.

Adding deprecation warnings will help users migrate to the new API:

    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=False, 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, **kwargs):

        super().__init__(**kwargs)

+        # Issue deprecation warnings for legacy options
+        import warnings
+        if environment is not None and endpoint is None:
+            warnings.warn(
+                "The 'environment' parameter is deprecated, use 'endpoint' instead.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+        if rest_host is not None:
+            warnings.warn(
+                "The 'rest_host' parameter is deprecated, use 'endpoint' instead.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+        if realtime_host is not None:
+            warnings.warn(
+                "The 'realtime_host' parameter is deprecated, use 'endpoint' instead.",
+                DeprecationWarning,
+                stacklevel=2
+            )

🌐 Web query:

How to add deprecation warnings in Python?

💡 Result:

To add deprecation warnings in Python, you can use the built-in warnings module. This allows you to notify users of your code that certain features, functions, or classes are deprecated and may be removed in future versions. Here's a simple example of how to do this:

import warnings

def deprecated_function():
    warnings.warn(
        "deprecated_function() is deprecated and will be removed in a future version.",
        DeprecationWarning,
        stacklevel=2
    )
    # Function implementation here
    print("This function is deprecated.")

# Usage
deprecated_function()

Explanation:

  • warnings.warn() is used to issue a warning.
  • The first argument is the warning message.
  • The second argument specifies the warning category, DeprecationWarning in this case.
  • stacklevel=2 adjusts the traceback to point to the caller's location instead of inside the warning function itself, which makes it clearer for the user.

By default, DeprecationWarning is ignored in normal Python scripts but is shown in interactive environments and during tests. If you want to ensure the warning is always shown, you can enable it explicitly in your script or test environment using:

import warnings
warnings.simplefilter('always', DeprecationWarning)

Let me know if you want me to provide an example for a deprecated class or decorator-style usage as well!

🏁 Script executed:

#!/bin/bash
set -e

# Verify presence of legacy parameters in the signature
rg -n "def __init__" ably/types/options.py
rg -n "rest_host" ably/types/options.py
rg -n "realtime_host" ably/types/options.py

Length of output: 2741

Add deprecation warnings for legacy options in init

In ably/types/options.py, inside the __init__ method—immediately after the

super().__init__(**kwargs)

call—insert warnings for any use of environment, rest_host, or realtime_host to help users migrate to endpoint. For example:

 class Options:
     def __init__(self, client_id=None, log_level=0, tls=True,
-                 rest_host=None, realtime_host=None, port=0,
+                 rest_host=None, realtime_host=None, port=0,
                  tls_port=0, use_binary_protocol=True, queue_messages=False,
                  recover=False, endpoint=None, environment=None,
                  http_open_timeout=None, …, **kwargs):
 
-        super().__init__(**kwargs)
+        super().__init__(**kwargs)
+
+        # Deprecation warnings for legacy connection parameters
+        import warnings
+        if environment is not None:
+            warnings.warn(
+                "The ‘environment’ parameter is deprecated; please use ‘endpoint’ instead.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+        if rest_host is not None:
+            warnings.warn(
+                "The ‘rest_host’ parameter is deprecated; please use ‘endpoint’ instead.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+        if realtime_host is not None:
+            warnings.warn(
+                "The ‘realtime_host’ parameter is deprecated; please use ‘endpoint’ instead.",
+                DeprecationWarning,
+                stacklevel=2
+            )
 
         # existing validation logic follows…
  • File: ably/types/options.py
  • Location: immediately after super().__init__(**kwargs) in the __init__ method

This will surface DeprecationWarnings (shown in interactive/test environments by default) and guide users toward the new endpoint parameter.

🤖 Prompt for AI Agents 
In ably/types/options.py around lines 12 to 17, inside the __init__ method,
immediately after the call to super().__init__(**kwargs), add deprecation
warnings using the warnings module for the legacy parameters environment,
rest_host, and realtime_host when they are used without the new endpoint
parameter. Use warnings.warn() with DeprecationWarning and stacklevel=2 to
notify users these options are deprecated and to encourage migration to
endpoint.

channel_retry_timeout=Defaults.channel_retry_timeout, add_request_ids=False, **kwargs):

super().__init__(**kwargs)

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')

Comment on lines +22 to +25
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Simplify validation condition for incompatible options.

The nested if statement can be simplified as suggested by the static analysis tool. Also, consider checking for all incompatible options at once.

-        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')
+        if endpoint is not None and (
+            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')
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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')
if endpoint is not None and (
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')
🧰 Tools
🪛 Ruff (0.11.9)

22-23: Use a single if statement instead of nested if statements

(SIM102)

🤖 Prompt for AI Agents 
In ably/types/options.py around lines 22 to 25, simplify the nested if statement
that checks for incompatible options by combining the conditions into a single
if statement using a logical OR to check if any of environment, rest_host, or
realtime_host are not None when endpoint is set. This will make the validation
clearer and more concise.

# TODO check these defaults
if fallback_retry_timeout is None:
fallback_retry_timeout = Defaults.fallback_retry_timeout
Expand Down Expand Up @@ -46,8 +51,11 @@ 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:
endpoint = environment

if endpoint is None:
endpoint = Defaults.endpoint

self.__client_id = client_id
self.__log_level = log_level
Expand All @@ -59,7 +67,7 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti
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
Expand Down Expand Up @@ -163,8 +171,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):
Expand Down Expand Up @@ -268,27 +276,19 @@ def __get_rest_hosts(self):
# Defaults
host = self.rest_host
if host is None:
host = Defaults.rest_host

environment = self.environment
host = Defaults.get_hostname(self.endpoint)

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 = '%s-%s' % (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:
if self.rest_host is not None:
fallback_hosts = []
else:
fallback_hosts = Defaults.get_fallback_hosts(self.endpoint)

# Shuffle
fallback_hosts = list(fallback_hosts)
Expand All @@ -304,11 +304,8 @@ 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

host = Defaults.get_hostname(self.endpoint)
return [host] + self.__fallback_hosts

def get_rest_hosts(self):
Expand Down
28 changes: 17 additions & 11 deletions test/ably/rest/restinit_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,15 +73,15 @@ def test_rest_host_and_environment(self):
ably = AblyRest(token='foo', rest_host="some.other.host")
assert "some.other.host" == ably.options.rest_host, "Unexpected host mismatch"

# environment: production
ably = AblyRest(token='foo', environment="production")
# environment: main
ably = AblyRest(token='foo', environment="main")
host = ably.options.get_rest_host()
assert "rest.ably.io" == host, "Unexpected host mismatch %s" % host
assert "main.realtime.ably.net" == host, "Unexpected host mismatch %s" % host

# environment: other
ably = AblyRest(token='foo', environment="sandbox")
ably = AblyRest(token='foo', environment="nonprod:sandbox")
host = ably.options.get_rest_host()
assert "sandbox-rest.ably.io" == host, "Unexpected host mismatch %s" % host
assert "sandbox.realtime.ably-nonprod.net" == host, "Unexpected host mismatch %s" % host

# both, as per #TO3k2
with pytest.raises(ValueError):
Expand All @@ -103,13 +103,13 @@ def test_fallback_hosts(self):
assert sorted(aux) == sorted(ably.options.get_fallback_rest_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 = AblyRest(token='foo', environment='nonprod:sandbox', http_max_retry_count=10)
assert sorted(Defaults.get_fallback_hosts('nonprod:sandbox')) == sorted(
ably.options.get_fallback_rest_hosts())

# Fallback hosts and environment 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_rest_hosts())

# RSC15f
ably = AblyRest(token='foo')
Expand Down Expand Up @@ -182,13 +182,19 @@ 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' == '{0}://{1}'.format(ably.http.preferred_scheme, ably.http.preferred_host)
assert 'https://main.realtime.ably.net' == '{0}://{1}'.format(
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' == '{0}://{1}'.format(ably.http.preferred_scheme, ably.http.preferred_host)
assert 'http://main.realtime.ably.net' == '{0}://{1}'.format(
ably.http.preferred_scheme, ably.http.preferred_host
)

assert ably.http.preferred_port == 80

@dont_vary_protocol
Expand All @@ -211,7 +217,7 @@ async def test_environment(self):
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()

Expand Down
10 changes: 5 additions & 5 deletions test/ably/rest/restpaginatedresult_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ async def asyncSetUp(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'},
Expand All @@ -44,8 +44,8 @@ async def asyncSetUp(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
Expand All @@ -55,11 +55,11 @@ async def asyncSetUp(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())

async def asyncTearDown(self):
Expand Down
4 changes: 2 additions & 2 deletions test/ably/rest/restrequest_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,8 +101,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:
Expand Down
8 changes: 4 additions & 4 deletions test/ably/testapp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,15 +14,15 @@
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')
rest_host = os.environ.get('ABLY_REST_HOST', 'sandbox.realtime.ably-nonprod.net')
realtime_host = os.environ.get('ABLY_REALTIME_HOST', 'sandbox.realtime.ably-nonprod.net')

environment = os.environ.get('ABLY_ENV', 'sandbox')
environment = os.environ.get('ABLY_ENV', 'nonprod:sandbox')

port = 80
tls_port = 443

if rest_host and not rest_host.endswith("rest.ably.io"):
if rest_host and not rest_host.endswith("realtime.ably-nonprod.net"):
tls = tls and rest_host != "localhost"
port = 8080
tls_port = 8081
Expand Down
61 changes: 61 additions & 0 deletions test/unit/options_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
import pytest

from ably.types.options import Options


def test_options_should_fail_early_with_incompatible_client_options():
with pytest.raises(ValueError):
Options(endpoint="foo", environment="foo")

with pytest.raises(ValueError):
Options(endpoint="foo", rest_host="foo")

with pytest.raises(ValueError):
Options(endpoint="foo", realtime_host="foo")


# REC1a
def test_options_should_return_the_default_hostnames():
opts = Options()
assert opts.get_realtime_host() == "main.realtime.ably.net"
assert "main.a.fallback.ably-realtime.com" in opts.get_fallback_realtime_hosts()


# REC1b4
def test_options_should_return_the_correct_routing_policy_hostnames():
opts = Options(endpoint="foo")
assert opts.get_realtime_host() == "foo.realtime.ably.net"
assert "foo.a.fallback.ably-realtime.com" in opts.get_fallback_realtime_hosts()


# REC1b3
def test_options_should_return_the_correct_nonprod_routing_policy_hostnames():
opts = Options(endpoint="nonprod:foo")
assert opts.get_realtime_host() == "foo.realtime.ably-nonprod.net"
assert "foo.a.fallback.ably-realtime-nonprod.com" in opts.get_fallback_realtime_hosts()


# REC1b2
def test_options_should_return_the_correct_fqdn_hostnames():
opts = Options(endpoint="foo.com")
assert opts.get_realtime_host() == "foo.com"
assert not opts.get_fallback_realtime_hosts()


# REC1b2
def test_options_should_return_an_ipv4_address():
opts = Options(endpoint="127.0.0.1")
assert opts.get_realtime_host() == "127.0.0.1"
assert not opts.get_fallback_realtime_hosts()


# REC1b2
def test_options_should_return_an_ipv6_address():
opts = Options(endpoint="::1")
assert opts.get_realtime_host() == "::1"


# REC1b2
def test_options_should_return_localhost():
opts = Options(endpoint="localhost")
assert opts.get_realtime_host() == "localhost"
Loading