Test HostUtil._get_host_info() using reference implementation of socket module

Author: MetRonnie

tests/unit/test_hostuserutil.py

@@ -18,6 +18,9 @@
 import re
 from secrets import token_hex
 import socket
+from types import SimpleNamespace
+from typing import Optional
+from unittest.mock import Mock
 
 import pytest
 
@@ -36,6 +39,92 @@
 LOCALHOST_ALIASES = socket.gethostbyname_ex('localhost')[1]
 
 
+@pytest.fixture
+def mock_socket(monkeypatch: pytest.MonkeyPatch):
+    """Reference implementation of socket functions, with some of their real
+    observed quirks.
+
+    Yes, it is horribly quirky. It is based on one linux system, and other
+    systems or different /etc/hosts setup will result in different behaviour.
+    """
+    this_name = 'NCC1701'
+    this_fqdn = f'{this_name}.starfleet.gov'
+    this_ip = '12.345.67.89'
+    this_ex = (this_fqdn, [this_name], [this_ip])
+
+    localhost = 'localhost'
+    localhost_fqdn = f'{localhost}.localdomain'
+    localhost_ip = '127.0.0.1'
+    localhost_aliases_v4 = [
+        localhost_fqdn,
+        f'{localhost}4',
+        f'{localhost}4.localdomain4',
+    ]
+    localhost_aliases_v6 = [
+        localhost_fqdn,
+        f'{localhost}6',
+        f'{localhost}6.localdomain6',
+    ]
+    localhost_ex = (
+        localhost,
+        [*localhost_aliases_v4, *localhost_aliases_v6],
+        [localhost_ip, localhost_ip],
+    )
+
+    def _getfqdn(x: Optional[str] = None):
+        if x:
+            x = x.lower()
+        if not x or this_fqdn.lower().startswith(x) or x == this_ip:
+            return this_fqdn
+        if x in {localhost, localhost_fqdn, localhost_ip}:
+            return localhost_fqdn
+        return x
+
+    def _gethostbyaddr(x: str):
+        x = x.lower()
+        if this_fqdn.lower().startswith(x) or x == this_ip:
+            return this_ex
+        if x in {localhost, localhost_fqdn, '::1', *localhost_aliases_v6}:
+            return (localhost, localhost_aliases_v6, ['::1'])
+        if x in {localhost_ip, *localhost_aliases_v4}:
+            return (localhost, localhost_aliases_v4, [localhost_ip])
+        raise socket.gaierror("oopsie")
+
+    def _gethostbyname_ex(x: str):
+        x = x.lower()
+        if x in {this_fqdn.lower(), this_name.lower()}:
+            return this_ex
+        if this_fqdn.lower().startswith(x):
+            return (this_fqdn, [], [this_ip])
+        if x in {localhost, localhost_fqdn}:
+            return localhost_ex
+        if x in localhost_aliases_v6:
+            return (localhost, localhost_aliases_v6, ['::1'])
+        if x in localhost_aliases_v4:
+            return (localhost, localhost_aliases_v4, [localhost_ip])
+        raise socket.gaierror("oopsie")
+
+    mock_getfqdn = Mock(side_effect=_getfqdn)
+    monkeypatch.setattr('cylc.flow.hostuserutil.socket.getfqdn', mock_getfqdn)
+    mock_gethostbyaddr = Mock(side_effect=_gethostbyaddr)
+    monkeypatch.setattr(
+        'cylc.flow.hostuserutil.socket.gethostbyaddr', mock_gethostbyaddr
+    )
+    mock_gethostbyname_ex = Mock(side_effect=_gethostbyname_ex)
+    monkeypatch.setattr(
+        'cylc.flow.hostuserutil.socket.gethostbyname_ex', mock_gethostbyname_ex
+    )
+    return SimpleNamespace(
+        this_fqdn=this_fqdn,
+        this_ip=this_ip,
+        this_ex=this_ex,
+        localhost_ex=localhost_ex,
+        getfqdn=mock_getfqdn,
+        gethostbyaddr=mock_gethostbyaddr,
+        gethostbyname_ex=mock_gethostbyname_ex,
+    )
+
+
 def test_is_remote_user_on_current_user():
     """is_remote_user with current user."""
     assert not is_remote_user(None)
@@ -101,3 +190,30 @@ def test_get_host_info__basic():
     with pytest.raises(IOError) as exc:
         hu._get_host_info(bad_host)
     assert bad_host in str(exc.value)
+
+
+def test_get_host_info__advanced(mock_socket):
+    hu = HostUtil(expire=3600)
+    assert mock_socket.gethostbyname_ex.call_count == 0
+    assert hu._get_host_info() == mock_socket.this_ex
+    assert mock_socket.gethostbyname_ex.call_count == 1
+    # Test caching:
+    hu._get_host_info()
+    assert mock_socket.gethostbyname_ex.call_count == 1
+    # Test variations of host name:
+    assert hu._get_host_info('NCC1701') == mock_socket.this_ex
+    assert hu._get_host_info('ncc1701.starfleet') == mock_socket.this_ex
+    # (Note:)
+    assert (
+        mock_socket.gethostbyname_ex('ncc1701.starfleet')
+        != mock_socket.this_ex
+    )
+    assert hu._get_host_info('localhost4') == mock_socket.localhost_ex
+    assert hu._get_host_info('localhost6') == mock_socket.localhost_ex
+    # Test IP address:
+    assert hu._get_host_info(mock_socket.this_ip) == mock_socket.this_ex
+    assert hu._get_host_info('127.0.0.1') == mock_socket.localhost_ex
+    # Test error:
+    with pytest.raises(IOError):
+        hu._get_host_info('nonexist')
+    assert 'nonexist' not in hu._host_exs