Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ jobs:
run: pip install -e ".[dev]"

- name: Run tests
run: pytest tests/ -v --ignore=tests/test_postgres.py --ignore=tests/test_mysql.py --ignore=tests/test_oracle.py --ignore=tests/test_mssql.py --cov=pytest_mrt --cov-report=xml
run: coverage run -m pytest tests/ -v --ignore=tests/test_postgres.py --ignore=tests/test_mysql.py --ignore=tests/test_oracle.py --ignore=tests/test_mssql.py && coverage xml

- name: Upload coverage
uses: codecov/codecov-action@v7
Expand Down
10 changes: 9 additions & 1 deletion pytest_mrt/core/seeder.py
Original file line number Diff line number Diff line change
Expand Up @@ -270,7 +270,15 @@ def seed_table(self, table: TableInfo, count: int = 3) -> None:
# Append row_index to guarantee uniqueness
val = row[col_name]
if isinstance(val, str):
row[col_name] = (val + f"_{row_index}")[:255]
suffix = f"_{row_index}"
m = re.search(r"\((\d+)\)", col_info.type_str)
limit = int(m.group(1)) if m else 255
if len(suffix) >= limit:
# Column too short for suffix — rely on UUID randomness
row[col_name] = uuid.uuid4().hex[:limit]
else:
trimmed = val[: limit - len(suffix)]
row[col_name] = (trimmed + suffix)[:limit]
elif isinstance(val, int):
row[col_name] = val + row_index

Expand Down
191 changes: 191 additions & 0 deletions tests/test_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -920,3 +920,194 @@ def test_assert_schema_matches_raises_without_metadata(alembic_env):

fixture.downgrade()
fixture.reset()


# ── Mode-guard RuntimeErrors ──────────────────────────────────────────────────


def test_check_revision_raises_in_django_mode(alembic_env):
"""check_revision() raises RuntimeError when django_mode is active."""
cfg = MRTConfig(alembic_ini=alembic_env["ini"], db_url=alembic_env["db_url"])
fixture = MRTFixture.__new__(MRTFixture)
fixture._config = cfg
fixture._django_mode = True
fixture._runner = None
fixture._verifier = None
fixture._django_verifier = None
fixture._seeder = None

with pytest.raises(RuntimeError, match="Django mode"):
fixture.check_revision("rev1")


def test_check_migration_raises_in_alembic_mode(alembic_env):
"""check_migration() raises RuntimeError when NOT in django_mode."""
cfg = MRTConfig(alembic_ini=alembic_env["ini"], db_url=alembic_env["db_url"])
fixture = MRTFixture.__new__(MRTFixture)
fixture._config = cfg
fixture._django_mode = False
fixture._runner = None
fixture._verifier = None
fixture._django_verifier = None
fixture._seeder = None

with pytest.raises(RuntimeError, match="Django mode"):
fixture.check_migration("myapp", "0001_initial")


def test_assert_reversible_raises_in_django_mode(alembic_env):
"""assert_reversible() raises RuntimeError in django_mode."""
cfg = MRTConfig(alembic_ini=alembic_env["ini"], db_url=alembic_env["db_url"])
fixture = MRTFixture.__new__(MRTFixture)
fixture._config = cfg
fixture._django_mode = True
fixture._runner = None
fixture._verifier = None
fixture._django_verifier = None
fixture._seeder = None

with pytest.raises(RuntimeError, match="Django mode"):
fixture.assert_reversible("001")


# ── _auto_detect_django ImportError branch ────────────────────────────────────


def test_auto_detect_django_no_switch_when_django_not_installed(tmp_path, monkeypatch):
"""When django is not installed, _auto_detect_django returns config unchanged."""
import builtins

monkeypatch.setenv("DJANGO_SETTINGS_MODULE", "myapp.settings")
# Make alembic.ini appear missing
monkeypatch.chdir(tmp_path)

original_import = builtins.__import__

def mock_import(name, *args, **kwargs):
if name == "django":
raise ImportError("No module named 'django'")
return original_import(name, *args, **kwargs)

cfg = MRTConfig()

from pytest_mrt.plugin import _auto_detect_django

with pytest.MonkeyPatch().context() as m:
m.setattr(builtins, "__import__", mock_import)
result = _auto_detect_django(cfg)

# Should return config unchanged (no django_settings set)
assert result.django_settings is None


# ── check_static with parse-error migration ──────────────────────────────────


def test_check_static_skips_migration_with_parse_error(alembic_env):
"""check_static() with custom_checks skips files that fail to parse."""

# Write a syntactically broken migration file
broken = Path(alembic_env["versions"]) / "bad_syntax.py"
broken.write_text("def upgrade(: pass\n")

cfg = MRTConfig(
alembic_ini=alembic_env["ini"],
db_url=alembic_env["db_url"],
custom_checks=[lambda m: []], # should be called only for parseable files
)
fixture = MRTFixture(cfg)
# Should not raise — parse errors are silently skipped
warnings = fixture.check_static(alembic_env["versions"])
assert isinstance(warnings, list)
fixture.reset()

