1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
# This file is part of beets.
# Copyright 2015, Thomas Scholtes.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
from __future__ import annotations
import os
import sys
import unittest
from contextlib import contextmanager
from typing import TYPE_CHECKING, Callable
from beets import plugins
from beets.test.helper import PluginTestCase, capture_log
if TYPE_CHECKING:
from collections.abc import Iterator
class HookTestCase(PluginTestCase):
plugin = "hook"
preload_plugin = False
def _get_hook(self, event: str, command: str) -> dict[str, str]:
return {"event": event, "command": command}
class HookLogsTest(HookTestCase):
HOOK: plugins.EventType = "write"
@contextmanager
def _configure_logs(self, command: str) -> Iterator[list[str]]:
config = {"hooks": [self._get_hook(self.HOOK, command)]}
with self.configure_plugin(config), capture_log("beets.hook") as logs:
plugins.send(self.HOOK)
yield logs
def test_hook_empty_command(self):
with self._configure_logs("") as logs:
assert 'hook: invalid command ""' in logs
# FIXME: fails on windows
@unittest.skipIf(sys.platform == "win32", "win32")
def test_hook_non_zero_exit(self):
with self._configure_logs('sh -c "exit 1"') as logs:
assert f"hook: hook for {self.HOOK} exited with status 1" in logs
def test_hook_non_existent_command(self):
with self._configure_logs("non-existent-command") as logs:
logs = "\n".join(logs)
assert f"hook: hook for {self.HOOK} failed: " in logs
# The error message is different for each OS. Unfortunately the text is
# different in each case, where the only shared text is the string
# 'file' and substring 'Err'
assert "Err" in logs
assert "file" in logs
class HookCommandTest(HookTestCase):
EVENTS: list[plugins.EventType] = ["write", "after_write"]
def setUp(self):
super().setUp()
self.paths = [str(self.temp_dir_path / e) for e in self.EVENTS]
def _test_command(
self,
make_test_path: Callable[[str, str], str],
send_path_kwarg: bool = False,
) -> None:
"""Check that each of the configured hooks is executed.
Configure hooks for each event:
1. Use the given 'make_test_path' callable to create a test path from the event
and the original path.
2. Configure a hook with a command to touch this path.
For each of the original paths:
1. Send a test event
2. Assert that a file has been created under the original path, which proves
that the configured hook command has been executed.
"""
events_with_paths = list(zip(self.EVENTS, self.paths))
hooks = [
self._get_hook(e, f"touch {make_test_path(e, p)}")
for e, p in events_with_paths
]
with self.configure_plugin({"hooks": hooks}):
for event, path in events_with_paths:
if send_path_kwarg:
plugins.send(event, path=path)
else:
plugins.send(event)
assert os.path.isfile(path)
@unittest.skipIf(sys.platform == "win32", "win32")
def test_hook_no_arguments(self):
self._test_command(lambda _, p: p)
@unittest.skipIf(sys.platform == "win32", "win32")
def test_hook_event_substitution(self):
self._test_command(lambda e, p: p.replace(e, "{event}"))
@unittest.skipIf(sys.platform == "win32", "win32")
def test_hook_argument_substitution(self):
self._test_command(lambda *_: "{path}", send_path_kwarg=True)
@unittest.skipIf(sys.platform == "win32", "win32")
def test_hook_bytes_interpolation(self):
self.paths = [p.encode() for p in self.paths]
self._test_command(lambda *_: "{path}", send_path_kwarg=True)
|
