refactor/fix: move logic into setup phase and correct order

pytest_socket.py

@@ -42,24 +42,6 @@ def pytest_addoption(parser):
     )
 
 
-@pytest.fixture(autouse=True)
-def _socket_marker(request):
-    """
-    Create an automatically-used fixture that every test function will load.
-
-    The last option to set the fixture wins priority.
-    The expected behavior is that higher granularity options should override
-    lower granularity options.
-    """
-    if request.config.__socket_disabled:
-        request.getfixturevalue('socket_disabled')
-
-    if request.node.get_closest_marker('disable_socket'):
-        request.getfixturevalue('socket_disabled')
-    if request.node.get_closest_marker('enable_socket'):
-        request.getfixturevalue('socket_enabled')
-
-
 @pytest.fixture
 def socket_disabled(pytestconfig):
     """ disable socket.socket for duration of this test function """
@@ -112,7 +94,35 @@ def pytest_configure(config):
     config.__socket_allow_hosts = config.getoption('--allow-hosts')
 
 
-def pytest_runtest_setup(item):
+def pytest_runtest_setup(item) -> None:
+    """During each test item's setup phase,
+    choose the behavior based on the configurations supplied.
+
+    This is the bulk of the logic for the plugin.
+    As the logic can be extensive, this method is allowed complexity.
+    It may be refactored in the future to be more readable.
+    """
+
+    # If test has the `enable_socket` marker, we accept this as most explicit.
+    if 'socket_enabled' in item.fixturenames or item.get_closest_marker('enable_socket'):
+        enable_socket()
+        return
+
+    # If the test has the `disable_socket` marker, it's explicitly disabled.
+    if 'socket_disabled' in item.fixturenames or item.get_closest_marker('disable_socket'):
+        disable_socket(item.config.__socket_allow_unix_socket)
+        return
+
+    # Resolve `allow_hosts` behaviors.
+    hosts = _resolve_allow_hosts(item)
+
+    # Finally, check the global config and disable socket if needed.
+    if item.config.__socket_disabled and not hosts:
+        disable_socket(item.config.__socket_allow_unix_socket)
+
+
+def _resolve_allow_hosts(item):
+    """Resolve `allow_hosts` behaviors."""
     mark_restrictions = item.get_closest_marker('allow_hosts')
     cli_restrictions = item.config.__socket_allow_hosts
     hosts = None
@@ -121,6 +131,7 @@ def pytest_runtest_setup(item):
     elif cli_restrictions:
         hosts = cli_restrictions
     socket_allow_hosts(hosts)
+    return hosts
 
 
 def pytest_runtest_teardown():

tests/test_precedence.py

@@ -0,0 +1,143 @@
+"""Test module to express precedence tests between the different
+configuration combinations"""
+
+
+def assert_socket_blocked(result, passed=0, skipped=0, failed=1):
+    """Uses built in methods to test for common failure scenarios.
+    Usually we only test for a single failure,
+    but sometimes we want to test for multiple conditions,
+    so we can pass in the expected counts."""
+    result.assert_outcomes(passed=passed, skipped=skipped, failed=failed)
+    result.stdout.fnmatch_lines(
+        "*Socket*Blocked*Error: A test tried to use socket.socket.*"
+    )
+
+
+def test_disable_via_fixture(testdir):
+    testdir.makepyfile(
+        """
+        import socket
+
+        def test_socket(socket_disabled):
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        """
+    )
+    result = testdir.runpytest()
+    assert_socket_blocked(result)
+
+
+def test_disable_via_marker(testdir):
+    testdir.makepyfile(
+        """
+        import socket
+        import pytest
+
+        @pytest.mark.disable_socket
+        def test_socket():
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        """
+    )
+    result = testdir.runpytest()
+    assert_socket_blocked(result)
+
+
+def test_global_disable_via_cli_flag(testdir):
+    testdir.makepyfile(
+        """
+        import socket
+
+        def test_socket():
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        """
+    )
+    result = testdir.runpytest("--disable-socket")
+    assert_socket_blocked(result)
+
+
+def test_global_disable_via_config(testdir):
+    testdir.makepyfile(
+        """
+        import socket
+
+        def test_socket():
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        """
+    )
+    testdir.makeini(
+        """
+        [pytest]
+        addopts = --disable-socket
+        """
+    )
+    result = testdir.runpytest()
+    assert_socket_blocked(result)
+
+
+def test_enable_via_fixture(testdir):
+    testdir.makepyfile(
+        """
+        import socket
+
+        def test_socket(socket_enabled):
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+        def test_socket_disabled():
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        """
+    )
+    result = testdir.runpytest("--disable-socket")
+    assert_socket_blocked(result, passed=1, failed=1)
+
+
+def test_enable_via_marker(testdir):
+    testdir.makepyfile(
+        """
+        import socket
+        import pytest
+
+        @pytest.mark.enable_socket
+        def test_socket():
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+        def test_socket_disabled():
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        """
+    )
+    result = testdir.runpytest("--disable-socket")
+    assert_socket_blocked(result, passed=1, failed=1)
+
+
+def test_enable_marker_for_test_class(testdir):
+    testdir.makepyfile(
+        """
+        import socket
+        import pytest
+
+        @pytest.mark.enable_socket
+        class TestSocket:
+            def test_socket(self):
+                socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+        def test_socket_disabled():
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        """
+    )
+    result = testdir.runpytest("--disable-socket")
+    assert_socket_blocked(result, passed=1, failed=1)
+
+
+def test_global_disable_and_allow_host(testdir, httpbin):
+    """Disable socket globally, but allow a specific host"""
+    testdir.makepyfile(
+        f"""
+        from urllib.request import urlopen
+
+        def test_urlopen():
+            assert urlopen("{httpbin.url}/")
+
+        def test_urlopen_disabled():
+            assert urlopen("https://google.com/")
+        """
+    )
+    result = testdir.runpytest("--disable-socket", f"--allow-hosts={httpbin.host}")
+    assert_socket_blocked(result, passed=1, failed=1)