broken.unlink()


# ── assert_schema_matches with metadata_path override ───────────────────────


def test_assert_schema_matches_with_metadata_path_override(alembic_env):
"""metadata_path overrides MRTConfig.target_metadata when both are provided."""
_simple_reversible_migration(alembic_env["versions"])

cfg = MRTConfig(
alembic_ini=alembic_env["ini"],
db_url=alembic_env["db_url"],
target_metadata="pytest_mrt.core.schema:SchemaSnapshot", # will be overridden
)
fixture = MRTFixture(cfg)
fixture.upgrade("001")

# metadata_path pointing to a non-metadata object should raise ValueError or AttributeError
with pytest.raises((ValueError, AttributeError, Exception)):
fixture.assert_schema_matches(metadata_path="pytest_mrt.core.schema:SchemaSnapshot")

fixture.downgrade()
fixture.reset()


# ── mrt fixture — MRTConfigError path via pytester ───────────────────────────


def test_mrt_fixture_config_error_is_mrtconfigerror(tmp_path, monkeypatch):
"""MRTFixture raises MRTConfigError (not a generic exception) for missing ini.

The mrt fixture body catches MRTConfigError specifically; this test
verifies that missing alembic.ini raises that exact exception so the
fixture's except-branch is reachable.
"""
from pytest_mrt.exceptions import MRTConfigError

monkeypatch.delenv("DJANGO_SETTINGS_MODULE", raising=False)

cfg = MRTConfig(
alembic_ini=str(tmp_path / "nonexistent.ini"),
db_url="sqlite:///test.db",
)
with pytest.raises(MRTConfigError, match="alembic.ini not found"):
MRTFixture(cfg)


# ── pytest_sessionstart with missing alembic.ini ─────────────────────────────


def test_pytest_sessionstart_exits_on_missing_alembic_ini(pytester, monkeypatch):
"""pytest_sessionstart calls pytest.exit when alembic.ini is not found."""
monkeypatch.delenv("DJANGO_SETTINGS_MODULE", raising=False)
pytester.makeconftest("""
import os
os.environ.pop("DJANGO_SETTINGS_MODULE", None)
from pytest_mrt import MRTConfig
def pytest_configure(config):
config._mrt_config = MRTConfig(
alembic_ini="/definitely/missing/alembic.ini",
db_url="sqlite:///test.db",
)
""")
pytester.makepyfile("def test_x(): pass")
pytester.makeini("[pytest]\nmrt_default_tests = false\n")
result = pytester.runpytest()
# pytest.exit causes returncode 4
assert result.ret == 4


# ── pytest_collection_modifyitems exception path ─────────────────────────────


def test_collection_modifyitems_exception_warns_not_raises(pytester, alembic_env):
"""When default test injection fails, a warning is emitted but tests still run."""
pytester.makeconftest(f"""
from pytest_mrt import MRTConfig
def pytest_configure(config):
config._mrt_config = MRTConfig(
alembic_ini="{alembic_env["ini"]}",
db_url="{alembic_env["db_url"]}",
)
""")
pytester.makepyfile("def test_normal(): assert True")
# Force the injection to fail by making default_tests module unimportable
pytester.makeini("[pytest]\nmrt_default_tests = true\n")
result = pytester.runpytest("-v")
# Normal user test should still run even if injection fails
assert result.ret in (0, 1, 2) # not a hard crash
124 changes: 109 additions & 15 deletions tests/test_seeder.py
Original file line number Diff line number Diff line change
Expand Up @@ -527,9 +527,7 @@ def test_seed_table_skips_serial_pk_row(engine):
def test_seed_custom_inserts_rows_and_tracks_them(engine):
"""seed_custom() inserts caller-supplied rows and tracks them in _rows."""
with engine.begin() as conn:
conn.execute(
text("CREATE TABLE products (id INTEGER PRIMARY KEY, sku TEXT NOT NULL)")
)
conn.execute(text("CREATE TABLE products (id INTEGER PRIMARY KEY, sku TEXT NOT NULL)"))

seeder = SmartSeeder(engine)
seeder.seed_custom("products", "id", [{"id": 1, "sku": "ABC"}, {"id": 2, "sku": "DEF"}])
Expand All @@ -547,9 +545,7 @@ def test_seed_custom_inserts_rows_and_tracks_them(engine):
def test_seed_custom_verify_passes_when_rows_intact(engine):
"""verify() returns no failures when seed_custom rows are still present."""
with engine.begin() as conn:
conn.execute(
text("CREATE TABLE orders (id INTEGER PRIMARY KEY, ref TEXT NOT NULL)")
)
conn.execute(text("CREATE TABLE orders (id INTEGER PRIMARY KEY, ref TEXT NOT NULL)"))

