ankn
a5f8fb49ab
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)
Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
"""Global pytest configuration.
|
|
|
|
This repository's unit/integration tests must be deterministic and must not make
|
|
any outbound network calls.
|
|
|
|
As a safety net, this conftest blocks common Python networking primitives by
|
|
default.
|
|
|
|
If you *intentionally* need network access for a specific test (generally not
|
|
recommended), you can opt out with:
|
|
|
|
@pytest.mark.allow_network
|
|
|
|
or for an entire module:
|
|
|
|
pytestmark = pytest.mark.allow_network
|
|
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import socket
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
|
|
def _blocked(*args: Any, **kwargs: Any) -> None: # pragma: no cover
|
|
raise AssertionError(
|
|
"Outbound network access is blocked during tests. "
|
|
"Patch the relevant client call, or mark the test with "
|
|
"@pytest.mark.allow_network if this is truly required."
|
|
)
|
|
|
|
|
|
def _blocked_connect(self: socket.socket, address: Any) -> None: # pragma: no cover
|
|
# Allow local-only connections needed for event loops / self-pipes.
|
|
try:
|
|
host = address[0] if isinstance(address, tuple) and address else None
|
|
except Exception:
|
|
host = None
|
|
|
|
if host in ("127.0.0.1", "localhost", "::1"):
|
|
return _ORIG_SOCKET_CONNECT(self, address)
|
|
|
|
raise AssertionError(
|
|
"Outbound network access is blocked during tests. "
|
|
"Patch the relevant client call, or mark the test with "
|
|
"@pytest.mark.allow_network if this is truly required."
|
|
)
|
|
|
|
|
|
_ORIG_SOCKET_CONNECT = socket.socket.connect
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _block_network(
|
|
monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest
|
|
) -> None:
|
|
"""Block outbound network unless explicitly allowed."""
|
|
|
|
if request.node.get_closest_marker("allow_network") is not None:
|
|
return
|
|
|
|
# Keep event-loop internals working (they rely on AF_UNIX socketpairs).
|
|
# Only block outbound connections.
|
|
monkeypatch.setattr(socket.socket, "connect", _blocked_connect, raising=True)
|
|
monkeypatch.setattr(socket, "create_connection", _blocked)
|
|
|
|
# Best-effort blocks for popular HTTP clients.
|
|
try: # urllib3 (requests)
|
|
import urllib3.connectionpool # type: ignore
|
|
|
|
monkeypatch.setattr(
|
|
urllib3.connectionpool.HTTPConnectionPool, "_new_conn", _blocked
|
|
)
|
|
monkeypatch.setattr(
|
|
urllib3.connectionpool.HTTPSConnectionPool, "_new_conn", _blocked
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
try: # httpx
|
|
import httpx # type: ignore
|
|
|
|
monkeypatch.setattr(httpx.Client, "send", _blocked)
|
|
monkeypatch.setattr(httpx.AsyncClient, "send", _blocked)
|
|
except Exception:
|
|
pass
|
|
|
|
try: # aiohttp
|
|
import aiohttp # type: ignore
|
|
|
|
monkeypatch.setattr(aiohttp.ClientSession, "_request", _blocked)
|
|
except Exception:
|
|
pass
|