Skip to content

Commit

Permalink
Improve coverage for archives, score and matches (#1209)
Browse files Browse the repository at this point in the history
* improve coverage

* omit archives.py from test-core

* add pragma no cover in a Windows specific line

* mock: add more options

add option to compute get_matches
mock: specify languages and video_types
add logger messages and is_broken attribute to mimick a broken provider

* score: improve coverage

add matches for other ids
ignore match_hearing_impaired

* add coverage for score.py

* add pragma

* include archives.py

* coverage for archives.py

* avoid buggy sphinx release

* add news

* coverage: omit __main__.py
  • Loading branch information
getzze authored Feb 19, 2025
1 parent 1ceaee5 commit 4ce6f6d
Show file tree
Hide file tree
Showing 14 changed files with 322 additions and 166 deletions.
6 changes: 6 additions & 0 deletions .github/workflows/CI.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,12 @@ jobs:
with codecs.open(os.environ["GITHUB_ENV"], mode="a", encoding="utf-8") as file_handler:
file_handler.write(f"FORCE_COLOR=1\nENV={py}\n")
shell: python
- name: Install other test software (on Linux only)
if: matrix.os == 'ubuntu-latest'
uses: awalsh128/cache-apt-pkgs-action@v1
with:
packages: rar
version: 1.0
- name: Setup test environment
run: |
hatch -v env create ${ENV}
Expand Down
1 change: 1 addition & 0 deletions changelog.d/1209.change.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Improve core coverage: score.py, matches.py and archives.py
3 changes: 2 additions & 1 deletion hatch.toml
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,8 @@ run-cov-core = [
"test-cov-core",
"""\
coverage report --skip-covered --show-missing --fail-under=100 \
--omit='src/subliminal/cli.py,src/subliminal/converters/*,src/subliminal/providers/*,src/subliminal/refiners/*' \
--omit='src/subliminal/cli.py,src/subliminal/__main__.py,'\
'src/subliminal/converters/*,src/subliminal/providers/*,src/subliminal/refiners/*' \
""",
]

Expand Down
6 changes: 4 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ dependencies = [
[project.optional-dependencies]
rar = ["rarfile>=2.7"]
docs = [
"sphinx",
"sphinx<8.2",
"sphinx_rtd_theme>=2",
"sphinxcontrib-programoutput",
"sphinx_autodoc_typehints",
Expand Down Expand Up @@ -141,14 +141,16 @@ exclude_also = [
"if TYPE_CHECKING:",
"@overload",
"except ImportError",
"\\.\\.\\.",
"except PackageNotFoundError",
"\\.\\.\\.^",
"raise NotImplementedError()",
"if __name__ == .__main__.:",
]
show_missing = true
skip_covered = true
fail_under = 80
omit = [
"src/subliminal/__main__.py",
"src/subliminal/cli.py",
]

Expand Down
10 changes: 5 additions & 5 deletions src/subliminal/archives.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,14 +46,14 @@ def is_supported_archive(filename: str) -> bool:
if filename.lower().endswith(ARCHIVE_EXTENSIONS):
return True

if filename.lower().endswith('.rar'):
if filename.lower().endswith('.rar'): # pragma: no cover
msg = 'Install the rarfile module to be able to read rar archives.'
warnings.warn(msg, UserWarning, stacklevel=2)

return False
return False # pragma: no cover


def scan_archive(path: str | os.PathLike, name: str | None = None) -> Video: # pragma: no cover
def scan_archive(path: str | os.PathLike, name: str | None = None) -> Video:
"""Scan an archive from a `path`.
:param str path: existing path to the archive.
Expand All @@ -78,7 +78,7 @@ def scan_archive(path: str | os.PathLike, name: str | None = None) -> Video: #
raise ArchiveError(msg)


def scan_archive_rar(path: str | os.PathLike, name: str | None = None) -> Video: # pragma: no cover
def scan_archive_rar(path: str | os.PathLike, name: str | None = None) -> Video:
"""Scan a rar archive from a `path`.
:param str path: existing path to the archive.
Expand All @@ -90,7 +90,7 @@ def scan_archive_rar(path: str | os.PathLike, name: str | None = None) -> Video:
path = os.fspath(path)
# check for non-existing path
if not os.path.exists(path): # pragma: no cover
msg = 'Path does not exist'
msg = f'Path does not exist: {path!r}'
raise ValueError(msg)

if not is_rarfile(path):
Expand Down
89 changes: 78 additions & 11 deletions src/subliminal/providers/mock.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,9 @@
from typing import TYPE_CHECKING, Any, ClassVar

from babelfish import LANGUAGES, Language # type: ignore[import-untyped]
from guessit import guessit # type: ignore[import-untyped]

from subliminal.exceptions import NotInitializedProviderError
from subliminal.exceptions import NotInitializedProviderError, ProviderError
from subliminal.matches import guess_matches
from subliminal.subtitle import Subtitle
from subliminal.video import Episode, Movie, Video
Expand All @@ -25,13 +26,25 @@
class MockSubtitle(Subtitle):
"""Mock Subtitle."""

provider_name: ClassVar[str] = 'mock'
_ids: ClassVar = count(0)

#: Provider name, modify in subclasses
provider_name: ClassVar[str] = 'mock'

#: Fake content, will be copied to 'content' with :meth:`MockProvider.download_subtitle`
fake_content: bytes

#: Video name to match to be listed by :meth:`MockProvider.list_subtitles`
video_name: str

#: A set of matches to add to the set when :meth:`get_matches` is called
matches: set[str]
force_matches: bool

#: Guesses used as argument to compute the matches with :func:`guess_matches`
parameters: dict[str, Any]

#: Release name to be parsed with guessit and added to the :attr:`parameters`
release_name: str

def __init__(
self,
Expand All @@ -41,7 +54,8 @@ def __init__(
fake_content: bytes = b'',
video_name: str = '',
matches: Set[str] | None = None,
parameters: dict[str, Any] | None = None,
parameters: Mapping[str, Any] | None = None,
release_name: str = '',
**kwargs: Any,
) -> None:
# generate unique id for mock subtitle
Expand All @@ -55,15 +69,22 @@ def __init__(
)
self.fake_content = fake_content
self.video_name = video_name
self.force_matches = matches is not None
self.matches = set(matches) if matches is not None else set()
self.parameters = dict(parameters) if parameters is not None else {}
self.release_name = release_name

def get_matches(self, video: Video) -> set[str]:
"""Get the matches against the `video`."""
if self.force_matches:
return self.matches
return guess_matches(video, self.parameters)
# Use the parameters as guesses
matches = guess_matches(video, self.parameters)

# Parse the release_name and guess more matches
if self.release_name:
video_type = 'episode' if isinstance(video, Episode) else 'movie'
matches |= guess_matches(video, guessit(self.release_name, {'type': video_type}))

# Force add more matches
return matches | self.matches


class MockProvider(Provider):
Expand All @@ -77,22 +98,27 @@ class MockProvider(Provider):

logged_in: bool
subtitle_pool: list[MockSubtitle]
is_broken: bool

def __init__(self, subtitle_pool: Sequence[MockSubtitle] | None = None) -> None:
self.logged_in = False
self.subtitle_pool = list(self.internal_subtitle_pool)
if subtitle_pool is not None:
self.subtitle_pool.extend(list(subtitle_pool))
self.is_broken = False

def initialize(self) -> None:
"""Initialize the provider."""
logger.info('Mock provider %s was initialized', self.__class__.__name__)
self.logged_in = True

def terminate(self) -> None:
"""Terminate the provider."""
if not self.logged_in:
logger.info('Mock provider %s was not terminated', self.__class__.__name__)
raise NotInitializedProviderError

logger.info('Mock provider %s was terminated', self.__class__.__name__)
self.logged_in = False

def query(
Expand All @@ -102,27 +128,66 @@ def query(
matches: Set[str] | None = None,
) -> list[MockSubtitle]:
"""Query the provider for subtitles."""
if self.is_broken:
msg = f'Mock provider {self.__class__.__name__} query raised an error'
raise ProviderError(msg)

subtitles = []
for lang in languages:
subtitle = MockSubtitle(language=lang, video=video, matches=matches)
subtitle = self.subtitle_class(language=lang, video=video, matches=matches)
subtitles.append(subtitle)
logger.info(
'Mock provider %s query for video %r and languages %s: %d',
self.__class__.__name__,
video.name if video else None,
languages,
len(subtitles),
)
return subtitles

def list_subtitles(self, video: Video, languages: Set[Language]) -> list[MockSubtitle]:
"""List all the subtitles for the video."""
return [
if self.is_broken:
msg = f'Mock provider {self.__class__.__name__} list_subtitles raised an error'
raise ProviderError(msg)

subtitles = [
subtitle
for subtitle in self.subtitle_pool
if subtitle.language in languages and subtitle.video_name == video.name
]
logger.info(
'Mock provider %s list subtitles for video %r and languages %s: %d',
self.__class__.__name__,
video.name,
languages,
len(subtitles),
)
return subtitles

def download_subtitle(self, subtitle: MockSubtitle) -> None:
"""Download the content of the subtitle."""
if self.is_broken:
msg = f'Mock provider {self.__class__.__name__} download_subtitle raised an error'
raise ProviderError(msg)

logger.info(
'Mock provider %s download subtitle %s',
self.__class__.__name__,
subtitle,
)
subtitle.content = subtitle.fake_content


def mock_subtitle_provider(name: str, subtitles_info: Sequence[Mapping[str, Any]]) -> str:
def mock_subtitle_provider(
name: str,
subtitles_info: Sequence[Mapping[str, Any]],
languages: Set[Language] | None = None,
video_types: tuple[type[Episode] | type[Movie], ...] = (Episode, Movie),
) -> str:
"""Mock a subtitle provider, providing subtitles."""
languages = set(languages) if languages else {Language(lang) for lang in LANGUAGES}

name_lower = name.lower()
subtitle_class_name = f'{name}Subtitle'
provider_class_name = f'{name}Provider'
Expand All @@ -138,6 +203,8 @@ def mock_subtitle_provider(name: str, subtitles_info: Sequence[Mapping[str, Any]
{
'subtitle_class': MyMockSubtitle,
'internal_subtitle_pool': subtitle_pool,
'languages': languages,
'video_types': video_types,
},
)

Expand Down
15 changes: 12 additions & 3 deletions src/subliminal/score.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,7 @@ def get_scores(video: Video) -> dict[str, Any]:

def match_hearing_impaired(subtitle: Subtitle, *, hearing_impaired: bool | None = None) -> bool:
"""Match hearing impaired, if it is defined for the subtitle."""
return (
return ( # pragma: no cover
hearing_impaired is not None
and subtitle.hearing_impaired is not None
and subtitle.hearing_impaired == hearing_impaired
Expand Down Expand Up @@ -181,16 +181,25 @@ def compute_score(subtitle: Subtitle, video: Video, **kwargs: Any) -> int:
if 'imdb_id' in matches:
logger.debug('Adding imdb_id match equivalents')
matches |= {'series', 'year', 'country', 'season', 'episode'}
if 'tvdb_id' in matches:
logger.debug('Adding tvdb_id match equivalents')
if 'series_tmdb_id' in matches:
logger.debug('Adding series_tmdb_id match equivalents')
matches |= {'series', 'year', 'country'}
if 'tmdb_id' in matches:
logger.debug('Adding tmdb_id match equivalents')
matches |= {'series', 'year', 'country', 'season', 'episode'}
if 'series_tvdb_id' in matches:
logger.debug('Adding series_tvdb_id match equivalents')
matches |= {'series', 'year', 'country'}
if 'tvdb_id' in matches:
logger.debug('Adding tvdb_id match equivalents')
matches |= {'series', 'year', 'country', 'season', 'episode'}
elif isinstance(video, Movie): # pragma: no branch
if 'imdb_id' in matches:
logger.debug('Adding imdb_id match equivalents')
matches |= {'title', 'year', 'country'}
if 'tmdb_id' in matches:
logger.debug('Adding tmdb_id match equivalents')
matches |= {'title', 'year', 'country'}

# compute the score
score = int(sum(scores.get(match, 0) for match in matches))
Expand Down
10 changes: 5 additions & 5 deletions src/subliminal/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -219,7 +219,7 @@ def creation_date(filepath: os.PathLike | str) -> float:
See https://stackoverflow.com/a/39501288/1709587 for explanation.
"""
# Use creation time (although it may not be correct)
if platform.system() == 'Windows':
if platform.system() == 'Windows': # pragma: no cover
return os.path.getctime(filepath)
stat = os.stat(filepath)
try:
Expand Down Expand Up @@ -366,11 +366,11 @@ def clip(value: float, minimum: float | None, maximum: float | None) -> float:

def split_doc_args(args: str | None) -> list[str]:
"""Split the arguments of a docstring (in Sphinx docstyle)."""
if not args:
if not args: # pragma: no cover
return []
split_regex = re.compile(r'(?m):((param|type)\s|(return|yield|raise|rtype|ytype)s?:)')
split_indices = [m.start() for m in split_regex.finditer(args)]
if len(split_indices) == 0:
if len(split_indices) == 0: # pragma: no cover
return []
next_indices = [*split_indices[1:], None]
parts = [args[i:j].strip() for i, j in zip(split_indices, next_indices)]
Expand All @@ -388,10 +388,10 @@ def get_argument_doc(fun: Callable) -> dict[str, str]:
ret = {}
for p in parts:
m = param_regex.match(p)
if not m:
if not m: # pragma: no cover
continue
_, name, desc = m.groups()
if name is None:
if name is None: # pragma: no cover
continue
ret[name] = ' '.join(desc.strip().split())

Expand Down
Loading

0 comments on commit 4ce6f6d

Please sign in to comment.