starred/MediaManager-maxdorninger
1
0
Fork 0
mirror of https://github.com/maxdorninger/MediaManager.git synced 2026-04-17 15:13:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

format files

Browse Source
This commit is contained in:
maxDorninger
2025-07-09 20:54:50 +02:00
parent 1cf22b56d3
commit b6a5385864
10 changed files with 476 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
alembic/env.py
View File

@@ -52,7 +52,7 @@ target_metadata = Base.metadata
MovieFile,
MovieRequest,
Notification,
UsenetDownload
UsenetDownload,
)

371
alembic/versions/93fb07842385_initial_migration.py
View File

@@ -24,168 +24,229 @@ def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
# Create user table
op.create_table('user',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('email', sa.String(length=320), nullable=False),
sa.Column('hashed_password', sa.String(length=1024), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=False),
sa.Column('is_superuser', sa.Boolean(), nullable=False),
sa.Column('is_verified', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_user_email'), 'user', ['email'], unique=True)
op.create_table(
"user",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("email", sa.String(length=320), nullable=False),
sa.Column("hashed_password", sa.String(length=1024), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False),
sa.Column("is_superuser", sa.Boolean(), nullable=False),
sa.Column("is_verified", sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
# Create oauth account table
op.create_table('oauth_account',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('oauth_name', sa.String(length=100), nullable=False),
sa.Column('access_token', sa.String(length=1024), nullable=False),
sa.Column('expires_at', sa.Integer(), nullable=True),
sa.Column('refresh_token', sa.String(length=1024), nullable=True),
sa.Column('account_id', sa.String(length=320), nullable=False),
sa.Column('account_email', sa.String(length=320), nullable=False),
sa.Column('user_id', sa.UUID(), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ondelete='cascade'),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_oauth_account_account_id'), 'oauth_account', ['account_id'], unique=False)
op.create_index(op.f('ix_oauth_account_oauth_name'), 'oauth_account', ['oauth_name'], unique=False)
op.create_table(
"oauth_account",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("oauth_name", sa.String(length=100), nullable=False),
sa.Column("access_token", sa.String(length=1024), nullable=False),
sa.Column("expires_at", sa.Integer(), nullable=True),
sa.Column("refresh_token", sa.String(length=1024), nullable=True),
sa.Column("account_id", sa.String(length=320), nullable=False),
sa.Column("account_email", sa.String(length=320), nullable=False),
sa.Column("user_id", sa.UUID(), nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="cascade"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(
op.f("ix_oauth_account_account_id"),
"oauth_account",
["account_id"],
unique=False,
)
op.create_index(
op.f("ix_oauth_account_oauth_name"),
"oauth_account",
["oauth_name"],
unique=False,
)
# Create torrent table
op.create_table('torrent',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('status', sa.Enum('finished', 'downloading', 'error', 'unknown', name='torrentstatus'), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.Column('imported', sa.Boolean(), nullable=False),
sa.Column('hash', sa.String(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table(
"torrent",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column(
"status",
sa.Enum(
"finished", "downloading", "error", "unknown", name="torrentstatus"
),
nullable=False,
),
sa.Column("title", sa.String(), nullable=False),
sa.Column(
"quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.Column("imported", sa.Boolean(), nullable=False),
sa.Column("hash", sa.String(), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
# Create indexer query result table
op.create_table('indexer_query_result',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('download_url', sa.String(), nullable=False),
sa.Column('seeders', sa.Integer(), nullable=False),
sa.Column('flags', postgresql.ARRAY(sa.String()), nullable=True),
sa.Column('quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.Column('season', postgresql.ARRAY(sa.Integer()), nullable=True),
sa.Column('size', sa.BigInteger(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_table(
"indexer_query_result",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("title", sa.String(), nullable=False),
sa.Column("download_url", sa.String(), nullable=False),
sa.Column("seeders", sa.Integer(), nullable=False),
sa.Column("flags", postgresql.ARRAY(sa.String()), nullable=True),
sa.Column(
"quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.Column("season", postgresql.ARRAY(sa.Integer()), nullable=True),
sa.Column("size", sa.BigInteger(), nullable=True),
sa.PrimaryKeyConstraint("id"),
)
# Create notification table
op.create_table('notification',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('message', sa.String(), nullable=False),
sa.Column('read', sa.Boolean(), nullable=False),
sa.Column('timestamp', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table(
"notification",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("message", sa.String(), nullable=False),
sa.Column("read", sa.Boolean(), nullable=False),
sa.Column("timestamp", sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
# Create show table
op.create_table('show',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('external_id', sa.Integer(), nullable=False),
sa.Column('metadata_provider', sa.String(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('overview', sa.String(), nullable=False),
sa.Column('year', sa.Integer(), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('external_id', 'metadata_provider')
)
op.create_table(
"show",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("external_id", sa.Integer(), nullable=False),
sa.Column("metadata_provider", sa.String(), nullable=False),
sa.Column("name", sa.String(), nullable=False),
sa.Column("overview", sa.String(), nullable=False),
sa.Column("year", sa.Integer(), nullable=True),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("external_id", "metadata_provider"),
)
# Create movie table
op.create_table('movie',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('external_id', sa.Integer(), nullable=False),
sa.Column('metadata_provider', sa.String(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('overview', sa.String(), nullable=False),
sa.Column('year', sa.Integer(), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('external_id', 'metadata_provider')
)
op.create_table(
"movie",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("external_id", sa.Integer(), nullable=False),
sa.Column("metadata_provider", sa.String(), nullable=False),
sa.Column("name", sa.String(), nullable=False),
sa.Column("overview", sa.String(), nullable=False),
sa.Column("year", sa.Integer(), nullable=True),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("external_id", "metadata_provider"),
)
# Create season table
op.create_table('season',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('show_id', sa.UUID(), nullable=False),
sa.Column('number', sa.Integer(), nullable=False),
sa.Column('external_id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('overview', sa.String(), nullable=False),
sa.ForeignKeyConstraint(['show_id'], ['show.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('show_id', 'number')
)
op.create_table(
"season",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("show_id", sa.UUID(), nullable=False),
sa.Column("number", sa.Integer(), nullable=False),
sa.Column("external_id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(), nullable=False),
sa.Column("overview", sa.String(), nullable=False),
sa.ForeignKeyConstraint(["show_id"], ["show.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("show_id", "number"),
)
# Create movie file table
op.create_table('movie_file',
sa.Column('movie_id', sa.UUID(), nullable=False),
sa.Column('file_path_suffix', sa.String(), nullable=False),
sa.Column('quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.Column('torrent_id', sa.UUID(), nullable=True),
sa.ForeignKeyConstraint(['movie_id'], ['movie.id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['torrent_id'], ['torrent.id'], ondelete='SET NULL'),
sa.PrimaryKeyConstraint('movie_id', 'file_path_suffix')
)
op.create_table(
"movie_file",
sa.Column("movie_id", sa.UUID(), nullable=False),
sa.Column("file_path_suffix", sa.String(), nullable=False),
sa.Column(
"quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.Column("torrent_id", sa.UUID(), nullable=True),
sa.ForeignKeyConstraint(["movie_id"], ["movie.id"], ondelete="CASCADE"),
sa.ForeignKeyConstraint(["torrent_id"], ["torrent.id"], ondelete="SET NULL"),
sa.PrimaryKeyConstraint("movie_id", "file_path_suffix"),
)
# Create movie request table
op.create_table('movie_request',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('movie_id', sa.UUID(), nullable=False),
sa.Column('wanted_quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.Column('min_quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.Column('authorized', sa.Boolean(), nullable=False),
sa.Column('requested_by_id', sa.UUID(), nullable=True),
sa.Column('authorized_by_id', sa.UUID(), nullable=True),
sa.ForeignKeyConstraint(['authorized_by_id'], ['user.id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['movie_id'], ['movie.id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['requested_by_id'], ['user.id'], ondelete='SET NULL'),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('movie_id', 'wanted_quality')
)
op.create_table(
"movie_request",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("movie_id", sa.UUID(), nullable=False),
sa.Column(
"wanted_quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.Column(
"min_quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.Column("authorized", sa.Boolean(), nullable=False),
sa.Column("requested_by_id", sa.UUID(), nullable=True),
sa.Column("authorized_by_id", sa.UUID(), nullable=True),
sa.ForeignKeyConstraint(["authorized_by_id"], ["user.id"], ondelete="SET NULL"),
sa.ForeignKeyConstraint(["movie_id"], ["movie.id"], ondelete="CASCADE"),
sa.ForeignKeyConstraint(["requested_by_id"], ["user.id"], ondelete="SET NULL"),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("movie_id", "wanted_quality"),
)
# Create episode table
op.create_table('episode',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('season_id', sa.UUID(), nullable=False),
sa.Column('number', sa.Integer(), nullable=False),
sa.Column('external_id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.ForeignKeyConstraint(['season_id'], ['season.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('season_id', 'number')
)
op.create_table(
"episode",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("season_id", sa.UUID(), nullable=False),
sa.Column("number", sa.Integer(), nullable=False),
sa.Column("external_id", sa.Integer(), nullable=False),
sa.Column("title", sa.String(), nullable=False),
sa.ForeignKeyConstraint(["season_id"], ["season.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("season_id", "number"),
)
# Create season file table
op.create_table('season_file',
sa.Column('season_id', sa.UUID(), nullable=False),
sa.Column('torrent_id', sa.UUID(), nullable=True),
sa.Column('file_path_suffix', sa.String(), nullable=False),
sa.Column('quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.ForeignKeyConstraint(['season_id'], ['season.id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['torrent_id'], ['torrent.id'], ondelete='SET NULL'),
sa.PrimaryKeyConstraint('season_id', 'file_path_suffix')
)
op.create_table(
"season_file",
sa.Column("season_id", sa.UUID(), nullable=False),
sa.Column("torrent_id", sa.UUID(), nullable=True),
sa.Column("file_path_suffix", sa.String(), nullable=False),
sa.Column(
"quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.ForeignKeyConstraint(["season_id"], ["season.id"], ondelete="CASCADE"),
sa.ForeignKeyConstraint(["torrent_id"], ["torrent.id"], ondelete="SET NULL"),
sa.PrimaryKeyConstraint("season_id", "file_path_suffix"),
)
# Create season request table
op.create_table('season_request',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('season_id', sa.UUID(), nullable=False),
sa.Column('wanted_quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.Column('min_quality', sa.Enum('uhd', 'fullhd', 'hd', 'sd', 'unknown', name='quality'), nullable=False),
sa.Column('requested_by_id', sa.UUID(), nullable=True),
sa.Column('authorized', sa.Boolean(), nullable=False),
sa.Column('authorized_by_id', sa.UUID(), nullable=True),
sa.ForeignKeyConstraint(['authorized_by_id'], ['user.id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['requested_by_id'], ['user.id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['season_id'], ['season.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('season_id', 'wanted_quality')
)
op.create_table(
"season_request",
sa.Column("id", sa.UUID(), nullable=False),
sa.Column("season_id", sa.UUID(), nullable=False),
sa.Column(
"wanted_quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.Column(
"min_quality",
sa.Enum("uhd", "fullhd", "hd", "sd", "unknown", name="quality"),
nullable=False,
),
sa.Column("requested_by_id", sa.UUID(), nullable=True),
sa.Column("authorized", sa.Boolean(), nullable=False),
sa.Column("authorized_by_id", sa.UUID(), nullable=True),
sa.ForeignKeyConstraint(["authorized_by_id"], ["user.id"], ondelete="SET NULL"),
sa.ForeignKeyConstraint(["requested_by_id"], ["user.id"], ondelete="SET NULL"),
sa.ForeignKeyConstraint(["season_id"], ["season.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("season_id", "wanted_quality"),
)
# ### end Alembic commands ###
@@ -193,20 +254,20 @@ def upgrade() -> None:
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('season_request')
op.drop_table('season_file')
op.drop_table('episode')
op.drop_table('movie_request')
op.drop_table('movie_file')
op.drop_table('season')
op.drop_table('movie')
op.drop_table('show')
op.drop_table('notification')
op.drop_table('indexer_query_result')
op.drop_table('torrent')
op.drop_index(op.f('ix_oauth_account_oauth_name'), table_name='oauth_account')
op.drop_index(op.f('ix_oauth_account_account_id'), table_name='oauth_account')
op.drop_table('oauth_account')
op.drop_index(op.f('ix_user_email'), table_name='user')
op.drop_table('user')
# ### end Alembic commands ###
op.drop_table("season_request")
op.drop_table("season_file")
op.drop_table("episode")
op.drop_table("movie_request")
op.drop_table("movie_file")
op.drop_table("season")
op.drop_table("movie")
op.drop_table("show")
op.drop_table("notification")
op.drop_table("indexer_query_result")
op.drop_table("torrent")
op.drop_index(op.f("ix_oauth_account_oauth_name"), table_name="oauth_account")
op.drop_index(op.f("ix_oauth_account_account_id"), table_name="oauth_account")
op.drop_table("oauth_account")
op.drop_index(op.f("ix_user_email"), table_name="user")
op.drop_table("user")
# ### end Alembic commands ###

4
media_manager/indexer/indexers/jackett.py
View File

@@ -67,8 +67,8 @@ class Jackett(GenericIndexer):
seeders=seeders,
flags=[],
size=int(item.find("size").text),
usenet=False, # always False, because Jackett doesn't support usenet
age=0 # always 0 for torrents, as Jackett does not provide age information in a convenient format
usenet=False, # always False, because Jackett doesn't support usenet
age=0, # always 0 for torrents, as Jackett does not provide age information in a convenient format
)
result_list.append(result)
log.debug(f"Raw result: {result.model_dump()}")

8
media_manager/indexer/indexers/prowlarr.py
View File

@@ -30,7 +30,7 @@ class Prowlarr(GenericIndexer):
params = {
"query": query,
"apikey": self.api_key,
"categories": "5000" if is_tv else "2000" # TV: 5000, Movies: 2000
"categories": "5000" if is_tv else "2000", # TV: 5000, Movies: 2000
}
response = requests.get(url, params=params)
@@ -47,7 +47,7 @@ class Prowlarr(GenericIndexer):
flags=result["indexerFlags"],
size=result["size"],
usenet=True,
age=0, # Torrent results do not need age information
age=0, # Torrent results do not need age information
)
)
else:
@@ -55,11 +55,11 @@ class Prowlarr(GenericIndexer):
IndexerQueryResult(
download_url=result["downloadUrl"],
title=result["sortTitle"],
seeders=0, # Usenet results do not have seeders
seeders=0, # Usenet results do not have seeders
flags=result["indexerFlags"],
size=result["size"],
usenet=False,
age=int(result["ageMinutes"])*60,
age=int(result["ageMinutes"]) * 60,
)
)
log.debug("torrent result: " + result.__str__())

2
media_manager/indexer/models.py
View File

@@ -1,6 +1,6 @@
from uuid import UUID
from sqlalchemy import String, Integer, Boolean
from sqlalchemy import String, Integer
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.sql.sqltypes import BigInteger

2
media_manager/movies/repository.py
View File

@@ -362,7 +362,7 @@ class MovieRepository:
quality=torrent.quality,
imported=torrent.imported,
file_path_suffix=file_path_suffix,
usenet=torrent.usenet
usenet=torrent.usenet,
)
formatted_results.append(movie_torrent)
return formatted_results

3
media_manager/movies/service.py
View File

@@ -178,8 +178,7 @@ class MovieService:
search_query = f"{movie.name}"
torrents: list[IndexerQueryResult] = self.indexer_service.search(
query=search_query,
is_tv=False
query=search_query, is_tv=False
)
if search_query_override:

3
media_manager/tv/service.py
View File

@@ -189,8 +189,7 @@ class TvService:
search_query = show.name + " s" + str(season_number).zfill(2)
torrents: list[IndexerQueryResult] = self.indexer_service.search(
query=search_query,
is_tv=True
query=search_query, is_tv=True
)
if search_query_override:

280
tests/indexer/test_schemas.py
View File

@@ -5,31 +5,61 @@ from media_manager.torrent.models import Quality
def test_quality_computed_field():
assert (
IndexerQueryResult(
title="Show S01 4K", download_url="https://example.com/1", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 4K",
download_url="https://example.com/1",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.uhd
)
assert (
IndexerQueryResult(
title="Show S01 1080p", download_url="https://example.com/2", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 1080p",
download_url="https://example.com/2",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.fullhd
)
assert (
IndexerQueryResult(
title="Show S01 720p", download_url="https://example.com/3", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 720p",
download_url="https://example.com/3",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.hd
)
assert (
IndexerQueryResult(
title="Show S01 480p", download_url="https://example.com/4", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 480p",
download_url="https://example.com/4",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.sd
)
assert (
IndexerQueryResult(
title="Show S01", download_url="https://example.com/5", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01",
download_url="https://example.com/5",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.unknown
)
@@ -39,52 +69,100 @@ def test_quality_computed_field_edge_cases():
# Case-insensitive
assert (
IndexerQueryResult(
title="Show S01 4k", download_url="https://example.com/6", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 4k",
download_url="https://example.com/6",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.uhd
)
assert (
IndexerQueryResult(
title="Show S01 1080P", download_url="https://example.com/7", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 1080P",
download_url="https://example.com/7",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.fullhd
)
assert (
IndexerQueryResult(
title="Show S01 720P", download_url="https://example.com/8", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 720P",
download_url="https://example.com/8",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.hd
)
assert (
IndexerQueryResult(
title="Show S01 480P", download_url="https://example.com/9", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 480P",
download_url="https://example.com/9",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.sd
)
# Multiple quality tags, prefer highest
assert (
IndexerQueryResult(
title="Show S01 4K 1080p 720p", download_url="https://example.com/10", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 4K 1080p 720p",
download_url="https://example.com/10",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.uhd
)
assert (
IndexerQueryResult(
title="Show S01 1080p 720p", download_url="https://example.com/11", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 1080p 720p",
download_url="https://example.com/11",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.fullhd
)
# No quality tag
assert (
IndexerQueryResult(
title="Show S01", download_url="https://example.com/12", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01",
download_url="https://example.com/12",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.unknown
)
# Quality tag in the middle
assert (
IndexerQueryResult(
title="4K Show S01", download_url="https://example.com/13", seeders=1, flags=[], size=1, usenet=False, age=1
title="4K Show S01",
download_url="https://example.com/13",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).quality
== Quality.uhd
)
@@ -93,16 +171,34 @@ def test_quality_computed_field_edge_cases():
def test_season_computed_field():
# Single season
assert IndexerQueryResult(
title="Show S01", download_url="https://example.com/14", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01",
download_url="https://example.com/14",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season == [1]
# Range of seasons
assert IndexerQueryResult(
title="Show S01 S03", download_url="https://example.com/15", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 S03",
download_url="https://example.com/15",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season == [1, 2, 3]
# No season
assert (
IndexerQueryResult(
title="Show", download_url="https://example.com/16", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show",
download_url="https://example.com/16",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season
== []
)
@@ -112,46 +208,106 @@ def test_season_computed_field_edge_cases():
# Multiple seasons, unordered
assert (
IndexerQueryResult(
title="Show S03 S01", download_url="https://example.com/17", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S03 S01",
download_url="https://example.com/17",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season
== []
)
# Season with leading zeros
assert IndexerQueryResult(
title="Show S01 S03", download_url="https://example.com/18", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 S03",
download_url="https://example.com/18",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season == [1, 2, 3]
assert IndexerQueryResult(
title="Show S01 S01", download_url="https://example.com/19", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 S01",
download_url="https://example.com/19",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season == [1]
# No season at all
assert (
IndexerQueryResult(
title="Show", download_url="https://example.com/20", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show",
download_url="https://example.com/20",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season
== []
)
# Season in lower/upper case
assert IndexerQueryResult(
title="Show s02", download_url="https://example.com/21", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show s02",
download_url="https://example.com/21",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season == [2]
assert IndexerQueryResult(
title="Show S02", download_url="https://example.com/22", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S02",
download_url="https://example.com/22",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season == [2]
# Season with extra text
assert IndexerQueryResult(
title="Show S01 Complete", download_url="https://example.com/23", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 Complete",
download_url="https://example.com/23",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
).season == [1]
def test_gt_and_lt_methods():
a = IndexerQueryResult(
title="Show S01 1080p", download_url="https://example.com/24", seeders=5, flags=[], size=1, usenet=False, age=1
title="Show S01 1080p",
download_url="https://example.com/24",
seeders=5,
flags=[],
size=1,
usenet=False,
age=1,
)
b = IndexerQueryResult(
title="Show S01 720p", download_url="https://example.com/25", seeders=10, flags=[], size=1, usenet=False, age=1
title="Show S01 720p",
download_url="https://example.com/25",
seeders=10,
flags=[],
size=1,
usenet=False,
age=1,
)
c = IndexerQueryResult(
title="Show S01 1080p", download_url="https://example.com/26", seeders=2, flags=[], size=1, usenet=False, age=1
title="Show S01 1080p",
download_url="https://example.com/26",
seeders=2,
flags=[],
size=1,
usenet=False,
age=1,
)
# a (fullhd) > b (hd)
assert a > b
@@ -161,7 +317,13 @@ def test_gt_and_lt_methods():
assert a > c
# If quality is equal, but seeders are equal, neither is greater
d = IndexerQueryResult(
title="Show S01 1080p", download_url="https://example.com/27", seeders=5, flags=[], size=1, usenet=False, age=1
title="Show S01 1080p",
download_url="https://example.com/27",
seeders=5,
flags=[],
size=1,
usenet=False,
age=1,
)
assert not (a < d)
assert not (a > d)
@@ -170,40 +332,88 @@ def test_gt_and_lt_methods():
def test_gt_and_lt_methods_edge_cases():
# Different qualities
a = IndexerQueryResult(
title="Show S01 4K", download_url="https://example.com/28", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 4K",
download_url="https://example.com/28",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
)
b = IndexerQueryResult(
title="Show S01 1080p", download_url="https://example.com/29", seeders=100, flags=[], size=1, usenet=False, age=1
title="Show S01 1080p",
download_url="https://example.com/29",
seeders=100,
flags=[],
size=1,
usenet=False,
age=1,
)
assert a > b
assert not (b > a)
# Same quality, different seeders
c = IndexerQueryResult(
title="Show S01 4K", download_url="https://example.com/30", seeders=2, flags=[], size=1, usenet=False, age=1
title="Show S01 4K",
download_url="https://example.com/30",
seeders=2,
flags=[],
size=1,
usenet=False,
age=1,
)
assert a < c
assert c > a
# Same quality and seeders
d = IndexerQueryResult(
title="Show S01 4K", download_url="https://example.com/31", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 4K",
download_url="https://example.com/31",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
)
assert not (a < d)
assert not (a > d)
# Unknown quality, should compare by seeders
e = IndexerQueryResult(
title="Show S01", download_url="https://example.com/32", seeders=5, flags=[], size=1, usenet=False, age=1
title="Show S01",
download_url="https://example.com/32",
seeders=5,
flags=[],
size=1,
usenet=False,
age=1,
)
f = IndexerQueryResult(
title="Show S01", download_url="https://example.com/33", seeders=10, flags=[], size=1, usenet=False, age=1
title="Show S01",
download_url="https://example.com/33",
seeders=10,
flags=[],
size=1,
usenet=False,
age=1,
)
assert e < f
assert f > e
# Mixed known and unknown quality
g = IndexerQueryResult(
title="Show S01 720p", download_url="https://example.com/34", seeders=1, flags=[], size=1, usenet=False, age=1
title="Show S01 720p",
download_url="https://example.com/34",
seeders=1,
flags=[],
size=1,
usenet=False,
age=1,
)
h = IndexerQueryResult(
title="Show S01", download_url="https://example.com/35", seeders=100, flags=[], size=1, usenet=False, age=1
title="Show S01",
download_url="https://example.com/35",
seeders=100,
flags=[],
size=1,
usenet=False,
age=1,
)
assert g > h
assert not (h > g)

7
tests/tv/test_service.py
View File

@@ -476,8 +476,7 @@ def test_get_all_available_torrents_for_a_season_no_override(
mock_tv_repository.get_show_by_id.assert_called_once_with(show_id=show_id)
mock_indexer_service.search.assert_called_once_with(
query=f"{show_name} s{str(season_number).zfill(2)}",
is_tv=True
query=f"{show_name} s{str(season_number).zfill(2)}", is_tv=True
)
assert len(results) == 3
assert torrent1 in results
@@ -525,7 +524,9 @@ def test_get_all_available_torrents_for_a_season_with_override(
search_query_override=override_query,
)
mock_indexer_service.search.assert_called_once_with(query=override_query,is_tv=True)
mock_indexer_service.search.assert_called_once_with(
query=override_query, is_tv=True
)
assert results == [torrent1]