diff --git a/tests/live/conftest.py b/tests/live/conftest.py index 48b5e78..7d23aad 100644 --- a/tests/live/conftest.py +++ b/tests/live/conftest.py @@ -30,6 +30,13 @@ _PYTHON = str(_VENV_PYTHON) if _VENV_PYTHON.exists() else sys.executable # Use search (not match) so lines prefixed by Twisted timestamps are handled. _RFC5424_RE = re.compile(r"<\d+>1 \S+ \S+ \S+ - \S+ ") +def _mysql_available() -> bool: + try: + s = socket.create_connection(("127.0.0.1", 3307), timeout=1) + s.close() + return True + except OSError: + return False def _free_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: diff --git a/tests/live/test_mysql_backend_live.py b/tests/live/test_mysql_backend_live.py index 5df3773..7a3a388 100644 --- a/tests/live/test_mysql_backend_live.py +++ b/tests/live/test_mysql_backend_live.py @@ -31,9 +31,9 @@ import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine +from tests.live.conftest import _mysql_available from decnet.web.db.mysql.repository import MySQLRepository - LIVE_URL = "mysql+asyncmy://root:root@127.0.0.1:3307/decnet" pytestmark = [ @@ -42,6 +42,10 @@ pytestmark = [ not (LIVE_URL and LIVE_URL.startswith("mysql")), reason="Set DECNET_DB_URL=mysql+aiomysql://... to run MySQL live tests", ), + pytest.mark.skipif( + not _mysql_available(), + reason="MySQL not available on 127.0.0.1:3307" + ) ] @@ -63,7 +67,7 @@ def _url_with_db(server_url: str, db_name: str) -> str: return urlunparse(parsed._replace(path=f"/{db_name}")) -@pytest.fixture(scope="session") +@pytest.fixture(scope="module") async def mysql_test_db_url(): """Create a per-worker throwaway database, yield its URL, drop it on teardown. diff --git a/tests/live/test_mysql_live.py b/tests/live/test_mysql_live.py index d42f1ae..e5c5c30 100644 --- a/tests/live/test_mysql_live.py +++ b/tests/live/test_mysql_live.py @@ -1,8 +1,12 @@ import pytest import pymysql -from tests.live.conftest import assert_rfc5424 +from tests.live.conftest import assert_rfc5424, _mysql_available +pytestmark = pytest.mark.skipif( + not _mysql_available(), + reason="MySQL not available on 127.0.0.1:3307" +) @pytest.mark.live class TestMySQLLive: