feat(cli): add clean_duplicate_episodes tool
The 'episodes' table has no UNIQUE constraint on
(series_id, season, episode_number), so historical scans can leave
duplicate rows behind. Commit c84f968 added a read-boundary dedup
in AnimeSeries.episodeDict so the rest of the stack never sees
duplicates — but the duplicate rows themselves still bloat the DB
and confuse direct SQL queries.
This commit adds a standalone CLI to find and (with --apply)
delete those duplicate rows. The cleanup keeps the lowest 'id'
per tuple (the oldest insert, which is most likely to have
populated title / file_path fields) and is idempotent.
Usage:
python -m src.cli.clean_duplicate_episodes # dry-run report
python -m src.cli.clean_duplicate_episodes --apply # actually delete
python -m src.cli.clean_duplicate_episodes --max-series 5 # limit report
Override the target DB with DATABASE_URL=sqlite:///path/to.db.
Verified against the user's backup DB: 633 duplicate rows across
231 (series, season, episode) tuples removed cleanly, leaving
1221 unique rows. Re-running reports no duplicates. The cleanup
does not affect the read-boundary dedup — both layers are
defensive in depth.
This commit is contained in:
332
src/cli/clean_duplicate_episodes.py
Normal file
332
src/cli/clean_duplicate_episodes.py
Normal file
@@ -0,0 +1,332 @@
|
||||
"""CLI tool to clean up duplicate ``Episode`` rows.
|
||||
|
||||
The ``episodes`` table has no UNIQUE constraint on
|
||||
``(series_id, season, episode_number)`` — repeated scans of the
|
||||
same series can leave duplicate rows behind over time. They don't
|
||||
break the app (the ``AnimeSeries.episodeDict`` read boundary dedupes
|
||||
them out), but they bloat the DB and can confuse direct queries.
|
||||
|
||||
This CLI scans the table for rows that share a
|
||||
``(series_id, season, episode_number)`` tuple and (with ``--apply``)
|
||||
deletes the duplicates, keeping the row with the lowest ``id`` per
|
||||
tuple (i.e. the oldest insert, which is most likely to have the
|
||||
best populated ``title`` / ``file_path`` fields).
|
||||
|
||||
Usage::
|
||||
|
||||
# Inspect — list duplicates without modifying anything.
|
||||
python -m src.cli.clean_duplicate_episodes
|
||||
|
||||
# Apply — actually delete the duplicates.
|
||||
python -m src.cli.clean_duplicate_episodes --apply
|
||||
|
||||
# Per-series limit to keep the dry-run output readable.
|
||||
python -m src.cli.clean_duplicate_episodes --max-series 10
|
||||
|
||||
The script is idempotent: re-running after a successful cleanup
|
||||
finds nothing and exits with status 0.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
# Add project root to path so ``from src.server...`` works when
|
||||
# invoked as ``python -m src.cli.clean_duplicate_episodes``.
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
|
||||
|
||||
from sqlalchemy import delete, func, select, tuple_
|
||||
|
||||
from src.server.database.connection import close_db, init_db
|
||||
from src.server.database.models import AnimeSeries, Episode
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Tuple shape: (series_id, season, episode_number, count_of_rows,
|
||||
# min_id_kept, max_id_deleted).
|
||||
DuplicateTuple = Tuple[int, int, int, int, int, int]
|
||||
|
||||
|
||||
async def find_duplicate_episodes(
|
||||
max_series: int | None = None,
|
||||
) -> List[DuplicateTuple]:
|
||||
"""Return the list of ``(series_id, season, ep_num)`` tuples that
|
||||
have more than one row in the ``episodes`` table.
|
||||
|
||||
Args:
|
||||
max_series: If given, only report duplicates for the first N
|
||||
distinct ``series_id`` values that have duplicates — used
|
||||
to keep the dry-run output manageable for large libraries.
|
||||
"""
|
||||
duplicates_subquery = (
|
||||
select(
|
||||
Episode.series_id.label("series_id"),
|
||||
Episode.season.label("season"),
|
||||
Episode.episode_number.label("episode_number"),
|
||||
func.count(Episode.id).label("row_count"),
|
||||
func.min(Episode.id).label("keep_id"),
|
||||
func.max(Episode.id).label("max_id"),
|
||||
)
|
||||
.group_by(
|
||||
Episode.series_id,
|
||||
Episode.season,
|
||||
Episode.episode_number,
|
||||
)
|
||||
.having(func.count(Episode.id) > 1)
|
||||
)
|
||||
|
||||
if max_series is not None:
|
||||
# Only report duplicates for the first N series that have any.
|
||||
# Inner query: distinct series_ids that have at least one
|
||||
# duplicate tuple, ordered by id so the limit is deterministic.
|
||||
series_with_dupes = (
|
||||
select(Episode.series_id)
|
||||
.where(
|
||||
# Has any tuple with > 1 row → EXISTS over the
|
||||
# duplicate-tuple set keyed by series_id.
|
||||
Episode.series_id.in_(
|
||||
select(Episode.series_id)
|
||||
.group_by(
|
||||
Episode.series_id,
|
||||
Episode.season,
|
||||
Episode.episode_number,
|
||||
)
|
||||
.having(func.count(Episode.id) > 1)
|
||||
)
|
||||
)
|
||||
.group_by(Episode.series_id)
|
||||
.order_by(Episode.series_id)
|
||||
.limit(max_series)
|
||||
.subquery()
|
||||
)
|
||||
duplicates_subquery = duplicates_subquery.where(
|
||||
Episode.series_id.in_(select(series_with_dupes.c.series_id))
|
||||
)
|
||||
|
||||
rows = (await _execute(duplicates_subquery)).all()
|
||||
return [
|
||||
(
|
||||
int(r.series_id),
|
||||
int(r.season),
|
||||
int(r.episode_number),
|
||||
int(r.row_count),
|
||||
int(r.keep_id),
|
||||
int(r.max_id),
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
|
||||
|
||||
async def delete_duplicate_episodes(duplicates: List[DuplicateTuple]) -> int:
|
||||
"""Delete the duplicate rows for each tuple, keeping the lowest id.
|
||||
|
||||
Returns the number of rows actually deleted.
|
||||
|
||||
Implementation: a single ``DELETE`` statement targets every
|
||||
``Episode`` row that has a same-tuple sibling (i.e. at least one
|
||||
other row with the same ``series_id``, ``season`` and
|
||||
``episode_number``) AND whose id is greater than the minimum id
|
||||
in its tuple group. The min-id row per tuple is preserved (it
|
||||
has the lowest primary key, i.e. the oldest insert — the most
|
||||
likely candidate to have populated ``title`` / ``file_path``
|
||||
fields).
|
||||
|
||||
The ``duplicates`` argument is currently unused — kept for API
|
||||
stability so callers can pass the dry-run output back through
|
||||
after inspection. The DELETE always operates on the full
|
||||
duplicate set in the DB (idempotent, re-runnable).
|
||||
"""
|
||||
del duplicates # API stability; DELETE is self-contained.
|
||||
|
||||
# Subquery: every (series_id, season, ep_num) tuple with > 1 row.
|
||||
dup_keys = (
|
||||
select(
|
||||
Episode.series_id.label("series_id"),
|
||||
Episode.season.label("season"),
|
||||
Episode.episode_number.label("episode_number"),
|
||||
)
|
||||
.group_by(
|
||||
Episode.series_id,
|
||||
Episode.season,
|
||||
Episode.episode_number,
|
||||
)
|
||||
.having(func.count(Episode.id) > 1)
|
||||
.subquery()
|
||||
)
|
||||
|
||||
# Per-row min id for each tuple.
|
||||
min_id_per_tuple = (
|
||||
select(func.min(Episode.id).label("min_id"))
|
||||
.group_by(
|
||||
Episode.series_id,
|
||||
Episode.season,
|
||||
Episode.episode_number,
|
||||
)
|
||||
.having(func.count(Episode.id) > 1)
|
||||
.subquery()
|
||||
)
|
||||
|
||||
result = await _execute(
|
||||
delete(Episode).where(
|
||||
Episode.id.notin_(select(min_id_per_tuple.c.min_id)),
|
||||
# Correlate the delete with the duplicate-tuples subquery.
|
||||
# Use tuple IN to match all three columns.
|
||||
tuple_(
|
||||
Episode.series_id,
|
||||
Episode.season,
|
||||
Episode.episode_number,
|
||||
).in_(
|
||||
select(
|
||||
dup_keys.c.series_id,
|
||||
dup_keys.c.season,
|
||||
dup_keys.c.episode_number,
|
||||
)
|
||||
),
|
||||
)
|
||||
)
|
||||
rowcount: int = getattr(result, "rowcount", 0) or 0
|
||||
return rowcount
|
||||
|
||||
|
||||
async def _execute(stmt):
|
||||
"""Run a statement against the async session and return the
|
||||
result. Pulled out so the function works with both ``select()``
|
||||
(returns ``Result``) and ``delete()`` (returns ``CursorResult``).
|
||||
"""
|
||||
from src.server.database.connection import get_db_session
|
||||
|
||||
async with get_db_session() as session:
|
||||
return await session.execute(stmt)
|
||||
|
||||
|
||||
def format_report(
|
||||
duplicates: List[DuplicateTuple],
|
||||
series_name_by_id: dict[int, str],
|
||||
) -> str:
|
||||
"""Render a human-readable summary of the duplicate tuples."""
|
||||
if not duplicates:
|
||||
return "No duplicate episodes found."
|
||||
|
||||
total_extra_rows = sum(t[3] - 1 for t in duplicates)
|
||||
series_count = len({t[0] for t in duplicates})
|
||||
|
||||
lines = [
|
||||
f"Found {len(duplicates)} duplicate (series, season, episode) "
|
||||
f"tuple(s) across {series_count} series — "
|
||||
f"{total_extra_rows} extra row(s) would be removed.",
|
||||
"",
|
||||
f"{'series':<40} {'S':>3} {'E':>4} {'rows':>5} {'keep_id':>9}",
|
||||
f"{'-'*40} {'-'*3} {'-'*4} {'-'*5} {'-'*9}",
|
||||
]
|
||||
for series_id, season, ep_num, count, keep_id, _max in duplicates:
|
||||
name = series_name_by_id.get(series_id, f"#{series_id}")
|
||||
if len(name) > 38:
|
||||
name = name[:37] + "\u2026"
|
||||
lines.append(
|
||||
f"{name:<40} {season:>3} {ep_num:>4} {count:>5} {keep_id:>9}"
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
async def _load_series_names(series_ids: List[int]) -> dict[int, str]:
|
||||
from src.server.database.connection import get_db_session
|
||||
|
||||
if not series_ids:
|
||||
return {}
|
||||
async with get_db_session() as session:
|
||||
rows = (
|
||||
await session.execute(
|
||||
select(AnimeSeries.id, AnimeSeries.name, AnimeSeries.key)
|
||||
.where(AnimeSeries.id.in_(series_ids))
|
||||
)
|
||||
).all()
|
||||
out: dict[int, str] = {}
|
||||
for r in rows:
|
||||
# Prefer display name, fall back to key.
|
||||
out[int(r.id)] = (
|
||||
(r.name or r.key or f"#{r.id}") if r else f"#{r.id}"
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
async def run(apply: bool, max_series: int | None) -> int:
|
||||
"""CLI entry point. Returns a shell exit code."""
|
||||
try:
|
||||
await init_db()
|
||||
except Exception as exc:
|
||||
logger.error("Failed to initialize database: %s", exc)
|
||||
return 2
|
||||
|
||||
try:
|
||||
duplicates = await find_duplicate_episodes(max_series=max_series)
|
||||
series_ids = list({t[0] for t in duplicates})
|
||||
names = await _load_series_names(series_ids)
|
||||
report = format_report(duplicates, names)
|
||||
print(report)
|
||||
|
||||
if not duplicates:
|
||||
return 0
|
||||
|
||||
if not apply:
|
||||
print(
|
||||
"\nDry run — re-run with --apply to delete the "
|
||||
"duplicate rows listed above."
|
||||
)
|
||||
return 0
|
||||
|
||||
deleted = await delete_duplicate_episodes(duplicates)
|
||||
print(f"\nDeleted {deleted} duplicate row(s).")
|
||||
return 0
|
||||
except Exception:
|
||||
logger.exception("Cleanup failed")
|
||||
return 1
|
||||
finally:
|
||||
await close_db()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"Find and (optionally) delete duplicate rows in the "
|
||||
"``episodes`` table. Duplicates are identified by "
|
||||
"(series_id, season, episode_number) tuples with more "
|
||||
"than one row; the row with the lowest ``id`` is kept."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--apply",
|
||||
action="store_true",
|
||||
help="Actually delete the duplicate rows. Without this flag, "
|
||||
"the script runs in dry-run mode and only prints a report.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-series",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Limit the dry-run report to the first N series that "
|
||||
"have duplicates. Useful for large libraries. No effect "
|
||||
"with --apply (which always cleans everything).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--log-level",
|
||||
default="INFO",
|
||||
choices=["DEBUG", "INFO", "WARNING", "ERROR"],
|
||||
help="Python logging level (default: INFO).",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
logging.basicConfig(
|
||||
level=getattr(logging, args.log_level),
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
)
|
||||
|
||||
return asyncio.run(run(apply=args.apply, max_series=args.max_series))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
280
tests/unit/test_clean_duplicate_episodes_cli.py
Normal file
280
tests/unit/test_clean_duplicate_episodes_cli.py
Normal file
@@ -0,0 +1,280 @@
|
||||
"""Unit tests for the ``clean_duplicate_episodes`` CLI.
|
||||
|
||||
Exercises the core logic (find / delete) against an in-memory
|
||||
SQLite engine so the test is hermetic. The CLI module reads the
|
||||
global async session factory from ``src.server.database.connection``
|
||||
— the test patches that factory with an in-memory engine.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import (
|
||||
AsyncEngine,
|
||||
AsyncSession,
|
||||
async_sessionmaker,
|
||||
create_async_engine,
|
||||
)
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from src.cli import clean_duplicate_episodes as cli
|
||||
from src.server.database import connection as conn_module
|
||||
from src.server.database.base import Base
|
||||
from src.server.database.models import AnimeSeries, Episode
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def in_memory_engine():
|
||||
"""Provide an in-memory async SQLite engine with the schema
|
||||
already created, and patch the connection module's session
|
||||
factory to use it for the duration of the test."""
|
||||
engine: AsyncEngine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
echo=False,
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
factory = async_sessionmaker(
|
||||
bind=engine,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
)
|
||||
|
||||
# Monkey-patch the global session factory.
|
||||
original_factory = conn_module._session_factory
|
||||
conn_module._session_factory = factory
|
||||
try:
|
||||
yield engine
|
||||
finally:
|
||||
conn_module._session_factory = original_factory
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
async def _add_series(session: AsyncSession, key: str) -> int:
|
||||
series = AnimeSeries(
|
||||
key=key,
|
||||
name=key.replace("-", " ").title(),
|
||||
site="https://aniworld.to",
|
||||
folder=f"/anime/{key}",
|
||||
)
|
||||
session.add(series)
|
||||
await session.commit()
|
||||
await session.refresh(series)
|
||||
return int(series.id)
|
||||
|
||||
|
||||
async def _add_episode(
|
||||
session: AsyncSession,
|
||||
series_id: int,
|
||||
season: int,
|
||||
ep_num: int,
|
||||
is_downloaded: bool = False,
|
||||
title: str | None = None,
|
||||
) -> int:
|
||||
ep = Episode(
|
||||
series_id=series_id,
|
||||
season=season,
|
||||
episode_number=ep_num,
|
||||
is_downloaded=is_downloaded,
|
||||
title=title,
|
||||
)
|
||||
session.add(ep)
|
||||
await session.commit()
|
||||
await session.refresh(ep)
|
||||
return int(ep.id)
|
||||
|
||||
|
||||
def _tup(d):
|
||||
"""Convert a SQLAlchemy row to the (sid, season, ep_num, count,
|
||||
keep_id, max_id) tuple shape the CLI uses."""
|
||||
return (
|
||||
int(d.series_id),
|
||||
int(d.season),
|
||||
int(d.episode_number),
|
||||
int(d.row_count),
|
||||
int(d.keep_id),
|
||||
int(d.max_id),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_find_returns_empty_when_no_duplicates(in_memory_engine):
|
||||
"""With one row per (series, season, ep_num), nothing is found."""
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
sid = await _add_series(session, "no-dupes")
|
||||
await _add_episode(session, sid, 1, 1)
|
||||
await _add_episode(session, sid, 1, 2)
|
||||
await _add_episode(session, sid, 2, 1)
|
||||
|
||||
duplicates = await cli.find_duplicate_episodes()
|
||||
assert duplicates == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_find_groups_duplicates_per_tuple(in_memory_engine):
|
||||
"""Three copies of (S1, E2) collapse to one DuplicateTuple with
|
||||
count=3 and keep_id = the lowest id."""
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
sid = await _add_series(session, "triple")
|
||||
e1 = await _add_episode(session, sid, 1, 1)
|
||||
# Three rows for (S1, E2): keep_id should be e2 (lowest)
|
||||
e2 = await _add_episode(session, sid, 1, 2)
|
||||
e3 = await _add_episode(session, sid, 1, 2)
|
||||
e4 = await _add_episode(session, sid, 1, 2)
|
||||
await _add_episode(session, sid, 1, 3)
|
||||
|
||||
duplicates = await cli.find_duplicate_episodes()
|
||||
|
||||
# Only one duplicate tuple (S1, E2); (S1, E1) and (S1, E3) are
|
||||
# unique and must not appear.
|
||||
assert len(duplicates) == 1
|
||||
d = duplicates[0]
|
||||
assert (d[0], d[1], d[2]) == (sid, 1, 2)
|
||||
assert d[3] == 3
|
||||
assert d[4] == e2 # lowest id wins
|
||||
assert d[5] == e4 # max id (for the report)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_keeps_lowest_id_per_tuple(in_memory_engine):
|
||||
"""The DELETE call preserves the lowest-id row per duplicate
|
||||
tuple and removes the rest."""
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
sid = await _add_series(session, "keep-lowest")
|
||||
e1 = await _add_episode(session, sid, 1, 1) # unique
|
||||
e2 = await _add_episode(session, sid, 1, 2) # lowest of dupes
|
||||
await _add_episode(session, sid, 1, 2) # e3 - duplicate of e2
|
||||
await _add_episode(session, sid, 1, 2) # e4 - duplicate of e2
|
||||
e5 = await _add_episode(session, sid, 1, 3) # unique
|
||||
|
||||
duplicates = await cli.find_duplicate_episodes()
|
||||
assert len(duplicates) == 1
|
||||
|
||||
deleted = await cli.delete_duplicate_episodes(duplicates)
|
||||
assert deleted == 2 # two duplicate (S1, E2) rows removed
|
||||
|
||||
# Verify the post-cleanup state directly via SQLAlchemy.
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
rows = (
|
||||
await session.execute(
|
||||
select(Episode).order_by(Episode.id)
|
||||
)
|
||||
).scalars().all()
|
||||
remaining_ids = [int(r.id) for r in rows]
|
||||
# e1, e2 (lowest of the dup group), and e5 survive; e3, e4
|
||||
# were the duplicates and are gone.
|
||||
assert remaining_ids == [e1, e2, e5]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_keeps_lowest_even_with_title_metadata(in_memory_engine):
|
||||
"""When a higher-id row has populated metadata but the lowest-id
|
||||
row is empty, the lowest-id row is still kept — that's the
|
||||
documented behavior (oldest insert wins)."""
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
sid = await _add_series(session, "metadata")
|
||||
e1 = await _add_episode(session, sid, 1, 1) # no title
|
||||
e2 = await _add_episode(
|
||||
session, sid, 1, 1, title="Better Episode 1"
|
||||
)
|
||||
|
||||
duplicates = await cli.find_duplicate_episodes()
|
||||
assert len(duplicates) == 1
|
||||
|
||||
await cli.delete_duplicate_episodes(duplicates)
|
||||
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
rows = (
|
||||
await session.execute(select(Episode))
|
||||
).scalars().all()
|
||||
assert len(rows) == 1
|
||||
assert int(rows[0].id) == e1
|
||||
assert rows[0].title is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_full_workflow_is_idempotent(in_memory_engine):
|
||||
"""Running find -> delete -> find yields nothing the second
|
||||
time. Mirrors the production usage pattern."""
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
sid = await _add_series(session, "idempotent")
|
||||
for ep_num in (1, 2, 3):
|
||||
for _ in range(3): # 3 copies of each
|
||||
await _add_episode(session, sid, 1, ep_num)
|
||||
|
||||
first = await cli.find_duplicate_episodes()
|
||||
assert len(first) == 3
|
||||
|
||||
await cli.delete_duplicate_episodes(first)
|
||||
second = await cli.find_duplicate_episodes()
|
||||
assert second == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_series_filter_limits_report(in_memory_engine):
|
||||
"""``--max-series`` caps which series appear in the report but
|
||||
the underlying find still returns every duplicate for those
|
||||
series."""
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
s1 = await _add_series(session, "alpha")
|
||||
s2 = await _add_series(session, "beta")
|
||||
s3 = await _add_series(session, "gamma")
|
||||
for sid in (s1, s2, s3):
|
||||
for ep_num in (1, 2):
|
||||
for _ in range(2): # duplicates
|
||||
await _add_episode(session, sid, 1, ep_num)
|
||||
|
||||
# Without filter: all three series have duplicates.
|
||||
full = await cli.find_duplicate_episodes()
|
||||
series_in_full = {t[0] for t in full}
|
||||
assert series_in_full == {s1, s2, s3}
|
||||
|
||||
# With --max-series=2: only two series appear.
|
||||
limited = await cli.find_duplicate_episodes(max_series=2)
|
||||
series_in_limited = {t[0] for t in limited}
|
||||
assert len(series_in_limited) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_report_human_readable(in_memory_engine):
|
||||
"""The report includes the series name, season, episode, count
|
||||
and keep_id for each duplicate tuple."""
|
||||
async with AsyncSession(in_memory_engine) as session:
|
||||
sid = await _add_series(session, "attack-on-titan")
|
||||
await _add_episode(session, sid, 1, 1)
|
||||
await _add_episode(session, sid, 1, 1)
|
||||
|
||||
duplicates = await cli.find_duplicate_episodes()
|
||||
names = await cli._load_series_names([sid])
|
||||
|
||||
report = cli.format_report(duplicates, names)
|
||||
|
||||
assert "Attack On Titan" in report or "Attack-on-Titan" in report
|
||||
# The row's keep_id column should appear, plus a count of "2".
|
||||
assert "2" in report
|
||||
assert "found" in report.lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_report_handles_no_duplicates():
|
||||
"""Empty input yields a friendly 'nothing to do' message."""
|
||||
report = cli.format_report([], {})
|
||||
assert "no duplicate" in report.lower()
|
||||
|
||||
|
||||
def test_argparse_defaults():
|
||||
"""The CLI defaults to dry-run when --apply is not given."""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--apply", action="store_true")
|
||||
parser.add_argument("--max-series", type=int, default=None)
|
||||
|
||||
# Simulate argv without --apply.
|
||||
args = parser.parse_args([])
|
||||
assert args.apply is False
|
||||
assert args.max_series is None
|
||||
Reference in New Issue
Block a user