563 lines
27 KiB
Python
563 lines
27 KiB
Python
"""Native ComfyUI VIDEO tests, isolated from the custom-node import graph.
|
|
|
|
The outer pytest process starts one bounded CPU-only subprocess. Inside it the
|
|
real pinned core is imported *after* cli_args.args.cpu=True; no fake VideoInput
|
|
classes, media transcodes, providers, server, or GPU queries are needed.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
|
|
if os.environ.get("_DISTRIBUTED_VIDEO_CODEC_CORE_TEST") != "1":
|
|
def test_native_video_codec_in_cpu_subprocess():
|
|
configured = os.environ.get("COMFYUI_SOURCE_ROOT", os.environ.get("COMFYUI_VIDEO_TEST_ROOT"))
|
|
if configured is None:
|
|
pytest.skip("Set COMFYUI_SOURCE_ROOT for CPU native VIDEO acceptance")
|
|
core = Path(configured).expanduser().resolve()
|
|
assert configured and (core / "comfy_api/latest/_input_impl/video_types.py").is_file(), (
|
|
"Configured COMFYUI_SOURCE_ROOT (or COMFYUI_VIDEO_TEST_ROOT) must provide native video classes"
|
|
)
|
|
env = dict(os.environ)
|
|
scratch = Path(os.environ.get("TMPDIR", str(Path.home() / ".hermes/cache/scratch")))
|
|
scratch.mkdir(parents=True, exist_ok=True)
|
|
env["TMPDIR"] = str(scratch)
|
|
for name in ("PYTHONPATH", "COMFYUI_SOURCE_ROOT"):
|
|
env.pop(name, None)
|
|
env.update(CUDA_VISIBLE_DEVICES="", HIP_VISIBLE_DEVICES="",
|
|
_DISTRIBUTED_VIDEO_CODEC_CORE_TEST="1", PYTHONDONTWRITEBYTECODE="1")
|
|
bootstrap = """
|
|
import sys
|
|
sys.path.insert(0, sys.argv[1])
|
|
import comfy.cli_args
|
|
comfy.cli_args.args.cpu = True
|
|
import torch
|
|
def forbidden(*args, **kwargs):
|
|
raise AssertionError('VIDEO tests must not query or initialize a GPU')
|
|
# Third-party core imports probe availability even in CPU mode: answer without
|
|
# executing a GPU query, then forbid even these probes during codec execution.
|
|
torch.cuda.is_available = lambda: False
|
|
torch.cuda.device_count = lambda: 0
|
|
for name in ('init', '_lazy_init', 'current_device'):
|
|
setattr(torch.cuda, name, forbidden)
|
|
import comfy_api.latest._input_impl.video_types
|
|
for name in ('is_available', 'device_count'):
|
|
setattr(torch.cuda, name, forbidden)
|
|
import pytest
|
|
raise SystemExit(pytest.main([sys.argv[2], '-q', '-p', 'no:cacheprovider']))
|
|
"""
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", bootstrap, str(core), str(Path(__file__).resolve())],
|
|
env=env, text=True, capture_output=True, timeout=150,
|
|
cwd=Path(__file__).resolve().parents[1],
|
|
)
|
|
print(result.stdout)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
else:
|
|
import copy
|
|
from fractions import Fraction
|
|
import importlib
|
|
import io
|
|
import json
|
|
import math
|
|
import types
|
|
|
|
import av
|
|
import numpy as np
|
|
import torch
|
|
from comfy_api.latest._input import VideoInput
|
|
from comfy_api.latest._input_impl.video_types import (
|
|
VideoFromComponents, VideoFromFile, VideoFromList,
|
|
)
|
|
from comfy_api.latest._util import VideoComponents
|
|
|
|
# Only this sibling package is synthetic; all native media types are real.
|
|
package_name = "_distributed_video_codec_test"
|
|
package = types.ModuleType(package_name)
|
|
package.__path__ = [str(Path(__file__).resolve().parents[1] / "serverless_bridge")]
|
|
sys.modules[package_name] = package
|
|
codec = importlib.import_module(package_name + ".video_codec")
|
|
|
|
def audio(dtype=torch.float64):
|
|
return {"waveform": torch.tensor([[[0.123456789, -0.0, -1.5, 2.0]]], dtype=dtype, device="cpu"),
|
|
"sample_rate": 48000}
|
|
|
|
def components(dtype=torch.float64, *, metadata=None, alpha=True, sound=True):
|
|
image = torch.arange(36, device="cpu", dtype=torch.float64).reshape(2, 2, 3, 3) / 997
|
|
image[0, 0, 0, 0] = -0.0
|
|
return VideoFromComponents(VideoComponents(
|
|
images=image.to(dtype), frame_rate=Fraction(30000, 1001),
|
|
audio=audio(dtype) if sound else None,
|
|
alpha=torch.full((2, 2, 3), 0.123456789, dtype=dtype, device="cpu") if alpha else None,
|
|
metadata=metadata,
|
|
), bit_depth=10, color_space="HDR")
|
|
|
|
def exact_tensor(actual, expected):
|
|
assert actual.device.type == "cpu"
|
|
assert actual.dtype == expected.dtype
|
|
assert actual.shape == expected.shape
|
|
assert torch.equal(actual.contiguous().view(torch.uint8), expected.contiguous().view(torch.uint8))
|
|
|
|
def file_value(data=b"opaque original media bytes", **kwargs):
|
|
return VideoFromFile(io.BytesIO(data), **kwargs)
|
|
|
|
def forbid(*args, **kwargs):
|
|
raise AssertionError("Unexpected materialization/transcoding/allocation")
|
|
|
|
@pytest.mark.parametrize("dtype", [torch.float16, torch.bfloat16, torch.float32, torch.float64])
|
|
@pytest.mark.parametrize("alpha,sound", [(False, False), (True, False), (False, True), (True, True)])
|
|
def test_components_exact_roundtrip(dtype, alpha, sound, monkeypatch):
|
|
metadata = {"title": "雪 🎞", "flags": [True, None, -4, 0.125], "nested": {"empty": []}}
|
|
source = components(dtype, metadata=metadata, alpha=alpha, sound=sound)
|
|
expected = source.get_components()
|
|
monkeypatch.setattr(VideoFromComponents, "get_components", forbid)
|
|
monkeypatch.setattr(VideoFromComponents, "save_to", forbid)
|
|
descriptor, payload = codec.encode_video(source)
|
|
assert descriptor["encoding"] == "native-video-v1"
|
|
assert descriptor["representation"] == "components"
|
|
assert descriptor["frame_rate"] == {"numerator": 30000, "denominator": 1001}
|
|
# This is the actual JSON wire descriptor, not a Python-only shortcut.
|
|
descriptor = json.loads(json.dumps(descriptor, allow_nan=False))
|
|
result = codec.decode_video(descriptor, payload)
|
|
assert type(result) is VideoFromComponents
|
|
assert isinstance(result, VideoInput)
|
|
actual = vars(result)["_VideoFromComponents__components"]
|
|
exact_tensor(actual.images, expected.images)
|
|
assert actual.frame_rate == expected.frame_rate
|
|
assert type(actual.frame_rate) is Fraction
|
|
assert actual.metadata == metadata
|
|
assert actual.metadata is not metadata
|
|
assert vars(result)["_VideoFromComponents__bit_depth"] == 10
|
|
assert vars(result)["_VideoFromComponents__color_space"] == "HDR"
|
|
if alpha:
|
|
exact_tensor(actual.alpha, expected.alpha)
|
|
else:
|
|
assert actual.alpha is None
|
|
if sound:
|
|
exact_tensor(actual.audio["waveform"], expected.audio["waveform"])
|
|
assert actual.audio["sample_rate"] == expected.audio["sample_rate"]
|
|
else:
|
|
assert actual.audio is None
|
|
actual.images.fill_(99)
|
|
assert not torch.equal(actual.images, expected.images)
|
|
actual.metadata["nested"]["empty"].append("new")
|
|
assert descriptor["metadata"]["nested"]["empty"] == []
|
|
assert metadata["nested"]["empty"] == []
|
|
|
|
@pytest.mark.parametrize("dtype", [torch.float16, torch.bfloat16, torch.float32, torch.float64])
|
|
def test_native_bhw1_alpha_shape_is_preserved(dtype):
|
|
source = components(dtype)
|
|
original = source.get_components()
|
|
vars(source)["_VideoFromComponents__components"].alpha = original.alpha.unsqueeze(-1)
|
|
descriptor, payload = codec.encode_video(source)
|
|
result = codec.decode_video(descriptor, payload).get_components()
|
|
exact_tensor(result.alpha, original.alpha.unsqueeze(-1))
|
|
assert descriptor["alpha"]["descriptor"]["shape"] == [2, 2, 3, 1]
|
|
|
|
@pytest.mark.parametrize("start,duration,crop", [
|
|
(-1.25, -2.5, (-3, 5, 7, -9)), (0, 0, None), (-0.0, 0.0, (0, 0, 2, 2)),
|
|
])
|
|
def test_file_preserves_bytes_cursor_signed_trim_crop_and_ownership(start, duration, crop, monkeypatch):
|
|
source = file_value(start_time=start, duration=duration, crop=crop)
|
|
stream = vars(source)["_VideoFromFile__file"]
|
|
stream.seek(7)
|
|
monkeypatch.setattr(VideoFromFile, "get_components", forbid)
|
|
monkeypatch.setattr(VideoFromFile, "save_to", forbid)
|
|
descriptor, payload = codec.encode_video(source)
|
|
assert stream.tell() == 7
|
|
assert payload == stream.getvalue()
|
|
assert descriptor["start_time"] == start
|
|
assert descriptor["duration"] == duration
|
|
result = codec.decode_video(json.loads(json.dumps(descriptor)), payload)
|
|
second = codec.decode_video(descriptor, payload)
|
|
attrs = vars(result)
|
|
assert type(result) is VideoFromFile
|
|
assert attrs["_VideoFromFile__start_time"] == start
|
|
assert type(attrs["_VideoFromFile__start_time"]) is type(start)
|
|
assert attrs["_VideoFromFile__duration"] == duration
|
|
assert attrs["_VideoFromFile__crop"] == crop
|
|
if type(start) is float:
|
|
assert math.copysign(1, attrs["_VideoFromFile__start_time"]) == math.copysign(1, start)
|
|
decoded = attrs["_VideoFromFile__file"]
|
|
assert type(decoded) is io.BytesIO
|
|
assert decoded is not stream
|
|
stream.close()
|
|
assert decoded.getvalue() == payload
|
|
decoded.write(b"changed")
|
|
assert vars(second)["_VideoFromFile__file"].getvalue() == payload
|
|
|
|
def test_real_tiny_pyav_video_is_still_readable_after_source_closed():
|
|
buffer = io.BytesIO()
|
|
with av.open(buffer, "w", format="matroska") as container:
|
|
stream = container.add_stream("ffv1", rate=Fraction(24, 1))
|
|
# Width 32 avoids a native fillborders filter limitation on tiny
|
|
# unaligned YUV frames; this is still a tiny CPU-only fixture.
|
|
stream.width = stream.height = 32
|
|
stream.pix_fmt = "yuv420p"
|
|
for value in (16, 64, 128):
|
|
frame = av.VideoFrame.from_ndarray(np.full((32, 32, 3), value, dtype=np.uint8), format="rgb24")
|
|
for packet in stream.encode(frame):
|
|
container.mux(packet)
|
|
for packet in stream.encode():
|
|
container.mux(packet)
|
|
data = buffer.getvalue()
|
|
source = VideoFromFile(buffer, crop=(0, 0, 2, 2))
|
|
descriptor, payload = codec.encode_video(source)
|
|
expected = source.get_components()
|
|
result = codec.decode_video(descriptor, payload)
|
|
buffer.close()
|
|
assert payload == data
|
|
actual = result.get_components()
|
|
exact_tensor(actual.images, expected.images)
|
|
assert actual.frame_rate == expected.frame_rate
|
|
assert tuple(actual.images.shape) == (3, 2, 2, 3)
|
|
|
|
def test_segments_order_complete_audio_no_transcode_and_independent_buffers(monkeypatch):
|
|
values = [file_value(b"first", start_time=-2), file_value(b"second", duration=-4, crop=(1, 2, 3, 4))]
|
|
source = VideoFromList([VideoFromList(values[:1]), values[1]], complete_audio=audio())
|
|
monkeypatch.setattr(VideoFromFile, "get_components", forbid)
|
|
monkeypatch.setattr(VideoFromFile, "save_to", forbid)
|
|
monkeypatch.setattr(VideoFromList, "get_components", forbid)
|
|
monkeypatch.setattr(VideoFromList, "save_to", forbid)
|
|
descriptor, payload = codec.encode_video(source)
|
|
assert descriptor["representation"] == "segments"
|
|
assert payload.startswith(b"firstsecond")
|
|
assert len(descriptor["segments"]) == 2
|
|
decoded = codec.decode_video(json.loads(json.dumps(descriptor)), payload)
|
|
assert type(decoded) is VideoFromList
|
|
assert [vars(v)["_VideoFromFile__file"].getvalue() for v in decoded.videos] == [b"first", b"second"]
|
|
exact_tensor(decoded.complete_audio["waveform"], source.complete_audio["waveform"])
|
|
for old, new in zip(source.videos, decoded.videos):
|
|
assert old is not new
|
|
vars(old)["_VideoFromFile__file"].close()
|
|
assert not vars(new)["_VideoFromFile__file"].closed
|
|
assert vars(decoded.videos[0])["_VideoFromFile__start_time"] == -2
|
|
assert vars(decoded.videos[1])["_VideoFromFile__crop"] == (1, 2, 3, 4)
|
|
|
|
def test_path_roots_symlinks_and_no_path_in_wire(tmp_path, monkeypatch):
|
|
allowed = tmp_path / "allowed"
|
|
allowed.mkdir()
|
|
media = allowed / "private-name.mkv"
|
|
media.write_bytes(b"media bytes")
|
|
sibling = tmp_path / "allowed-sibling"
|
|
sibling.mkdir()
|
|
secret = sibling / "secret.mkv"
|
|
secret.write_bytes(b"private")
|
|
(allowed / "escape.mkv").symlink_to(secret)
|
|
(allowed / "inside.mkv").symlink_to(media)
|
|
descriptor, payload = codec.encode_video(VideoFromFile(str(media)), allowed_roots=[allowed])
|
|
assert payload == b"media bytes"
|
|
assert "private-name" not in json.dumps(descriptor)
|
|
assert codec.encode_video(VideoFromFile(str(allowed / "inside.mkv")), allowed_roots=[allowed])[1] == payload
|
|
for path in (secret, allowed / "escape.mkv", allowed, allowed / "missing"):
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(VideoFromFile(str(path)), allowed_roots=[allowed])
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(VideoFromFile(str(media)), allowed_roots=[])
|
|
import folder_paths
|
|
for name in ("get_input_directory", "get_output_directory", "get_temp_directory"):
|
|
monkeypatch.setattr(folder_paths, name, lambda: str(allowed))
|
|
assert codec.encode_video(VideoFromFile(str(media)))[1] == payload
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(VideoFromFile(str(secret)))
|
|
result = codec.decode_video(descriptor, payload)
|
|
media.unlink()
|
|
assert vars(result)["_VideoFromFile__file"].getvalue() == b"media bytes"
|
|
|
|
def test_path_special_files_and_symlink_swap_are_rejected(tmp_path, monkeypatch):
|
|
fifo = tmp_path / "fifo"
|
|
os.mkfifo(fifo)
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(VideoFromFile(str(fifo)), allowed_roots=[tmp_path])
|
|
allowed = tmp_path / "allowed"
|
|
allowed.mkdir()
|
|
actual = allowed / "subdir"
|
|
actual.mkdir()
|
|
media = actual / "clip.mkv"
|
|
media.write_bytes(b"allowed")
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(outside / "clip.mkv").write_bytes(b"forbidden")
|
|
real_resolve = codec.Path.resolve
|
|
swapped = False
|
|
|
|
def swap_after_resolution(path, *args, **kwargs):
|
|
nonlocal swapped
|
|
resolved = real_resolve(path, *args, **kwargs)
|
|
if path == media and not swapped:
|
|
swapped = True
|
|
actual.rename(allowed / "old")
|
|
actual.symlink_to(outside, target_is_directory=True)
|
|
return resolved
|
|
|
|
monkeypatch.setattr(codec.Path, "resolve", swap_after_resolution)
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(VideoFromFile(str(media)), allowed_roots=[allowed])
|
|
assert swapped
|
|
|
|
def test_file_size_budget_checked_before_read(tmp_path, monkeypatch):
|
|
path = tmp_path / "large.mkv"
|
|
path.write_bytes(b"12345")
|
|
monkeypatch.setattr(codec.os, "read", forbid)
|
|
with pytest.raises(ValueError, match="limit|budget"):
|
|
codec.encode_video(VideoFromFile(str(path)), max_bytes=4, allowed_roots=[tmp_path])
|
|
|
|
@pytest.mark.parametrize("limit", [True, False, -1, 1.5, None])
|
|
def test_invalid_limits(limit):
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(file_value(), max_bytes=limit)
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video({}, b"", max_bytes=limit)
|
|
|
|
def test_aggregate_budget_and_exact_boundary():
|
|
value = VideoFromList([file_value(b"12"), file_value(b"345")])
|
|
descriptor, payload = codec.encode_video(value, max_bytes=5)
|
|
assert payload == b"12345"
|
|
codec.decode_video(descriptor, payload, max_bytes=5)
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value, max_bytes=4)
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload, max_bytes=4)
|
|
stream = vars(value.videos[0])["_VideoFromFile__file"]
|
|
stream.seek(1)
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value.videos[0], max_bytes=1)
|
|
assert stream.tell() == 1
|
|
|
|
@pytest.mark.parametrize("kind", ["components", "file", "segments"])
|
|
def test_unknown_subclasses_never_materialized(kind):
|
|
bases = {"components": VideoFromComponents, "file": VideoFromFile, "segments": VideoFromList}
|
|
cls = type("UnknownVideo", (bases[kind],), {"get_components": forbid, "save_to": forbid})
|
|
value = object.__new__(cls)
|
|
with pytest.raises(ValueError, match="class|type|unsupported|Unsupported"):
|
|
codec.encode_video(value)
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(object())
|
|
|
|
@pytest.mark.parametrize("kind", ["components", "file", "segments"])
|
|
def test_bridge_rejects_subclasses_without_materializing(kind, monkeypatch):
|
|
nodes = importlib.import_module(package_name + ".nodes")
|
|
bases = {"components": VideoFromComponents, "file": VideoFromFile, "segments": VideoFromList}
|
|
cls = type("UnknownVideo", (bases[kind],), {"get_components": forbid, "save_to": forbid})
|
|
monkeypatch.setattr(torch.Tensor, "cpu", forbid)
|
|
with pytest.raises(ValueError, match="Unsupported native VIDEO class"):
|
|
nodes.encode_value("VIDEO", object.__new__(cls))
|
|
|
|
@pytest.mark.parametrize("dtype", [torch.float16, torch.bfloat16, torch.float32, torch.float64])
|
|
@pytest.mark.parametrize("kind", ["components", "segments", "file"])
|
|
def test_bridge_stages_cpu_tensors_without_materialization(dtype, kind, monkeypatch):
|
|
nodes = importlib.import_module(package_name + ".nodes")
|
|
if kind == "components":
|
|
source = components(dtype, metadata={"nested": ["雪", None, True]})
|
|
state = vars(source)["_VideoFromComponents__components"]
|
|
state.alpha = state.alpha.unsqueeze(-1)
|
|
tensors = [state.images, state.alpha, state.audio["waveform"]]
|
|
elif kind == "segments":
|
|
source = VideoFromList([file_value(start_time=-1.5, duration=-2, crop=(-1, 2, 3, 4))],
|
|
complete_audio=audio(dtype))
|
|
tensors = [source.complete_audio["waveform"]]
|
|
else:
|
|
source = file_value(start_time=-0.0, duration=-2, crop=(-1, 2, 3, 4))
|
|
tensors = []
|
|
for tensor in tensors:
|
|
tensor.requires_grad_(True)
|
|
expected = codec.encode_video(source)
|
|
detached, staged = [], []
|
|
real_detach, real_cpu = torch.Tensor.detach, torch.Tensor.cpu
|
|
|
|
def detach(tensor):
|
|
detached.append(id(tensor))
|
|
return real_detach(tensor)
|
|
|
|
def cpu(tensor, *args, **kwargs):
|
|
assert not tensor.requires_grad
|
|
staged.append(tensor)
|
|
return real_cpu(tensor, *args, **kwargs)
|
|
|
|
for cls in (VideoFromComponents, VideoFromFile, VideoFromList):
|
|
monkeypatch.setattr(cls, "get_components", forbid)
|
|
monkeypatch.setattr(cls, "save_to", forbid)
|
|
monkeypatch.setattr(torch.Tensor, "detach", detach)
|
|
monkeypatch.setattr(torch.Tensor, "cpu", cpu)
|
|
actual = nodes.encode_value("VIDEO", source)
|
|
assert len(staged) == len(tensors)
|
|
assert all(id(tensor) in detached for tensor in tensors)
|
|
assert all(tensor.requires_grad for tensor in tensors)
|
|
assert actual == expected
|
|
decoded = nodes.decode_value("VIDEO", *actual)
|
|
assert type(decoded) is type(source)
|
|
assert codec.encode_video(decoded) == expected
|
|
assert len(staged) == len(tensors) # Pure codec never stages implicitly.
|
|
|
|
@pytest.mark.parametrize("field", ["images", "alpha", "audio", "complete_audio"])
|
|
def test_pure_video_rejects_non_cpu_storage_without_transfer(field, monkeypatch):
|
|
if field == "complete_audio":
|
|
source = VideoFromList([file_value()], complete_audio=audio())
|
|
target = source.complete_audio["waveform"]
|
|
else:
|
|
source = components()
|
|
target = getattr(vars(source)["_VideoFromComponents__components"], field)
|
|
if field == "audio":
|
|
target = target["waveform"]
|
|
# A property spy on real CPU tensors exercises the guard without a GPU
|
|
# allocation, subclass, device discovery or CUDA initialization.
|
|
real_device = torch.Tensor.device
|
|
monkeypatch.setattr(torch.Tensor, "device", property(
|
|
lambda tensor: types.SimpleNamespace(type="cuda") if tensor is target
|
|
else real_device.__get__(tensor)))
|
|
monkeypatch.setattr(torch.Tensor, "cpu", forbid)
|
|
with pytest.raises(ValueError, match="must be on CPU"):
|
|
codec.encode_video(source)
|
|
|
|
@pytest.mark.parametrize("field,bad", [
|
|
("_VideoFromFile__start_time", True), ("_VideoFromFile__start_time", float("nan")),
|
|
("_VideoFromFile__duration", float("inf")), ("_VideoFromFile__crop", (1, 2, True, 4)),
|
|
("_VideoFromFile__crop", [1, 2, 3, 4]), ("_VideoFromFile__crop", (1, 2, 3)),
|
|
("_VideoFromFile__file", b"not a BytesIO"),
|
|
])
|
|
def test_invalid_native_file_attributes(field, bad):
|
|
value = file_value()
|
|
setattr(value, field, bad)
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value)
|
|
|
|
def test_missing_attributes_and_invalid_component_state():
|
|
value = file_value()
|
|
del vars(value)["_VideoFromFile__duration"]
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value)
|
|
for field, bad in [("frame_rate", 24.0), ("frame_rate", Fraction(0)),
|
|
("alpha", torch.ones(1, 1, 1, device="cpu"))]:
|
|
value = components()
|
|
setattr(vars(value)["_VideoFromComponents__components"], field, bad)
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value)
|
|
|
|
@pytest.mark.parametrize("metadata", [
|
|
{"value": float("nan")}, {"value": float("inf")}, {1: "non-string key"},
|
|
{"value": (1, 2)}, {"value": object()}, {"large": "x" * 65536},
|
|
])
|
|
def test_invalid_metadata_rejected_both_directions(metadata):
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(components(metadata=metadata))
|
|
descriptor, payload = codec.encode_video(components())
|
|
descriptor["metadata"] = metadata
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
|
|
def test_metadata_depth_cycles_and_exact_size():
|
|
metadata = {}
|
|
current = metadata
|
|
for _ in range(40):
|
|
current["child"] = {}
|
|
current = current["child"]
|
|
for value in (metadata,):
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(components(metadata=value))
|
|
cyclic = {}
|
|
cyclic["self"] = cyclic
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(components(metadata=cyclic))
|
|
exact = {"x": "a" * (65536 - 8)}
|
|
assert len(json.dumps(exact, separators=(",", ":")).encode()) == 65536
|
|
descriptor, payload = codec.encode_video(components(metadata=exact))
|
|
assert codec.decode_video(descriptor, payload).get_components().metadata == exact
|
|
|
|
@pytest.mark.parametrize("field,bad", [
|
|
("encoding", "other"), ("representation", "other"), ("byte_length", True),
|
|
("byte_length", -1), ("byte_length", 12345), ("start_time", False),
|
|
("duration", float("inf")), ("crop", [0, 0, True, 4]), ("crop", [1, 2, 3]),
|
|
])
|
|
def test_malformed_file_descriptors(field, bad):
|
|
descriptor, payload = codec.encode_video(file_value())
|
|
descriptor[field] = bad
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
|
|
@pytest.mark.parametrize("field,bad", [("offset", True), ("offset", -1), ("offset", 1),
|
|
("length", True), ("length", -1), ("length", 2**80)])
|
|
def test_invalid_offsets_lengths(field, bad):
|
|
descriptor, payload = codec.encode_video(file_value())
|
|
descriptor["data"][field] = bad
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
|
|
def test_truncation_trailing_gaps_overlap_and_unknown_fields():
|
|
value = VideoFromList([file_value(b"123"), file_value(b"456")])
|
|
original, payload = codec.encode_video(value)
|
|
for wire in (payload[:-1], payload + b"x"):
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(original, wire)
|
|
for offset in (2, 4):
|
|
descriptor = copy.deepcopy(original)
|
|
descriptor["segments"][1]["offset"] = offset
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
descriptor = copy.deepcopy(original)
|
|
descriptor["surprise"] = "bad"
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
descriptor = copy.deepcopy(original)
|
|
descriptor["segments"][0]["descriptor"]["byte_length"] += 1
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
descriptor = copy.deepcopy(original)
|
|
descriptor["segments"][0]["descriptor"] = descriptor
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
|
|
def test_all_descriptors_validated_before_tensor_allocation(monkeypatch):
|
|
original, payload = codec.encode_video(components())
|
|
monkeypatch.setattr(codec, "decode_payload", forbid)
|
|
cases = []
|
|
for key, value in [("shape", [2**80, 2, 3]), ("shape", [True, 2, 3]),
|
|
("dtype", "pickle"), ("layout", "BHWC")]:
|
|
descriptor = copy.deepcopy(original)
|
|
descriptor["alpha"]["descriptor"][key] = value
|
|
cases.append(descriptor)
|
|
descriptor = copy.deepcopy(original)
|
|
descriptor["audio"]["descriptor"]["sample_rate"] = True
|
|
cases.append(descriptor)
|
|
descriptor = copy.deepcopy(original)
|
|
descriptor["alpha"]["offset"] += 1
|
|
cases.append(descriptor)
|
|
for descriptor in cases:
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
|
|
@pytest.mark.parametrize("field,bad", [("numerator", True), ("denominator", True),
|
|
("denominator", 0), ("numerator", -3),
|
|
("denominator", -1001)])
|
|
def test_invalid_fraction_fields(field, bad):
|
|
descriptor, payload = codec.encode_video(components())
|
|
descriptor["frame_rate"][field] = bad
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
|
|
def test_segment_limit_and_exact_maximum(monkeypatch):
|
|
value = VideoFromList([file_value(b"x") for _ in range(64)])
|
|
descriptor, payload = codec.encode_video(value)
|
|
result = codec.decode_video(descriptor, payload)
|
|
assert len(result.videos) == 64
|
|
value.videos.append(file_value(b"x"))
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value)
|
|
descriptor["segments"].append(copy.deepcopy(descriptor["segments"][-1]))
|
|
monkeypatch.setattr(codec, "decode_payload", forbid)
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|
|
value.videos = []
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value)
|
|
value.videos = [components()]
|
|
with pytest.raises(ValueError):
|
|
codec.encode_video(value)
|
|
|
|
@pytest.mark.parametrize("descriptor,payload", [(None, b""), ([], b""), ({}, bytearray()), ({}, "")])
|
|
def test_invalid_top_level_types(descriptor, payload):
|
|
with pytest.raises(ValueError):
|
|
codec.decode_video(descriptor, payload)
|