test(integration): MySQL 8 + Postgres 16 con Testcontainers

.github/workflows/ci.yml

@@ -10,8 +10,8 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  lint-type-test:
-    name: ${{ matrix.python-version }} · lint · type · test
+  unit:
+    name: unit · ${{ matrix.python-version }}
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
@@ -37,5 +37,26 @@ jobs:
       - name: Mypy
         run: mypy
 
-      - name: Pytest
+      - name: Pytest (unit)
         run: pytest -q
+
+  integration:
+    name: integration · MySQL + Postgres (testcontainers)
+    runs-on: ubuntu-latest
+    needs: unit
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+
+      - name: Install
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e ".[dev]"
+
+      - name: Pytest (integration)
+        run: pytest -m integration tests/integration -v

pyproject.toml

@@ -33,6 +33,7 @@ dev = [
     "ruff>=0.6",
     "mypy>=1.10",
     "psycopg[binary]>=3.1",
+    "testcontainers[mysql,postgres]>=4.7",
 ]
 
 [project.urls]
@@ -70,4 +71,7 @@ ignore_missing_imports = true
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
-addopts = "-ra --strict-markers"
+addopts = "-ra --strict-markers -m 'not integration'"
+markers = [
+    "integration: requiere Docker (MySQL/Postgres reales con Testcontainers)",
+]

tests/integration/conftest.py

@@ -0,0 +1,67 @@
+"""Fixtures que levantan MySQL y Postgres reales con Testcontainers.
+
+Cada test marcado con ``@pytest.mark.integration`` consume una de estas
+conexiones reales. Los contenedores se reutilizan en toda la sesión.
+"""
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+
+import pytest
+
+import shiba
+
+try:
+    from testcontainers.mysql import MySqlContainer
+    from testcontainers.postgres import PostgresContainer
+except ImportError:
+    MySqlContainer = None  # type: ignore[assignment,misc]
+    PostgresContainer = None  # type: ignore[assignment,misc]
+
+
+_SKIP_INTEGRATION = os.getenv("SHIBA_SKIP_INTEGRATION") == "1"
+
+
+@pytest.fixture(scope="session")
+def mysql_url() -> Iterator[str]:
+    if _SKIP_INTEGRATION or MySqlContainer is None:
+        pytest.skip("integration tests deshabilitados o testcontainers no instalado")
+    with MySqlContainer("mysql:8.0", dialect="pymysql") as container:
+        yield container.get_connection_url()
+
+
+@pytest.fixture(scope="session")
+def postgres_url() -> Iterator[str]:
+    if _SKIP_INTEGRATION or PostgresContainer is None:
+        pytest.skip("integration tests deshabilitados o testcontainers no instalado")
+    with PostgresContainer("postgres:16-alpine", driver=None) as container:
+        yield container.get_connection_url()
+
+
+def _normalize_to_shiba_dsn(url: str, scheme: str) -> str:
+    """``mysql+pymysql://...`` → ``mysql://...`` (Shiba ignora el driver hint)."""
+    if "+" in url.split("://", 1)[0]:
+        head, tail = url.split("://", 1)
+        return f"{scheme}://{tail}"
+    return url
+
+
+@pytest.fixture
+def mysql_cx(mysql_url: str) -> Iterator[shiba.ShibaConnection]:
+    dsn = _normalize_to_shiba_dsn(mysql_url, "mysql")
+    cx = shiba.connect(dsn)
+    try:
+        yield cx
+    finally:
+        cx.close()
+
+
+@pytest.fixture
+def postgres_cx(postgres_url: str) -> Iterator[shiba.ShibaConnection]:
+    dsn = _normalize_to_shiba_dsn(postgres_url, "postgres")
+    cx = shiba.connect(dsn)
+    try:
+        yield cx
+    finally:
+        cx.close()

tests/integration/test_mysql_e2e.py

