Initial import
This commit is contained in:
@@ -0,0 +1,131 @@
|
||||
"""Tests for the DLNAClient SOAP and DIDL-Lite parsers."""
|
||||
|
||||
from r36s_dlna_browser.dlna.client import _extract_browse_result, _parse_didl
|
||||
from r36s_dlna_browser.dlna.models import ItemType
|
||||
|
||||
|
||||
_SAMPLE_DIDL = """\
|
||||
<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/"
|
||||
xmlns:dc="http://purl.org/dc/elements/1.1/"
|
||||
xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/">
|
||||
<container id="1" parentID="0" childCount="3">
|
||||
<dc:title>Music</dc:title>
|
||||
<upnp:class>object.container.storageFolder</upnp:class>
|
||||
</container>
|
||||
<item id="100" parentID="1">
|
||||
<dc:title>Song.mp3</dc:title>
|
||||
<upnp:class>object.item.audioItem.musicTrack</upnp:class>
|
||||
<upnp:albumArtURI>http://srv/art.jpg</upnp:albumArtURI>
|
||||
<res protocolInfo="http-get:*:audio/mpeg:*" size="4000000" duration="0:03:21">http://srv/song.mp3</res>
|
||||
</item>
|
||||
<item id="200" parentID="1">
|
||||
<dc:title>Video.mkv</dc:title>
|
||||
<upnp:class>object.item.videoItem</upnp:class>
|
||||
<res protocolInfo="http-get:*:video/x-matroska:*">http://srv/video.mkv</res>
|
||||
</item>
|
||||
</DIDL-Lite>
|
||||
"""
|
||||
|
||||
|
||||
class TestParseDIDL:
|
||||
def test_parses_container(self):
|
||||
items = _parse_didl(_SAMPLE_DIDL, "http://srv")
|
||||
containers = [i for i in items if i.is_container]
|
||||
assert len(containers) == 1
|
||||
assert containers[0].title == "Music"
|
||||
assert containers[0].child_count == 3
|
||||
|
||||
def test_parses_audio_item(self):
|
||||
items = _parse_didl(_SAMPLE_DIDL, "http://srv")
|
||||
audio = [i for i in items if i.item_type == ItemType.AUDIO]
|
||||
assert len(audio) == 1
|
||||
assert audio[0].title == "Song.mp3"
|
||||
assert audio[0].resource_url == "http://srv/song.mp3"
|
||||
assert audio[0].mime_type == "audio/mpeg"
|
||||
assert audio[0].size == 4000000
|
||||
assert audio[0].duration == "0:03:21"
|
||||
assert audio[0].album_art_url == "http://srv/art.jpg"
|
||||
|
||||
def test_parses_video_item(self):
|
||||
items = _parse_didl(_SAMPLE_DIDL, "http://srv")
|
||||
video = [i for i in items if i.item_type == ItemType.VIDEO]
|
||||
assert len(video) == 1
|
||||
assert video[0].title == "Video.mkv"
|
||||
assert "video/x-matroska" in video[0].mime_type
|
||||
|
||||
def test_total_count(self):
|
||||
items = _parse_didl(_SAMPLE_DIDL, "http://srv")
|
||||
assert len(items) == 3
|
||||
|
||||
def test_empty_didl(self):
|
||||
empty = '<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/"></DIDL-Lite>'
|
||||
assert _parse_didl(empty, "http://srv") == []
|
||||
|
||||
def test_malformed_xml(self):
|
||||
assert _parse_didl("<<<not xml>>>", "http://srv") == []
|
||||
|
||||
def test_relative_url_resolution(self):
|
||||
didl = """\
|
||||
<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/"
|
||||
xmlns:dc="http://purl.org/dc/elements/1.1/"
|
||||
xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/">
|
||||
<item id="300" parentID="1">
|
||||
<dc:title>Relative</dc:title>
|
||||
<upnp:class>object.item.audioItem</upnp:class>
|
||||
<res protocolInfo="http-get:*:audio/flac:*">media/track.flac</res>
|
||||
</item>
|
||||
</DIDL-Lite>
|
||||
"""
|
||||
items = _parse_didl(didl, "http://myserver:8200")
|
||||
assert items[0].resource_url == "http://myserver:8200/media/track.flac"
|
||||
|
||||
|
||||
class TestExtractBrowseResult:
|
||||
def test_extracts_unnamespaced_result(self):
|
||||
soap = """\
|
||||
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/">
|
||||
<s:Body>
|
||||
<u:BrowseResponse xmlns:u="urn:schemas-upnp-org:service:ContentDirectory:1">
|
||||
<Result><DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/"><container id="1" parentID="0" /></DIDL-Lite></Result>
|
||||
</u:BrowseResponse>
|
||||
</s:Body>
|
||||
</s:Envelope>
|
||||
"""
|
||||
result = _extract_browse_result(soap)
|
||||
assert result is not None
|
||||
assert "DIDL-Lite" in result
|
||||
assert 'container id="1"' in result
|
||||
|
||||
def test_extracts_namespaced_result(self):
|
||||
soap = """\
|
||||
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" xmlns:u="urn:schemas-upnp-org:service:ContentDirectory:1">
|
||||
<s:Body>
|
||||
<u:BrowseResponse>
|
||||
<u:Result><DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/"></DIDL-Lite></u:Result>
|
||||
</u:BrowseResponse>
|
||||
</s:Body>
|
||||
</s:Envelope>
|
||||
"""
|
||||
assert _extract_browse_result(soap) is not None
|
||||
|
||||
def test_extracts_embedded_xml_result(self):
|
||||
soap = """\
|
||||
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/">
|
||||
<s:Body>
|
||||
<BrowseResponse>
|
||||
<Result>
|
||||
<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/">
|
||||
<container id="1" parentID="0" />
|
||||
</DIDL-Lite>
|
||||
</Result>
|
||||
</BrowseResponse>
|
||||
</s:Body>
|
||||
</s:Envelope>
|
||||
"""
|
||||
result = _extract_browse_result(soap)
|
||||
assert result is not None
|
||||
assert "DIDL-Lite" in result
|
||||
|
||||
def test_returns_none_when_missing(self):
|
||||
soap = "<Envelope><Body><BrowseResponse /></Body></Envelope>"
|
||||
assert _extract_browse_result(soap) is None
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Tests for platform/controls input mapping."""
|
||||
|
||||
import ctypes
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import sdl2
|
||||
|
||||
from r36s_dlna_browser.platform.controls import Action, map_key
|
||||
|
||||
|
||||
def _make_key_event(sym: int) -> sdl2.SDL_Event:
|
||||
event = sdl2.SDL_Event()
|
||||
event.type = sdl2.SDL_KEYDOWN
|
||||
event.key.keysym.sym = sym
|
||||
return event
|
||||
|
||||
|
||||
def _make_button_event(button: int) -> sdl2.SDL_Event:
|
||||
event = sdl2.SDL_Event()
|
||||
event.type = sdl2.SDL_CONTROLLERBUTTONDOWN
|
||||
event.cbutton.button = button
|
||||
return event
|
||||
|
||||
|
||||
class TestKeyMapping:
|
||||
def test_up(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_UP)) == Action.UP
|
||||
|
||||
def test_down(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_DOWN)) == Action.DOWN
|
||||
|
||||
def test_confirm_return(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_RETURN)) == Action.CONFIRM
|
||||
|
||||
def test_confirm_space(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_SPACE)) == Action.CONFIRM
|
||||
|
||||
def test_back_escape(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_ESCAPE)) == Action.BACK
|
||||
|
||||
def test_quit(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_q)) == Action.QUIT
|
||||
|
||||
def test_hud_mode(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_h)) == Action.HUD_MODE
|
||||
|
||||
def test_page_up(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_PAGEUP)) == Action.PAGE_UP
|
||||
|
||||
def test_page_down(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_PAGEDOWN)) == Action.PAGE_DOWN
|
||||
|
||||
def test_unmapped_key(self):
|
||||
assert map_key(_make_key_event(sdl2.SDLK_z)) is None
|
||||
|
||||
def test_wrong_event_type(self):
|
||||
event = sdl2.SDL_Event()
|
||||
event.type = sdl2.SDL_MOUSEMOTION
|
||||
assert map_key(event) is None
|
||||
|
||||
|
||||
class TestButtonMapping:
|
||||
def test_dpad_up(self):
|
||||
assert map_key(_make_button_event(sdl2.SDL_CONTROLLER_BUTTON_DPAD_UP)) == Action.UP
|
||||
|
||||
def test_button_a(self):
|
||||
assert map_key(_make_button_event(sdl2.SDL_CONTROLLER_BUTTON_A)) == Action.CONFIRM
|
||||
|
||||
def test_button_b(self):
|
||||
assert map_key(_make_button_event(sdl2.SDL_CONTROLLER_BUTTON_B)) == Action.BACK
|
||||
|
||||
def test_start(self):
|
||||
assert map_key(_make_button_event(sdl2.SDL_CONTROLLER_BUTTON_START)) == Action.QUIT
|
||||
|
||||
def test_left_shoulder(self):
|
||||
assert map_key(_make_button_event(sdl2.SDL_CONTROLLER_BUTTON_LEFTSHOULDER)) == Action.PAGE_UP
|
||||
|
||||
def test_button_y(self):
|
||||
assert map_key(_make_button_event(sdl2.SDL_CONTROLLER_BUTTON_Y)) == Action.HUD_MODE
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Tests for DIDL-Lite parsing and domain model mapping."""
|
||||
|
||||
from r36s_dlna_browser.dlna.models import (
|
||||
ItemType,
|
||||
MediaItem,
|
||||
classify_upnp_class,
|
||||
parse_didl_item,
|
||||
)
|
||||
|
||||
|
||||
class TestClassifyUpnpClass:
|
||||
def test_container(self):
|
||||
assert classify_upnp_class("object.container.storageFolder") == ItemType.CONTAINER
|
||||
|
||||
def test_audio(self):
|
||||
assert classify_upnp_class("object.item.audioItem.musicTrack") == ItemType.AUDIO
|
||||
|
||||
def test_video(self):
|
||||
assert classify_upnp_class("object.item.videoItem") == ItemType.VIDEO
|
||||
|
||||
def test_image(self):
|
||||
assert classify_upnp_class("object.item.imageItem.photo") == ItemType.IMAGE
|
||||
|
||||
def test_unknown(self):
|
||||
assert classify_upnp_class("object.item.something") == ItemType.UNKNOWN
|
||||
|
||||
def test_empty(self):
|
||||
assert classify_upnp_class("") == ItemType.UNKNOWN
|
||||
|
||||
|
||||
class TestParseDIDLItem:
|
||||
def test_audio_item(self):
|
||||
didl = {
|
||||
"id": "42",
|
||||
"title": "My Song",
|
||||
"parent_id": "10",
|
||||
"upnp_class": "object.item.audioItem.musicTrack",
|
||||
"resources": [
|
||||
{
|
||||
"url": "http://server/song.mp3",
|
||||
"protocol_info": "http-get:*:audio/mpeg:*",
|
||||
"size": "5000000",
|
||||
"duration": "0:03:45",
|
||||
}
|
||||
],
|
||||
"album_art_uri": "http://server/art.jpg",
|
||||
}
|
||||
item = parse_didl_item(didl)
|
||||
assert item.object_id == "42"
|
||||
assert item.title == "My Song"
|
||||
assert item.item_type == ItemType.AUDIO
|
||||
assert item.resource_url == "http://server/song.mp3"
|
||||
assert item.mime_type == "http-get:*:audio/mpeg:*"
|
||||
assert item.size == 5000000
|
||||
assert item.duration == "0:03:45"
|
||||
assert item.album_art_url == "http://server/art.jpg"
|
||||
assert not item.is_container
|
||||
|
||||
def test_container_item(self):
|
||||
didl = {
|
||||
"id": "5",
|
||||
"title": "Music",
|
||||
"parent_id": "0",
|
||||
"upnp_class": "object.container.storageFolder",
|
||||
"child_count": "12",
|
||||
}
|
||||
item = parse_didl_item(didl)
|
||||
assert item.object_id == "5"
|
||||
assert item.title == "Music"
|
||||
assert item.item_type == ItemType.CONTAINER
|
||||
assert item.child_count == 12
|
||||
assert item.is_container
|
||||
|
||||
def test_missing_resources(self):
|
||||
didl = {"id": "99", "title": "NoRes", "upnp_class": "object.item.audioItem"}
|
||||
item = parse_didl_item(didl)
|
||||
assert item.resource_url == ""
|
||||
assert item.size == 0
|
||||
|
||||
def test_single_resource_dict(self):
|
||||
didl = {
|
||||
"id": "1",
|
||||
"title": "Track",
|
||||
"upnp_class": "object.item.audioItem",
|
||||
"resources": {"url": "http://srv/track.flac", "size": "100"},
|
||||
}
|
||||
item = parse_didl_item(didl)
|
||||
assert item.resource_url == "http://srv/track.flac"
|
||||
assert item.size == 100
|
||||
|
||||
def test_resource_string(self):
|
||||
didl = {
|
||||
"id": "2",
|
||||
"title": "Track2",
|
||||
"upnp_class": "object.item.videoItem",
|
||||
"resources": ["http://srv/vid.mp4"],
|
||||
}
|
||||
item = parse_didl_item(didl)
|
||||
assert item.resource_url == "http://srv/vid.mp4"
|
||||
|
||||
def test_alt_key_names(self):
|
||||
"""Handles alternative key formats (@id, @parentID, class, etc.)."""
|
||||
didl = {
|
||||
"@id": "7",
|
||||
"title": "Alt",
|
||||
"@parentID": "3",
|
||||
"class": "object.container",
|
||||
"@childCount": "5",
|
||||
}
|
||||
item = parse_didl_item(didl)
|
||||
assert item.object_id == "7"
|
||||
assert item.parent_id == "3"
|
||||
assert item.item_type == ItemType.CONTAINER
|
||||
assert item.child_count == 5
|
||||
|
||||
def test_empty_dict(self):
|
||||
item = parse_didl_item({})
|
||||
assert item.object_id == ""
|
||||
assert item.title == ""
|
||||
assert item.item_type == ItemType.UNKNOWN
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Tests for BrowserState navigation stack and pagination."""
|
||||
|
||||
from r36s_dlna_browser.dlna.browser_state import BrowserState
|
||||
from r36s_dlna_browser.dlna.models import MediaServer, MediaItem, ItemType
|
||||
|
||||
|
||||
def _server(name: str = "TestServer") -> MediaServer:
|
||||
return MediaServer(friendly_name=name, location=f"http://{name}:8080/desc.xml")
|
||||
|
||||
|
||||
def _items(n: int, container: bool = False) -> list[MediaItem]:
|
||||
t = ItemType.CONTAINER if container else ItemType.AUDIO
|
||||
return [
|
||||
MediaItem(
|
||||
object_id=str(i),
|
||||
title=f"Item {i}",
|
||||
item_type=t,
|
||||
resource_url="" if container else f"http://srv/file{i}.mp3",
|
||||
)
|
||||
for i in range(n)
|
||||
]
|
||||
|
||||
|
||||
class TestServerList:
|
||||
def test_empty_initial(self):
|
||||
s = BrowserState()
|
||||
assert s.servers == []
|
||||
assert s.selected_server is None
|
||||
|
||||
def test_set_servers(self):
|
||||
s = BrowserState()
|
||||
servers = [_server("A"), _server("B")]
|
||||
s.set_servers(servers)
|
||||
assert len(s.servers) == 2
|
||||
assert s.server_cursor == 0
|
||||
assert s.selected_server.friendly_name == "A"
|
||||
|
||||
def test_cursor_bounds(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server("A"), _server("B"), _server("C")])
|
||||
s.server_cursor = 2
|
||||
assert s.selected_server.friendly_name == "C"
|
||||
s.server_cursor = 5 # out of bounds
|
||||
assert s.selected_server is None
|
||||
|
||||
|
||||
class TestBrowseStack:
|
||||
def test_not_in_browse_initially(self):
|
||||
s = BrowserState()
|
||||
assert not s.in_browse_mode
|
||||
|
||||
def test_enter_server(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server()])
|
||||
obj_id = s.enter_server(s.selected_server)
|
||||
assert obj_id == "0"
|
||||
assert s.in_browse_mode
|
||||
assert s.current_level.object_id == "0"
|
||||
|
||||
def test_set_items_and_cursor(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server()])
|
||||
s.enter_server(s.selected_server)
|
||||
items = _items(10)
|
||||
s.set_items(items, "0")
|
||||
assert len(s.current_items) == 10
|
||||
assert s.cursor == 0
|
||||
s.cursor = 3
|
||||
assert s.cursor == 3
|
||||
assert s.selected_item().title == "Item 3"
|
||||
|
||||
def test_cursor_clamps(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server()])
|
||||
s.enter_server(s.selected_server)
|
||||
s.set_items(_items(5), "0")
|
||||
s.cursor = 100
|
||||
assert s.cursor == 4 # clamped to last
|
||||
s.cursor = -5
|
||||
assert s.cursor == 0
|
||||
|
||||
def test_push_and_pop(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server()])
|
||||
s.enter_server(s.selected_server)
|
||||
s.set_items(_items(3, container=True), "0")
|
||||
|
||||
# Drill into child
|
||||
s.set_items(_items(5), "1")
|
||||
assert len(s.current_items) == 5
|
||||
|
||||
# Go back
|
||||
assert s.go_back()
|
||||
assert len(s.current_items) == 3
|
||||
assert s.current_level.object_id == "0"
|
||||
|
||||
# Go back again exits browse
|
||||
assert s.go_back()
|
||||
assert not s.in_browse_mode
|
||||
|
||||
def test_go_back_at_root(self):
|
||||
s = BrowserState()
|
||||
assert not s.go_back()
|
||||
|
||||
def test_reset(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server()])
|
||||
s.enter_server(s.selected_server)
|
||||
s.set_items(_items(5), "0")
|
||||
s.error = "fail"
|
||||
s.playback_title = "song"
|
||||
s.playback_paused = True
|
||||
s.reset()
|
||||
assert not s.in_browse_mode
|
||||
assert s.error == ""
|
||||
assert s.playback_title == ""
|
||||
assert not s.playback_paused
|
||||
|
||||
|
||||
class TestScrollOffset:
|
||||
def test_scroll_offset_default(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server()])
|
||||
s.enter_server(s.selected_server)
|
||||
s.set_items(_items(3), "0")
|
||||
assert s.scroll_offset == 0
|
||||
|
||||
def test_scroll_offset_set(self):
|
||||
s = BrowserState()
|
||||
s.set_servers([_server()])
|
||||
s.enter_server(s.selected_server)
|
||||
s.set_items(_items(30), "0")
|
||||
s.scroll_offset = 10
|
||||
assert s.scroll_offset == 10
|
||||
@@ -0,0 +1,357 @@
|
||||
"""Tests for the SDL-rendered GStreamer backend without using a real pipeline."""
|
||||
|
||||
import ctypes
|
||||
from types import SimpleNamespace
|
||||
|
||||
import sdl2
|
||||
|
||||
from r36s_dlna_browser.player.gstreamer_backend import GStreamerBackend
|
||||
|
||||
|
||||
class FakeMessageType:
|
||||
ERROR = 1
|
||||
EOS = 2
|
||||
BUFFERING = 4
|
||||
STATE_CHANGED = 8
|
||||
|
||||
|
||||
class FakeFlowReturn:
|
||||
OK = 0
|
||||
|
||||
|
||||
class FakeState:
|
||||
NULL = 0
|
||||
READY = 1
|
||||
PAUSED = 2
|
||||
PLAYING = 3
|
||||
|
||||
|
||||
class FakeStateChangeReturn:
|
||||
FAILURE = -1
|
||||
SUCCESS = 0
|
||||
|
||||
|
||||
class FakeFormat:
|
||||
TIME = 1
|
||||
|
||||
|
||||
class FakeSeekFlags:
|
||||
FLUSH = 1
|
||||
KEY_UNIT = 2
|
||||
|
||||
|
||||
class FakeCaps:
|
||||
def __init__(self, width, height):
|
||||
self._width = width
|
||||
self._height = height
|
||||
|
||||
def get_size(self):
|
||||
return 1
|
||||
|
||||
def get_structure(self, _index):
|
||||
return FakeStructure(self._width, self._height)
|
||||
|
||||
|
||||
class FakeStructure:
|
||||
def __init__(self, width, height):
|
||||
self._width = width
|
||||
self._height = height
|
||||
|
||||
def has_field(self, name):
|
||||
return name in {"width", "height"}
|
||||
|
||||
def get_value(self, name):
|
||||
return self._width if name == "width" else self._height
|
||||
|
||||
|
||||
class FakeBuffer:
|
||||
def __init__(self, payload: bytes):
|
||||
self._payload = payload
|
||||
|
||||
def get_size(self):
|
||||
return len(self._payload)
|
||||
|
||||
def extract_dup(self, _offset, _size):
|
||||
return self._payload
|
||||
|
||||
|
||||
class FakeSample:
|
||||
def __init__(self, width=1280, height=720, payload: bytes | None = None):
|
||||
self._caps = FakeCaps(width, height)
|
||||
self._buffer = FakeBuffer(payload or (b"\x00" * (width * height * 4)))
|
||||
|
||||
def get_caps(self):
|
||||
return self._caps
|
||||
|
||||
def get_buffer(self):
|
||||
return self._buffer
|
||||
|
||||
|
||||
class FakeAppSink:
|
||||
def __init__(self):
|
||||
self.props = {}
|
||||
self.connected = []
|
||||
self.sample = None
|
||||
|
||||
def set_property(self, name, value):
|
||||
self.props[name] = value
|
||||
|
||||
def connect(self, signal_name, callback):
|
||||
self.connected.append((signal_name, callback))
|
||||
|
||||
def emit(self, signal_name):
|
||||
assert signal_name == "pull-sample"
|
||||
return self.sample
|
||||
|
||||
|
||||
class FakeBus:
|
||||
def __init__(self):
|
||||
self.message = None
|
||||
|
||||
def timed_pop_filtered(self, *_args):
|
||||
msg, self.message = self.message, None
|
||||
return msg
|
||||
|
||||
|
||||
class FakePipeline:
|
||||
def __init__(self):
|
||||
self.props = {}
|
||||
self.state = FakeState.NULL
|
||||
self.uri = None
|
||||
self.position = 12 * 1_000_000_000
|
||||
self.duration = 93 * 1_000_000_000
|
||||
self.bus = FakeBus()
|
||||
self.seek_calls = []
|
||||
|
||||
def set_state(self, state):
|
||||
self.state = state
|
||||
return FakeStateChangeReturn.SUCCESS
|
||||
|
||||
def set_property(self, name, value):
|
||||
self.props[name] = value
|
||||
if name == "uri":
|
||||
self.uri = value
|
||||
|
||||
def get_property(self, name):
|
||||
if name == "volume":
|
||||
return self.props.get("volume", 1.0)
|
||||
return self.props.get(name)
|
||||
|
||||
def get_bus(self):
|
||||
return self.bus
|
||||
|
||||
def query_position(self, _format):
|
||||
return True, self.position
|
||||
|
||||
def query_duration(self, _format):
|
||||
return True, self.duration
|
||||
|
||||
def seek_simple(self, _format, flags, target):
|
||||
self.seek_calls.append((flags, target))
|
||||
self.position = target
|
||||
return True
|
||||
|
||||
|
||||
class FakeMessage:
|
||||
def __init__(self, msg_type, src=None, buffering=0, state=None, error_text="boom", structure_name=None):
|
||||
self.type = msg_type
|
||||
self.src = src
|
||||
self._buffering = buffering
|
||||
self._state = state or (FakeState.NULL, FakeState.PLAYING, FakeState.NULL)
|
||||
self._error_text = error_text
|
||||
self._structure_name = structure_name
|
||||
|
||||
def parse_buffering(self):
|
||||
return self._buffering
|
||||
|
||||
def parse_state_changed(self):
|
||||
return self._state
|
||||
|
||||
def parse_error(self):
|
||||
return SimpleNamespace(message=self._error_text), None
|
||||
|
||||
def get_structure(self):
|
||||
if not self._structure_name:
|
||||
return None
|
||||
return SimpleNamespace(get_name=lambda: self._structure_name)
|
||||
|
||||
|
||||
class FakeGst:
|
||||
State = FakeState
|
||||
StateChangeReturn = FakeStateChangeReturn
|
||||
Format = FakeFormat
|
||||
SeekFlags = FakeSeekFlags
|
||||
MessageType = FakeMessageType
|
||||
FlowReturn = FakeFlowReturn
|
||||
SECOND = 1_000_000_000
|
||||
MSECOND = 1_000_000
|
||||
Caps = SimpleNamespace(from_string=lambda value: value)
|
||||
|
||||
|
||||
class FakeVideoInfoValue:
|
||||
def __init__(self, width, height, stride):
|
||||
self.width = width
|
||||
self.height = height
|
||||
self.stride = [stride]
|
||||
|
||||
|
||||
class FakeVideoInfo:
|
||||
@staticmethod
|
||||
def new_from_caps(caps):
|
||||
structure = caps.get_structure(0)
|
||||
width = structure.get_value("width")
|
||||
height = structure.get_value("height")
|
||||
return FakeVideoInfoValue(width, height, width * 4)
|
||||
|
||||
|
||||
class FakeGstVideo:
|
||||
VideoInfo = FakeVideoInfo
|
||||
|
||||
|
||||
class TestGStreamerBackend:
|
||||
def _make_backend(self):
|
||||
pipeline = FakePipeline()
|
||||
sink = FakeAppSink()
|
||||
backend = GStreamerBackend(
|
||||
gst_module=FakeGst,
|
||||
gst_video_module=FakeGstVideo,
|
||||
appsink_factory=lambda: sink,
|
||||
playbin_factory=lambda: pipeline,
|
||||
subsystem="x11",
|
||||
)
|
||||
backend._start_bus_thread = lambda: None
|
||||
backend._stop_bus_thread = lambda: None
|
||||
return backend, pipeline, sink
|
||||
|
||||
def test_not_playing_initially(self):
|
||||
backend, _pipeline, _sink = self._make_backend()
|
||||
assert not backend.is_playing()
|
||||
|
||||
def test_play_sets_uri_and_configures_appsink(self):
|
||||
backend, pipeline, sink = self._make_backend()
|
||||
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
assert backend.is_playing()
|
||||
assert pipeline.uri == "http://example.com/video.mp4"
|
||||
assert pipeline.props["video-sink"] is sink
|
||||
assert sink.props["emit-signals"] is True
|
||||
assert sink.props["caps"] == "video/x-raw,format=BGRA"
|
||||
|
||||
def test_new_sample_marks_frame_dirty_and_updates_resolution(self):
|
||||
events = []
|
||||
backend, _pipeline, sink = self._make_backend()
|
||||
backend.set_event_callback(lambda event, *args: events.append((event, *args)))
|
||||
sink.sample = FakeSample(width=640, height=360)
|
||||
|
||||
assert backend._on_new_sample(sink) == FakeFlowReturn.OK
|
||||
|
||||
assert backend.has_new_frame() is True
|
||||
assert ("resolution", "640x360") in events
|
||||
|
||||
def test_toggle_pause_switches_pipeline_state(self):
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
assert backend.toggle_pause() is True
|
||||
assert pipeline.state == FakeState.PAUSED
|
||||
assert backend.toggle_pause() is False
|
||||
assert pipeline.state == FakeState.PLAYING
|
||||
|
||||
def test_seek_uses_relative_position(self):
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
backend.seek(10)
|
||||
|
||||
assert pipeline.seek_calls == [(FakeSeekFlags.FLUSH | FakeSeekFlags.KEY_UNIT, 22 * 1_000_000_000)]
|
||||
|
||||
def test_change_volume_clamps_to_supported_range(self):
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
assert backend.change_volume(15) == 115
|
||||
assert pipeline.props["volume"] == 1.15
|
||||
assert backend.change_volume(50) == 130
|
||||
assert pipeline.props["volume"] == 1.3
|
||||
|
||||
def test_eos_notifies_stopped(self):
|
||||
events = []
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.set_event_callback(lambda event, *args: events.append((event, *args)))
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
backend._handle_bus_message(FakeMessage(FakeMessageType.EOS, src=pipeline))
|
||||
|
||||
assert ("stopped",) in events
|
||||
assert not backend.is_playing()
|
||||
assert pipeline.state == FakeState.NULL
|
||||
|
||||
def test_buffering_and_metrics_feed_hud(self):
|
||||
events = []
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.set_event_callback(lambda event, *args: events.append((event, *args)))
|
||||
backend.play("http://example.com/video.mp4")
|
||||
backend._resolution = "1280x720"
|
||||
|
||||
backend._handle_bus_message(FakeMessage(FakeMessageType.BUFFERING, buffering=76))
|
||||
backend._emit_playback_metrics()
|
||||
|
||||
assert ("buffering", 76) in events
|
||||
assert ("position", 12.0) in events
|
||||
assert ("duration", 93.0) in events
|
||||
assert ("volume", 100) in events
|
||||
assert ("resolution", "1280x720") in events
|
||||
|
||||
def test_state_changed_updates_pause_event(self):
|
||||
events = []
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.set_event_callback(lambda event, *args: events.append((event, *args)))
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
backend._handle_bus_message(
|
||||
FakeMessage(
|
||||
FakeMessageType.STATE_CHANGED,
|
||||
src=pipeline,
|
||||
state=(FakeState.PLAYING, FakeState.PAUSED, FakeState.NULL),
|
||||
)
|
||||
)
|
||||
|
||||
assert ("paused", True) in events
|
||||
|
||||
def test_error_message_propagates(self):
|
||||
events = []
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.set_event_callback(lambda event, *args: events.append((event, *args)))
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
backend._handle_bus_message(FakeMessage(FakeMessageType.ERROR, src=pipeline, error_text="decoder fail"))
|
||||
|
||||
assert ("error", "decoder fail") in events
|
||||
assert not backend.is_playing()
|
||||
|
||||
def test_stop_sets_pipeline_to_null(self):
|
||||
backend, pipeline, _sink = self._make_backend()
|
||||
backend.play("http://example.com/video.mp4")
|
||||
|
||||
backend.stop()
|
||||
|
||||
assert pipeline.state == FakeState.NULL
|
||||
assert not backend.is_playing()
|
||||
|
||||
def test_render_uploads_latest_frame_and_clears_dirty_flag(self, monkeypatch):
|
||||
backend, _pipeline, _sink = self._make_backend()
|
||||
backend._latest_frame = SimpleNamespace(width=320, height=180, pitch=1280, pixels=b"\x00" * (320 * 180 * 4))
|
||||
backend._frame_dirty = True
|
||||
|
||||
calls = []
|
||||
|
||||
monkeypatch.setattr(sdl2, "SDL_CreateTexture", lambda *_args: object())
|
||||
monkeypatch.setattr(sdl2, "SDL_UpdateTexture", lambda *_args: 0)
|
||||
monkeypatch.setattr(sdl2, "SDL_RenderCopy", lambda _renderer, _texture, _src, dst: calls.append((dst.x, dst.y, dst.w, dst.h)))
|
||||
|
||||
backend.set_viewport(640, 480, 48, 80, 12, 12)
|
||||
|
||||
assert backend.render(object()) is True
|
||||
assert backend.has_new_frame() is False
|
||||
assert calls == [(12, 51, 616, 346)]
|
||||
@@ -0,0 +1,40 @@
|
||||
"""Tests for runtime environment setup helpers."""
|
||||
|
||||
import os
|
||||
|
||||
from r36s_dlna_browser.platform import runtime
|
||||
|
||||
|
||||
def test_wayland_does_not_override_video_driver(monkeypatch):
|
||||
monkeypatch.setenv("WAYLAND_DISPLAY", "wayland-0")
|
||||
monkeypatch.setenv("DISPLAY", ":0")
|
||||
monkeypatch.delenv("SDL_VIDEODRIVER", raising=False)
|
||||
monkeypatch.setattr(runtime, "is_r36s", lambda: False)
|
||||
|
||||
runtime.sdl_env_setup()
|
||||
|
||||
assert "SDL_VIDEODRIVER" not in os.environ
|
||||
|
||||
|
||||
def test_respects_existing_sdl_videodriver(monkeypatch):
|
||||
monkeypatch.setenv("WAYLAND_DISPLAY", "wayland-0")
|
||||
monkeypatch.setenv("DISPLAY", ":0")
|
||||
monkeypatch.setenv("SDL_VIDEODRIVER", "wayland")
|
||||
monkeypatch.setattr(runtime, "is_r36s", lambda: False)
|
||||
|
||||
runtime.sdl_env_setup()
|
||||
|
||||
assert os.environ["SDL_VIDEODRIVER"] == "wayland"
|
||||
|
||||
|
||||
def test_r36s_sets_alsa_audio_driver(monkeypatch):
|
||||
monkeypatch.delenv("SDL_AUDIODRIVER", raising=False)
|
||||
monkeypatch.delenv("SDL_VIDEODRIVER", raising=False)
|
||||
monkeypatch.delenv("DISPLAY", raising=False)
|
||||
monkeypatch.delenv("WAYLAND_DISPLAY", raising=False)
|
||||
monkeypatch.setattr(runtime, "is_r36s", lambda: True)
|
||||
|
||||
runtime.sdl_env_setup()
|
||||
|
||||
assert os.environ["SDL_AUDIODRIVER"] == "alsa"
|
||||
assert os.environ["SDL_VIDEODRIVER"] == "kmsdrm"
|
||||
@@ -0,0 +1,122 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from queue import Queue
|
||||
|
||||
from r36s_dlna_browser.dlna.browser_state import BrowserState
|
||||
from r36s_dlna_browser.platform.controls import Action
|
||||
from r36s_dlna_browser.ui import theme
|
||||
from r36s_dlna_browser.ui import sdl_app as sdl_app_module
|
||||
from r36s_dlna_browser.ui.sdl_app import SDLApp, Screen
|
||||
|
||||
|
||||
class DummyPlayer:
|
||||
def __init__(self) -> None:
|
||||
self._new_frame = False
|
||||
|
||||
def attach_window(self, _window) -> None:
|
||||
pass
|
||||
|
||||
def set_viewport(self, *_args) -> None:
|
||||
pass
|
||||
|
||||
def has_new_frame(self) -> bool:
|
||||
return self._new_frame
|
||||
|
||||
def render(self, _renderer) -> bool:
|
||||
self._new_frame = False
|
||||
return True
|
||||
|
||||
|
||||
def _make_app() -> SDLApp:
|
||||
return SDLApp(Queue(), Queue(), BrowserState(), DummyPlayer())
|
||||
|
||||
|
||||
def test_non_playback_draws_only_when_dirty() -> None:
|
||||
app = _make_app()
|
||||
app._screen = Screen.SERVERS
|
||||
app._needs_redraw = False
|
||||
|
||||
assert app._should_draw() is False
|
||||
|
||||
app._mark_dirty()
|
||||
|
||||
assert app._should_draw() is True
|
||||
|
||||
|
||||
def test_playback_snapshot_change_waits_for_refresh_interval(monkeypatch) -> None:
|
||||
app = _make_app()
|
||||
app._screen = Screen.PLAYBACK
|
||||
app._needs_redraw = False
|
||||
app._last_playback_snapshot = app._playback_snapshot()
|
||||
app._last_playback_draw_at = 10.0
|
||||
app._state.playback_position = 1.0
|
||||
|
||||
monkeypatch.setattr(sdl_app_module.time, "monotonic", lambda: 10.1)
|
||||
|
||||
assert app._should_draw() is False
|
||||
|
||||
|
||||
def test_playback_snapshot_change_redraws_after_refresh_interval(monkeypatch) -> None:
|
||||
app = _make_app()
|
||||
app._screen = Screen.PLAYBACK
|
||||
app._needs_redraw = False
|
||||
app._last_playback_snapshot = app._playback_snapshot()
|
||||
app._last_playback_draw_at = 10.0
|
||||
app._state.playback_position = 1.0
|
||||
|
||||
monkeypatch.setattr(sdl_app_module.time, "monotonic", lambda: 10.3)
|
||||
|
||||
assert app._should_draw() is True
|
||||
|
||||
|
||||
def test_playback_snapshot_without_changes_does_not_redraw(monkeypatch) -> None:
|
||||
app = _make_app()
|
||||
app._screen = Screen.PLAYBACK
|
||||
app._needs_redraw = False
|
||||
app._last_playback_snapshot = app._playback_snapshot()
|
||||
app._last_playback_draw_at = 10.0
|
||||
|
||||
monkeypatch.setattr(sdl_app_module.time, "monotonic", lambda: 10.5)
|
||||
|
||||
assert app._should_draw() is False
|
||||
|
||||
|
||||
def test_playback_draws_immediately_for_new_video_frame() -> None:
|
||||
app = _make_app()
|
||||
app._screen = Screen.PLAYBACK
|
||||
app._needs_redraw = False
|
||||
app._player._new_frame = True
|
||||
|
||||
assert app._should_draw() is True
|
||||
|
||||
|
||||
def test_hud_mode_cycles_auto_fixed_hidden(monkeypatch) -> None:
|
||||
app = _make_app()
|
||||
app._screen = Screen.PLAYBACK
|
||||
monkeypatch.setattr(sdl_app_module.time, "monotonic", lambda: 12.0)
|
||||
|
||||
app._handle_action(Action.HUD_MODE)
|
||||
assert app._state.playback_hud_mode == theme.PLAYBACK_HUD_PINNED
|
||||
assert app._state.playback_hud_visible is True
|
||||
|
||||
app._handle_action(Action.HUD_MODE)
|
||||
assert app._state.playback_hud_mode == theme.PLAYBACK_HUD_HIDDEN
|
||||
assert app._state.playback_hud_visible is False
|
||||
|
||||
app._handle_action(Action.HUD_MODE)
|
||||
assert app._state.playback_hud_mode == theme.PLAYBACK_HUD_AUTO
|
||||
assert app._state.playback_hud_visible is True
|
||||
|
||||
|
||||
def test_paused_playback_keeps_hud_visible_in_auto_mode(monkeypatch) -> None:
|
||||
app = _make_app()
|
||||
app._screen = Screen.PLAYBACK
|
||||
app._state.playback_hud_mode = theme.PLAYBACK_HUD_AUTO
|
||||
app._state.playback_hud_visible = False
|
||||
app._state.playback_paused = True
|
||||
|
||||
monkeypatch.setattr(sdl_app_module.time, "monotonic", lambda: 99.0)
|
||||
|
||||
app._refresh_playback_hud_visibility()
|
||||
|
||||
assert app._state.playback_hud_visible is True
|
||||
Reference in New Issue
Block a user