From 224124afae3a66055fbed1d9183ebc21d6af2e81 Mon Sep 17 00:00:00 2001 From: Mike Fiedler Date: Wed, 22 Dec 2021 15:01:41 -0500 Subject: [PATCH] refactor/fix: move logic into setup phase and correct order (#88) --- pytest_socket.py | 49 ++++++++------ tests/test_precedence.py | 143 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 19 deletions(-) create mode 100644 tests/test_precedence.py diff --git a/pytest_socket.py b/pytest_socket.py index 2360eef..9da153f 100644 --- a/pytest_socket.py +++ b/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(): diff --git a/tests/test_precedence.py b/tests/test_precedence.py new file mode 100644 index 0000000..d3ddbb7 --- /dev/null +++ b/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)