test: fail on unexpected network access

CONTRIBUTING.md

@@ -196,6 +196,11 @@ except FileNotFoundError as err:
 
 ## Testing
 
+Tests block network sockets by default. Mark tests that need network access with
+`pytest.mark.network` and run them with `--with-network`. Unexpected socket
+attempts fail the session even when the application catches the blocking error.
+Unix sockets remain available for async tests.
+
 ### Structure
 
 - Place tests under `tests/`.

pyproject.toml

@@ -60,7 +60,8 @@ build = [
 ]
 mypy = [
     "mypy",
-    "pytest",
+    "pytest>=8.4",
+    "pytest-socket>=0.8.0",
     "types-html5lib",
     "types-psutil",
     "types-PyYAML",
@@ -70,8 +71,9 @@ mypy = [
 ]
 test = [
     "coverage[toml]!=4.4,>=4.0",
-    "pytest",
+    "pytest>=8.4",
     "pytest-cov",
+    "pytest-socket>=0.8.0",
     "pytest-xdist",
     "requests-mock",
     "spdx-tools",
@@ -295,7 +297,10 @@ build = "sphinx-build -M html docs docs/_build -j auto --keep-going {args:--fail
 
 [tool.pytest.ini_options]
 # run tests in parallel with 3 pytest-xdist workers
-addopts = "-n 3"
+addopts = "-n 3 --disable-socket --allow-unix-socket"
+filterwarnings = [
+    'always::UserWarning:pytest_socket$',
+]
 testpaths = ["tests"]
 markers = [
     "network: test that need network access",

tests/conftest.py

@@ -1,14 +1,17 @@
 import pathlib
 import typing
+import warnings
 
 import pytest
+import pytest_socket
 from click.testing import CliRunner
 
 from fromager import context, packagesettings
 from fromager.packagesettings import SbomSettings
 
 TESTDATA_PATH = pathlib.Path(__file__).parent.absolute() / "testdata"
 E2E_PATH = pathlib.Path(__file__).parent.parent.absolute() / "e2e"
+_blocked_network_attempts: set[tuple[str, str]] = set()
 
 
 def pytest_addoption(parser: pytest.Parser) -> None:
@@ -23,12 +26,44 @@ def pytest_addoption(parser: pytest.Parser) -> None:
 def pytest_collection_modifyitems(
     config: pytest.Config, items: list[pytest.Item]
 ) -> None:
-    if config.getoption("--with-network"):
-        return
-    skip_network = pytest.mark.skip(reason="need --with-network option to run")
+    network_marker = (
+        pytest.mark.enable_socket
+        if config.getoption("--with-network")
+        else pytest.mark.skip(reason="need --with-network option to run")
+    )
     for item in items:
         if "network" in item.keywords:
-            item.add_marker(skip_network)
+            item.add_marker(network_marker)
+
+
+def pytest_warning_recorded(
+    warning_message: warnings.WarningMessage,
+    when: str,
+    nodeid: str,
+    location: tuple[str, int, str] | None,
+) -> None:
+    """Record blocked socket attempts."""
+    # pytest-socket has no warning category; assume warnings come from its module.
+    if warning_message.filename == pytest_socket.__file__:
+        _blocked_network_attempts.add((nodeid, str(warning_message.message)))
+
+
+def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
+    """Fail on blocked requests even when their exceptions were caught."""
+    # Caught socket errors and ignored background futures can leave tests passing.
+    # Fail the session if any blocked attempts were recorded.
+    if _blocked_network_attempts and exitstatus == pytest.ExitCode.OK:
+        session.exitstatus = pytest.ExitCode.TESTS_FAILED
+
+
+def pytest_terminal_summary(terminalreporter: pytest.TerminalReporter) -> None:
+    """List tests that attempted unexpected network access."""
+    if not _blocked_network_attempts:
+        return
+
+    terminalreporter.write_sep("=", "unexpected network attempts")
+    for nodeid, message in sorted(_blocked_network_attempts):
+        terminalreporter.write_line(f"{nodeid}: {message}")
 
 
 @pytest.fixture

tests/test_bootstrapper.py

@@ -1000,42 +1000,29 @@ def test_bootstrap_with_single_requirement(tmp_context: WorkContext) -> None:
     """bootstrap([req]) resolves and processes the requirement."""
     bt = bootstrapper.Bootstrapper(tmp_context)
     req = Requirement("testpkg==1.0")
-    captured: list[Resolve] = []
 
-    def capture_run(self: Resolve, bt_arg: bootstrapper.Bootstrapper) -> list[Phase]:
-        captured.append(self)
-        return []
-
-    with (
-        patch.object(
-            bt,
-            "_resolve_and_add_top_level",
-            return_value=("http://example.test/testpkg-1.0.tar.gz", Version("1.0")),
-        ),
-        patch.object(Resolve, "run", capture_run),
-        patch.object(bt, "_record_stack_state"),
-    ):
-        bt.bootstrap([req])
+    started = _bootstrap_and_record_started(
+        bt,
+        [req],
+        {str(req): [("https://pkg.test/testpkg-1.0.tar.gz", Version("1.0"))]},
+    )
 
-    assert len(captured) == 1
-    assert isinstance(captured[0], Resolve)
-    assert captured[0].work_item.req == req
-    assert captured[0].work_item.req_type == RequirementType.TOP_LEVEL
+    assert started == {"testpkg==1.0"}
+    assert set(tmp_context.dependency_graph.nodes) == {ROOT, "testpkg==1.0"}
 
 
 def test_bootstrap_skips_failed_resolution(tmp_context: WorkContext) -> None:
-    """bootstrap() skips requirements whose resolution returns None."""
-    bt = bootstrapper.Bootstrapper(tmp_context)
+    """bootstrap() records unresolved requirements and skips processing them."""
+    bt = bootstrapper.Bootstrapper(tmp_context, test_mode=True)
     req = Requirement("badpkg")
 
-    with (
-        patch.object(bt, "_resolve_and_add_top_level", return_value=None),
-        patch.object(Resolve, "run") as mock_run,
-        patch.object(bt, "_record_stack_state"),
-    ):
-        bt.bootstrap([req])
+    started = _bootstrap_and_record_started(bt, [req], {str(req): []})
 
-    mock_run.assert_not_called()
+    assert started == set()
+    assert set(tmp_context.dependency_graph.nodes) == {ROOT}
+    assert len(bt.failed_packages) == 1
+    assert bt.failed_packages[0]["package"] == "badpkg"
+    assert bt.failed_packages[0]["failure_type"] == "resolution"
 
 
 def test_bootstrap_two_requirements_both_processed(tmp_context: WorkContext) -> None:
@@ -1044,24 +1031,17 @@ def test_bootstrap_two_requirements_both_processed(tmp_context: WorkContext) ->
     req1 = Requirement("pkg1==1.0")
     req2 = Requirement("pkg2==2.0")
 
-    dispatch_calls: list = []
-
-    def capture_run(self: Resolve, bt_arg: bootstrapper.Bootstrapper) -> list[Phase]:
-        dispatch_calls.append(self.work_item.req.name)
-        return []
-
-    with (
-        patch.object(
-            bt,
-            "_resolve_and_add_top_level",
-            return_value=("http://example.test/pkg-1.0.tar.gz", Version("1.0")),
-        ),
-        patch.object(Resolve, "run", capture_run),
-        patch.object(bt, "_record_stack_state"),
-    ):
-        bt.bootstrap([req1, req2])
+    started = _bootstrap_and_record_started(
+        bt,
+        [req1, req2],
+        {
+            str(req1): [("https://pkg.test/pkg1-1.0.tar.gz", Version("1.0"))],
+            str(req2): [("https://pkg.test/pkg2-2.0.tar.gz", Version("2.0"))],
+        },
+    )
 
-    assert sorted(dispatch_calls) == ["pkg1", "pkg2"]
+    assert started == {"pkg1==1.0", "pkg2==2.0"}
+    assert set(tmp_context.dependency_graph.nodes) == {ROOT, *started}
 
 
 def _bootstrap_and_record_started(