@@ -0,0 +1,93 @@
+"""Smoke end-to-end contra MySQL 8 real."""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+def test_create_and_crud(mysql_cx) -> None:
+    mysql_cx.create_table("users").increments("id", primary_key=True).string(
+        "name", 64
+    ).integer("age").json("settings").build()
+    try:
+        tbl = mysql_cx.table("users")
+        tbl.insert({"name": "Alice", "age": 30, "settings": json.dumps({"theme": "dark"})})
+        tbl.insert({"name": "Bob", "age": 18, "settings": json.dumps({"theme": "light"})})
+
+        rows = mysql_cx.table("users").order_by("age").get()
+        assert [r["name"] for r in rows] == ["Bob", "Alice"]
+
+        n = mysql_cx.table("users").where("age", ">=", 18).count()
+        assert n == 2
+
+        mysql_cx.table("users").where("name", "Bob").update({"age": 19})
+        bob = mysql_cx.table("users").where("name", "Bob").first()
+        assert bob is not None and bob["age"] == 19
+
+        mysql_cx.table("users").where("name", "Bob").delete()
+        assert mysql_cx.table("users").count() == 1
+    finally:
+        mysql_cx.raw("DROP TABLE IF EXISTS users")
+
+
+def test_upsert(mysql_cx) -> None:
+    mysql_cx.raw("DROP TABLE IF EXISTS items")
+    mysql_cx.create_table("items").integer("id").primary().string("name").build()
+    try:
+        mysql_cx.table("items").upsert({"id": 1, "name": "A"})
+        mysql_cx.table("items").upsert({"id": 1, "name": "B"})  # actualiza
+        rows = mysql_cx.table("items").get()
+        assert len(rows) == 1
+        assert rows[0]["name"] == "B"
+    finally:
+        mysql_cx.raw("DROP TABLE IF EXISTS items")
+
+
+def test_transaction_rollback(mysql_cx) -> None:
+    mysql_cx.raw("DROP TABLE IF EXISTS t")
+    mysql_cx.create_table("t").integer("id").primary().build()
+    try:
+        with pytest.raises(RuntimeError), mysql_cx.transaction():
+            mysql_cx.table("t").insert({"id": 1})
+            raise RuntimeError("abort")
+        assert mysql_cx.table("t").count() == 0
+
+        with mysql_cx.transaction():
+            mysql_cx.table("t").insert({"id": 1})
+        assert mysql_cx.table("t").count() == 1
+    finally:
+        mysql_cx.raw("DROP TABLE IF EXISTS t")
+
+
+def test_orm_end_to_end(mysql_cx) -> None:
+    from shiba import Model, fields, set_default_connection
+
+    set_default_connection(mysql_cx)
+
+    class Customer(Model):
+        __table__ = "customers"
+        id: int = fields.PrimaryKey()
+        name: str
+        active: bool = True
+
+    mysql_cx.raw("DROP TABLE IF EXISTS customers")
+    Customer.create_table()
+    try:
+        Customer(name="Alice").save()
+        Customer(name="Bob", active=False).save()
+
+        rows = Customer.where("active", True).get()
+        assert len(rows) == 1
+        assert isinstance(rows[0], Customer)
+        assert rows[0].name == "Alice"
+
+        first = Customer.find(1)
+        assert first is not None and first.name == "Alice"
+        first.name = "Alice Renamed"
+        first.save()
+        assert Customer.find(1).name == "Alice Renamed"  # type: ignore[union-attr]
+    finally:
+        mysql_cx.raw("DROP TABLE IF EXISTS customers")

tests/integration/test_postgres_e2e.py

@@ -0,0 +1,59 @@
+"""Smoke end-to-end contra Postgres 16 real."""
+from __future__ import annotations
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+def test_create_and_crud(postgres_cx) -> None:
+    postgres_cx.raw("DROP TABLE IF EXISTS users")
+    postgres_cx.create_table("users").increments("id", primary_key=True).string(
+        "name", 64
+    ).integer("age").json("settings").build()
+    try:
+        tbl = postgres_cx.table("users")
+        tbl.insert({"name": "Alice", "age": 30, "settings": '{"theme": "dark"}'})
+        tbl.insert({"name": "Bob", "age": 18, "settings": '{"theme": "light"}'})
+
+        rows = postgres_cx.table("users").order_by("age").get()
+        assert [r["name"] for r in rows] == ["Bob", "Alice"]
+
+        n = postgres_cx.table("users").where("age", ">=", 18).count()
+        assert n == 2
+
+        postgres_cx.table("users").where("name", "Bob").update({"age": 19})
+        bob = postgres_cx.table("users").where("name", "Bob").first()
+        assert bob is not None and bob["age"] == 19
+
+        postgres_cx.table("users").where("name", "Bob").delete()
+        assert postgres_cx.table("users").count() == 1
+    finally:
+        postgres_cx.raw("DROP TABLE IF EXISTS users")
+
+
+def test_upsert_with_on(postgres_cx) -> None:
+    postgres_cx.raw("DROP TABLE IF EXISTS items")
+    postgres_cx.create_table("items").integer("id").primary().string("name").build()
+    try:
+        postgres_cx.table("items").upsert({"id": 1, "name": "A"}, on=["id"])
+        postgres_cx.table("items").upsert({"id": 1, "name": "B"}, on=["id"])
+        rows = postgres_cx.table("items").get()
+        assert len(rows) == 1
+        assert rows[0]["name"] == "B"
+    finally:
+        postgres_cx.raw("DROP TABLE IF EXISTS items")
+
+
+def test_identity_pk_generated(postgres_cx) -> None:
+    postgres_cx.raw("DROP TABLE IF EXISTS pk_test")
+    postgres_cx.create_table("pk_test").increments("id", primary_key=True).string(
+        "label"
+    ).build()
+    try:
+        postgres_cx.table("pk_test").insert({"label": "a"})
+        postgres_cx.table("pk_test").insert({"label": "b"})
+        ids = [r["id"] for r in postgres_cx.table("pk_test").order_by("id").get()]
+        assert ids == [1, 2]
+    finally:
+        postgres_cx.raw("DROP TABLE IF EXISTS pk_test")