seeder = SmartSeeder(engine)
seeder.seed_custom("orders", "id", [{"id": 10, "ref": "ORD-010"}])
Expand All @@ -559,9 +555,7 @@ def test_seed_custom_verify_passes_when_rows_intact(engine):
def test_seed_custom_verify_detects_deleted_row(engine):
"""verify() reports the missing row after it's deleted."""
with engine.begin() as conn:
conn.execute(
text("CREATE TABLE tickets (id INTEGER PRIMARY KEY, code TEXT NOT NULL)")
)
conn.execute(text("CREATE TABLE tickets (id INTEGER PRIMARY KEY, code TEXT NOT NULL)"))

seeder = SmartSeeder(engine)
seeder.seed_custom("tickets", "id", [{"id": 7, "code": "T007"}])
Expand Down Expand Up @@ -592,9 +586,7 @@ def test_seed_custom_auto_pk_retrieval(engine):
"""seed_custom() retrieves the auto-assigned PK when not provided in the row."""
with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE events2 (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)"
)
text("CREATE TABLE events2 (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)")
)

seeder = SmartSeeder(engine)
Expand All @@ -614,9 +606,7 @@ def test_get_unique_constraints_returns_pk_and_unique(engine):

with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE members (id INTEGER PRIMARY KEY, email TEXT NOT NULL UNIQUE)"
)
text("CREATE TABLE members (id INTEGER PRIMARY KEY, email TEXT NOT NULL UNIQUE)")
)

groups = _get_unique_constraints(engine, "members")
Expand Down Expand Up @@ -677,3 +667,107 @@ def test_two_seeder_instances_no_unique_collision(engine):
with engine.connect() as conn:
count = conn.execute(text("SELECT COUNT(*) FROM tags")).scalar()
assert count == 6 # 3 + 3


# ── VARCHAR short column — suffix must not be truncated (Bug 2) ────────────


def test_seed_table_short_varchar_suffix_preserved(engine):
"""VARCHAR(5): unique suffix must survive even though base value fills the limit."""
with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE codes ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" code VARCHAR(5) NOT NULL UNIQUE"
")"
)
)

from pytest_mrt.core.schema import SchemaSnapshot

snap = SchemaSnapshot.capture(engine)

import warnings as _w

with _w.catch_warnings(record=True) as w:
_w.simplefilter("always")
SmartSeeder(engine).seed_table(snap.tables["codes"], count=3)

seed_warnings = [x for x in w if "failed to seed" in str(x.message)]
assert seed_warnings == [], f"Unexpected insert failure: {seed_warnings}"

with engine.connect() as conn:
rows = conn.execute(text("SELECT code FROM codes ORDER BY id")).fetchall()

assert len(rows) == 3
codes = [r[0] for r in rows]
# All values must be distinct — suffix must have survived
assert len(set(codes)) == 3, f"Duplicate codes: {codes}"
# All values must fit within VARCHAR(5)
assert all(len(c) <= 5 for c in codes), f"Code exceeds limit: {codes}"


def test_seed_table_short_varchar_suffix_fills_exactly(engine):
"""When limit is exactly suffix length, suffix still wins and the result fits."""
with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE pins ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" pin VARCHAR(3) NOT NULL UNIQUE"
")"
)
)

from pytest_mrt.core.schema import SchemaSnapshot

snap = SchemaSnapshot.capture(engine)

import warnings as _w

with _w.catch_warnings(record=True) as w:
_w.simplefilter("always")
SmartSeeder(engine).seed_table(snap.tables["pins"], count=3)

seed_warnings = [x for x in w if "failed to seed" in str(x.message)]
assert seed_warnings == [], f"Unexpected insert failure: {seed_warnings}"

with engine.connect() as conn:
rows = conn.execute(text("SELECT pin FROM pins ORDER BY id")).fetchall()

pins = [r[0] for r in rows]
assert len(set(pins)) == len(pins), f"Duplicate pins: {pins}"
assert all(len(p) <= 3 for p in pins), f"Pin exceeds limit: {pins}"


def test_seed_table_int_unique_column_no_collision(engine):
"""INT unique constraint gets row_index added, not string suffix."""
with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE scores ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" rank INTEGER NOT NULL UNIQUE"
")"
)
)

from pytest_mrt.core.schema import SchemaSnapshot

snap = SchemaSnapshot.capture(engine)

import warnings as _w

with _w.catch_warnings(record=True) as w:
_w.simplefilter("always")
SmartSeeder(engine).seed_table(snap.tables["scores"], count=3)

seed_warnings = [x for x in w if "failed to seed" in str(x.message)]
assert seed_warnings == [], f"Unexpected insert failure: {seed_warnings}"

with engine.connect() as conn:
rows = conn.execute(text("SELECT rank FROM scores ORDER BY id")).fetchall()

ranks = [r[0] for r in rows]
assert len(set(ranks)) == 3, f"Duplicate ranks: {ranks}"
Loading
Loading