341 lines
10 KiB
Python
341 lines
10 KiB
Python
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
SCRIPT_PATH = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "scripts"
|
|
/ "match_filenames.py"
|
|
)
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"match_filenames",
|
|
SCRIPT_PATH,
|
|
)
|
|
if SPEC is None or SPEC.loader is None:
|
|
raise RuntimeError(f"Could not load script: {SCRIPT_PATH}")
|
|
|
|
match_filenames = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = match_filenames
|
|
SPEC.loader.exec_module(match_filenames)
|
|
|
|
|
|
class DetectionReportTests(unittest.TestCase):
|
|
def write_report(
|
|
self,
|
|
directory: Path,
|
|
report: object,
|
|
) -> Path:
|
|
report_file = directory / "report.json"
|
|
report_file.write_text(
|
|
json.dumps(report),
|
|
encoding="utf-8",
|
|
)
|
|
return report_file
|
|
|
|
def valid_report(
|
|
self,
|
|
files: list[dict[str, object]] | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"schema_version": 1,
|
|
"files": files if files is not None else [],
|
|
"future_field": "allowed",
|
|
}
|
|
|
|
def assert_report_error(
|
|
self,
|
|
report: object,
|
|
message: str,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
report_file = self.write_report(
|
|
Path(temporary_directory),
|
|
report,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, message):
|
|
match_filenames.load_detection_report(report_file)
|
|
|
|
def test_valid_schema_version_one(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
directory = Path(temporary_directory)
|
|
video_file = directory / "movie.mp4"
|
|
report_file = self.write_report(
|
|
directory,
|
|
self.valid_report(
|
|
[
|
|
{
|
|
"path": str(video_file),
|
|
"classification": "wcx",
|
|
"extra": 123,
|
|
}
|
|
]
|
|
),
|
|
)
|
|
|
|
entries = match_filenames.load_detection_report(
|
|
report_file
|
|
)
|
|
|
|
self.assertEqual(
|
|
entries,
|
|
{
|
|
match_filenames.normalize_absolute_path(
|
|
video_file
|
|
): "wcx"
|
|
},
|
|
)
|
|
|
|
def test_only_wcx_classification_is_selected(self) -> None:
|
|
inventoried = [
|
|
Path("/videos/wcx.mp4"),
|
|
Path("/videos/uncertain.mp4"),
|
|
Path("/videos/not-wcx.mp4"),
|
|
Path("/videos/error.mp4"),
|
|
]
|
|
report_entries = {
|
|
inventoried[0]: "wcx",
|
|
inventoried[1]: "uncertain",
|
|
inventoried[2]: "not_wcx",
|
|
inventoried[3]: "error",
|
|
}
|
|
|
|
self.assertEqual(
|
|
match_filenames.select_wcx_files(
|
|
inventoried,
|
|
report_entries,
|
|
),
|
|
[inventoried[0]],
|
|
)
|
|
|
|
def test_selection_is_intersection_with_inventory(self) -> None:
|
|
included = Path("/videos/included.mp4")
|
|
not_wcx = Path("/videos/not-wcx.mp4")
|
|
outside = Path("/elsewhere/outside.mp4")
|
|
|
|
selected = match_filenames.select_wcx_files(
|
|
[included, not_wcx],
|
|
{
|
|
included: "wcx",
|
|
not_wcx: "not_wcx",
|
|
outside: "wcx",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(selected, [included])
|
|
|
|
def test_relative_and_absolute_paths_normalize_consistently(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
previous_directory = Path.cwd()
|
|
os.chdir(temporary_directory)
|
|
try:
|
|
relative_path = Path("movie.mp4")
|
|
absolute_path = Path(temporary_directory) / "movie.mp4"
|
|
|
|
selected = match_filenames.select_wcx_files(
|
|
[relative_path],
|
|
{absolute_path: "wcx"},
|
|
)
|
|
finally:
|
|
os.chdir(previous_directory)
|
|
|
|
self.assertEqual(selected, [relative_path])
|
|
|
|
def test_missing_report_file(self) -> None:
|
|
with self.assertRaisesRegex(
|
|
FileNotFoundError,
|
|
"does not exist",
|
|
):
|
|
match_filenames.load_detection_report(
|
|
Path("/missing/detection-report.json")
|
|
)
|
|
|
|
def test_invalid_json(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
report_file = (
|
|
Path(temporary_directory)
|
|
/ "report.json"
|
|
)
|
|
report_file.write_text("{invalid", encoding="utf-8")
|
|
|
|
with self.assertRaisesRegex(
|
|
ValueError,
|
|
"not valid JSON",
|
|
):
|
|
match_filenames.load_detection_report(report_file)
|
|
|
|
def test_top_level_must_be_object(self) -> None:
|
|
self.assert_report_error([], "top level")
|
|
|
|
def test_wrong_schema_version(self) -> None:
|
|
self.assert_report_error(
|
|
{"schema_version": 2, "files": []},
|
|
"schema_version",
|
|
)
|
|
|
|
def test_files_must_be_list(self) -> None:
|
|
self.assert_report_error(
|
|
{"schema_version": 1, "files": {}},
|
|
"files must be a list",
|
|
)
|
|
|
|
def test_missing_path(self) -> None:
|
|
self.assert_report_error(
|
|
self.valid_report(
|
|
[{"classification": "wcx"}]
|
|
),
|
|
"missing path",
|
|
)
|
|
|
|
def test_path_must_be_string(self) -> None:
|
|
self.assert_report_error(
|
|
self.valid_report(
|
|
[{"path": 123, "classification": "wcx"}]
|
|
),
|
|
"path must be a string",
|
|
)
|
|
|
|
def test_missing_classification(self) -> None:
|
|
self.assert_report_error(
|
|
self.valid_report(
|
|
[{"path": "/videos/movie.mp4"}]
|
|
),
|
|
"missing classification",
|
|
)
|
|
|
|
def test_classification_must_be_string(self) -> None:
|
|
self.assert_report_error(
|
|
self.valid_report(
|
|
[
|
|
{
|
|
"path": "/videos/movie.mp4",
|
|
"classification": 1,
|
|
}
|
|
]
|
|
),
|
|
"classification must be a string",
|
|
)
|
|
|
|
def test_unknown_classification(self) -> None:
|
|
self.assert_report_error(
|
|
self.valid_report(
|
|
[
|
|
{
|
|
"path": "/videos/movie.mp4",
|
|
"classification": "maybe",
|
|
}
|
|
]
|
|
),
|
|
"unknown classification",
|
|
)
|
|
|
|
def test_duplicate_normalized_path(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
directory = Path(temporary_directory)
|
|
path = directory / "movie.mp4"
|
|
self.assert_report_error(
|
|
self.valid_report(
|
|
[
|
|
{
|
|
"path": str(path),
|
|
"classification": "wcx",
|
|
},
|
|
{
|
|
"path": str(
|
|
directory
|
|
/ "."
|
|
/ "movie.mp4"
|
|
),
|
|
"classification": "not_wcx",
|
|
},
|
|
]
|
|
),
|
|
"duplicate path",
|
|
)
|
|
|
|
def test_empty_report_selects_nothing(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
directory = Path(temporary_directory)
|
|
report_file = self.write_report(
|
|
directory,
|
|
self.valid_report(),
|
|
)
|
|
entries = match_filenames.load_detection_report(
|
|
report_file
|
|
)
|
|
|
|
self.assertEqual(entries, {})
|
|
self.assertEqual(
|
|
match_filenames.select_wcx_files(
|
|
[Path("/videos/movie.mp4")],
|
|
entries,
|
|
),
|
|
[],
|
|
)
|
|
|
|
def test_without_detection_report_argument_is_unchanged(self) -> None:
|
|
arguments = [
|
|
"match_filenames.py",
|
|
"/videos",
|
|
"--database",
|
|
"/database/wcx.db",
|
|
]
|
|
with patch.object(sys, "argv", arguments):
|
|
args = match_filenames.parse_arguments()
|
|
|
|
self.assertIsNone(args.detection_report)
|
|
self.assertEqual(args.directory, Path("/videos"))
|
|
|
|
def test_empty_selection_does_not_open_database_or_run_ffprobe(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
directory = Path(temporary_directory)
|
|
video_file = directory / "movie.mp4"
|
|
video_file.touch()
|
|
report_file = self.write_report(
|
|
directory,
|
|
self.valid_report(
|
|
[
|
|
{
|
|
"path": str(video_file),
|
|
"classification": "not_wcx",
|
|
}
|
|
]
|
|
),
|
|
)
|
|
arguments = [
|
|
"match_filenames.py",
|
|
str(directory),
|
|
"--database",
|
|
"/missing/wcx.db",
|
|
"--detection-report",
|
|
str(report_file),
|
|
]
|
|
|
|
with patch.object(sys, "argv", arguments):
|
|
with patch.object(
|
|
match_filenames,
|
|
"load_movies",
|
|
) as load_movies:
|
|
with patch.object(
|
|
match_filenames,
|
|
"read_file_duration",
|
|
) as read_file_duration:
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
match_filenames.main()
|
|
|
|
load_movies.assert_not_called()
|
|
read_file_duration.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|