From a906a496432ba1ed3d9e73864ee7d9f8f4181a67 Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:08:04 +0200 Subject: [PATCH] fix: count swallowed network calls at import --- tests/test_scraper_contract.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/test_scraper_contract.py b/tests/test_scraper_contract.py index ae42f580..16b38f8a 100644 --- a/tests/test_scraper_contract.py +++ b/tests/test_scraper_contract.py @@ -366,7 +366,9 @@ import importlib, socket, sys from pathlib import Path root, package = Path(sys.argv[1]), sys.argv[2] sys.path[:0] = [str(root), str(root / "scripts")] +attempts = [] def refuse(*_args, **_kwargs): + attempts.append(1) raise OSError("network refused at import") socket.socket.connect = refuse socket.create_connection = refuse @@ -375,10 +377,15 @@ for path in sorted((root / package.replace(".", "/")).rglob("*.py")): if path.name == "__init__.py": continue name = ".".join(path.relative_to(root).with_suffix("").parts) + before = len(attempts) try: importlib.import_module(name) except Exception as exc: print(f"{name}: {type(exc).__name__}: {exc}") + continue + # A fetch whose error the module swallows ("or 'master'") still tried. + if len(attempts) > before: + print(f"{name}: reached for the network") """ @@ -401,6 +408,25 @@ class ImportingAScraperFetchesNothing(unittest.TestCase): def test_every_scraper_imports_offline(self): self.assertEqual(self._offenders(REPO_ROOT, "scripts.scraper"), "") + def test_a_swallowed_import_time_fetch_is_caught(self): + with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as directory: + package = Path(directory) / "planted" + package.mkdir() + (package / "__init__.py").write_text("") + (package / "quiet_scraper.py").write_text( + "import urllib.request\n" + "def _tag():\n" + " try:\n" + " return urllib.request.urlopen('https://api.github.com/x').read()\n" + " except OSError:\n" + " return None\n" + "TAG = _tag() or 'master'\n" + ) + self.assertIn( + "planted.quiet_scraper: reached for the network", + self._offenders(Path(directory), "planted"), + ) + def test_an_import_time_fetch_is_caught(self): with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as directory: package = Path(directory) / "planted"