Repository: streamlink/streamlink
Pull request #3967 (2021-08-31 09:01:07)

plugin.api.validate: switch to lxml.etree
- replace xml.etree from standard library with lxml.etree in
  plugin.api.validate, utils.parse_xml, plugins and tests
- add validate.xml_xpath() and validate.xml_xpath_string() methods
- fix validate.get() and return attribute value for given attribute key
- refactor validate.xml_element() validation method
- fix and add tests

----

#3952 

bastimeyer: This is not part of this PR, but I think it's time to add `parse_{xml,html,json}` utility methods to the validation module. It's a bit silly that almost every plugin has to import this on its own and define a validate.transform() call for it.

Patch:
diff --git a/src/streamlink/plugin/api/validate.py b/src/streamlink/plugin/api/validate.py
index 932affe7..21ac3824 100644
--- a/src/streamlink/plugin/api/validate.py
+++ b/src/streamlink/plugin/api/validate.py
@@ -20,14 +20,15 @@ from copy import copy as copy_obj
 from functools import singledispatch
 from typing import Any, Tuple, Union
 from urllib.parse import urlparse
-from xml.etree import ElementTree as ET
+
+from lxml.etree import Element, iselement
 
 from streamlink.exceptions import PluginError
 
 __all__ = [
     "any", "all", "filter", "get", "getattr", "hasattr", "length", "optional",
     "transform", "text", "union", "union_get", "url", "startswith", "endswith", "contains",
-    "xml_element", "xml_find", "xml_findall", "xml_findtext",
+    "xml_element", "xml_find", "xml_findall", "xml_findtext", "xml_xpath", "xml_xpath_string",
     "validate", "Schema", "SchemaContainer"
 ]
 
@@ -164,8 +165,8 @@ def get(item: Union[Any, Tuple[Any]], default: Any = None, strict: bool = False)
         idx = 0
         try:
             for key in item:
-                if ET.iselement(value):
-                    value = value.attrib
+                if iselement(value):
+                    value = value.attrib[key]
                 # Use .group() if this is a regex match object
                 elif _is_re_match(value):
                     value = value.group(key)
@@ -275,12 +276,12 @@ def url(**attributes):
 def xml_find(xpath):
     """Find a XML element via xpath."""
     def xpath_find(value):
-        validate(ET.iselement, value)
+        validate(iselement, value)
         value = value.find(xpath)
         if value is None:
-            raise ValueError("XPath '{0}' did not return an element".format(xpath))
+            raise ValueError(f"XPath '{xpath}' did not return an element")
 
-        return validate(ET.iselement, value)
+        return validate(iselement, value)
 
     return transform(xpath_find)
 
@@ -288,7 +289,7 @@ def xml_find(xpath):
 def xml_findall(xpath):
     """Find a list of XML elements via xpath."""
     def xpath_findall(value):
-        validate(ET.iselement, value)
+        validate(iselement, value)
         return value.findall(xpath)
 
     return transform(xpath_findall)
@@ -302,6 +303,18 @@ def xml_findtext(xpath):
     )
 
 
+def xml_xpath(xpath):
+    def transform_xpath(value):
+        validate(iselement, value)
+        return value.xpath(xpath) or None
+
+    return transform(transform_xpath)
+
+
+def xml_xpath_string(xpath):
+    return xml_xpath(f"string({xpath})")
+
+
 @singledispatch
 def validate(schema, value):
     if callable(schema):
@@ -393,27 +406,31 @@ def validate_type(schema, value):
 
 @validate.register(xml_element)
 def validate_xml_element(schema, value):
-    validate(ET.iselement, value)
-    new = ET.Element(value.tag, attrib=value.attrib)
+    validate(iselement, value)
+    _tag = value.tag
+    _attrib = value.attrib
+    _text = value.text
 
     if schema.attrib is not None:
         try:
-            new.attrib = validate(schema.attrib, value.attrib)
+            _attrib = validate(schema.attrib, dict(value.attrib))
         except ValueError as err:
-            raise ValueError("Unable to validate XML attributes: {0}".format(err))
+            raise ValueError(f"Unable to validate XML attributes: {err}")
 
     if schema.tag is not None:
         try:
-            new.tag = validate(schema.tag, value.tag)
+            _tag = validate(schema.tag, value.tag)
         except ValueError as err:
-            raise ValueError("Unable to validate XML tag: {0}".format(err))
+            raise ValueError(f"Unable to validate XML tag: {err}")
 
     if schema.text is not None:
         try:
-            new.text = validate(schema.text, value.text)
+            _text = validate(schema.text, value.text)
         except ValueError as err:
-            raise ValueError("Unable to validate XML text: {0}".format(err))
+            raise ValueError(f"Unable to validate XML text: {err}")
 
+    new = Element(_tag, _attrib)
+    new.text = _text
     for child in value:
         new.append(child)
 
diff --git a/src/streamlink/plugins/radiko.py b/src/streamlink/plugins/radiko.py
index 0096fcd1..3997c08e 100644
--- a/src/streamlink/plugins/radiko.py
+++ b/src/streamlink/plugins/radiko.py
@@ -3,9 +3,10 @@ import datetime
 import hashlib
 import random
 import re
-import xml.etree.ElementTree as ET
 from urllib.parse import urlencode
 
+from lxml.etree import XML
+
 from streamlink.plugin import Plugin, pluginmatcher
 from streamlink.stream import HLSStream
 
@@ -96,7 +97,7 @@ class Radiko(Plugin):
             date = today.strftime('%Y%m%d')
         api = 'http://radiko.jp/v3/program/station/date/{}/{}.xml'.format(date, station_id)
         r = self.session.http.get(api)
-        tree = ET.XML(r.content)
+        tree = XML(r.content)
         for x in tree[2][0][1].findall('prog'):
             if x.attrib['ft'] == start_at:
                 return x.attrib['to']
diff --git a/src/streamlink/utils/__init__.py b/src/streamlink/utils/__init__.py
index 9aed6362..f2ff849f 100644
--- a/src/streamlink/utils/__init__.py
+++ b/src/streamlink/utils/__init__.py
@@ -1,6 +1,5 @@
 import json
 import re
-import xml.etree.ElementTree as ET
 import zlib
 from collections import OrderedDict
 from importlib.machinery import FileFinder, SOURCE_SUFFIXES, SourceFileLoader
@@ -8,6 +7,8 @@ from importlib.util import module_from_spec
 from typing import Dict, Generic, Optional, TypeVar
 from urllib.parse import parse_qsl, urljoin, urlparse
 
+from lxml.etree import XML
+
 from streamlink.exceptions import PluginError
 from streamlink.utils.lazy_formatter import LazyFormatter
 from streamlink.utils.named_pipe import NamedPipe
@@ -91,22 +92,22 @@ def parse_xml(data, name="XML", ignore_ns=False, exception=PluginError, schema=N
      - Wraps errors in custom exception with a snippet of the data in the message
     """
     if isinstance(data, str):
-        data = bytearray(data, "utf8")
+        data = bytes(data, "utf8")
 
     if ignore_ns:
         data = re.sub(br"[\t ]xmlns=\"(.+?)\"", b"", data)
 
     if invalid_char_entities:
-        data = re.sub(br'&(?!(?:#(?:[0-9]+|[Xx][0-9A-Fa-f]+)|[A-Za-z0-9]+);)', b'&amp;', data)
+        data = re.sub(br"&(?!(?:#(?:[0-9]+|[Xx][0-9A-Fa-f]+)|[A-Za-z0-9]+);)", b"&amp;", data)
 
     try:
-        tree = ET.fromstring(data)
+        tree = XML(data)
     except Exception as err:
         snippet = repr(data)
         if len(snippet) > 35:
             snippet = snippet[:35] + " ..."
 
-        raise exception("Unable to parse {0}: {1} ({2})".format(name, err, snippet))
+        raise exception(f"Unable to parse {name}: {err} ({snippet})")
 
     if schema:
         tree = schema.validate(tree, name=name, exception=exception)
diff --git a/src/streamlink_cli/argparser.py b/src/streamlink_cli/argparser.py
index ac3c6bb4..cf95a7ea 100644
--- a/src/streamlink_cli/argparser.py
+++ b/src/streamlink_cli/argparser.py
@@ -588,6 +588,9 @@ def build_parser():
         Write stream data to FILENAME instead of playing it.
 
         You will be prompted if the file already exists.
+
+        The formatting variables available for the --title option may be used.
+        Unsupported characters in substituted variables will be replaced with an underscore.
         """
     )
     output.add_argument(
@@ -619,6 +622,9 @@ def build_parser():
         Open the stream in the player, while at the same time writing it to FILENAME.
 
         You will be prompted if the file already exists.
+
+        The formatting variables available for the --title option may be used.
+        Unsupported characters in substituted variables will be replaced with an underscore.
         """
     )
     output.add_argument(
@@ -628,6 +634,28 @@ def build_parser():
         Write stream data to stdout, while at the same time writing it to FILENAME.
 
         You will be prompted if the file already exists.
+
+        The formatting variables available for the --title option may be used.
+        Unsupported characters in substituted variables will be replaced with an underscore.
+        """
+    )
+    output.add_argument(
+        "--fs-safe-rules",
+        choices=["POSIX", "Windows"],
+        type=str,
+        help="""
+        The rules used to make formatting variables filesystem-safe are chosen
+        automatically according to the type of system in use. This overrides
+        the automatic detection.
+
+        Intended for use when Streamlink is running on a UNIX-like OS but writing
+        to Windows filesystems such as NTFS; USB devices using VFAT or exFAT; CIFS
+        shares that are enforcing Windows filename limitations, etc.
+
+        These characters are replaced with an underscore for the rules in use:
+
+          POSIX  : \\x00-\\x1F /
+          Windows: \\x00-\\x1F \\x7F " * / : < > ? \\ |
         """
     )
 
diff --git a/src/streamlink_cli/main.py b/src/streamlink_cli/main.py
index 27d1b477..a65715fa 100644
--- a/src/streamlink_cli/main.py
+++ b/src/streamlink_cli/main.py
@@ -26,13 +26,13 @@ from streamlink.cache import Cache
 from streamlink.exceptions import FatalPluginError
 from streamlink.plugin import PluginOptions
 from streamlink.stream import StreamProcess
-from streamlink.utils import LazyFormatter, NamedPipe
+from streamlink.utils import NamedPipe
 from streamlink_cli.argparser import build_parser
 from streamlink_cli.compat import DeprecatedPath, is_win32, stdout
 from streamlink_cli.console import ConsoleOutput, ConsoleUserInputRequester
 from streamlink_cli.constants import CONFIG_FILES, DEFAULT_STREAM_METADATA, LOG_DIR, PLUGIN_DIRS, STREAM_SYNONYMS
 from streamlink_cli.output import FileOutput, PlayerOutput
-from streamlink_cli.utils import HTTPServer, ignored, progress, stream_to_url
+from streamlink_cli.utils import Formatter, HTTPServer, ignored, progress, stream_to_url
 
 ACCEPTABLE_ERRNO = (errno.EPIPE, errno.EINVAL, errno.ECONNRESET)
 try:
@@ -54,19 +54,17 @@ def check_file_output(filename, force):
 
     if os.path.isfile(filename) and not force:
         if sys.stdin.isatty():
-            answer = console.ask("File {0} already exists! Overwrite it? [y/N] ",
-                                 filename)
-
+            answer = console.ask(f"File {filename} already exists! Overwrite it? [y/N] ")
             if answer.lower() != "y":
                 sys.exit()
         else:
-            log.error("File {0} already exists, use --force to overwrite it.".format(filename))
+            log.error(f"File {filename} already exists, use --force to overwrite it.")
             sys.exit()
 
     return FileOutput(filename)
 
 
-def create_output(plugin):
+def create_output(formatter: Formatter):
     """Decides where to write the stream.
 
     Depending on arguments it can be one of these:
@@ -84,11 +82,11 @@ def create_output(plugin):
         if args.output == "-":
             out = FileOutput(fd=stdout)
         else:
-            out = check_file_output(args.output, args.force)
+            out = check_file_output(formatter.filename(args.output, args.fs_safe_rules), args.force)
     elif args.stdout:
         out = FileOutput(fd=stdout)
     elif args.record_and_pipe:
-        record = check_file_output(args.record_and_pipe, args.force)
+        record = check_file_output(formatter.filename(args.record_and_pipe, args.fs_safe_rules), args.force)
         out = FileOutput(fd=stdout, record=record)
     else:
         http = namedpipe = record = None
@@ -106,18 +104,21 @@ def create_output(plugin):
         elif args.player_http:
             http = create_http_server()
 
-        title = create_title(plugin)
-
         if args.record:
-            record = check_file_output(args.record, args.force)
+            record = check_file_output(formatter.filename(args.record, args.fs_safe_rules), args.force)
 
         log.info("Starting player: {0}".format(args.player))
 
-        out = PlayerOutput(args.player, args=args.player_args,
-                           quiet=not args.verbose_player,
-                           kill=not args.player_no_close,
-                           namedpipe=namedpipe, http=http,
-                           record=record, title=title)
+        out = PlayerOutput(
+            args.player,
+            args=args.player_args,
+            quiet=not args.verbose_player,
+            kill=not args.player_no_close,
+            namedpipe=namedpipe,
+            http=http,
+            record=record,
+            title=formatter.title(args.title, defaults=DEFAULT_STREAM_METADATA) if args.title else args.url
+        )
 
     return out
 
@@ -138,21 +139,6 @@ def create_http_server(*_args, **_kwargs):
     return http
 
 
-def create_title(plugin=None):
-    if args.title and plugin:
-        title = LazyFormatter.format(
-            args.title,
-            title=lambda: plugin.get_title() or DEFAULT_STREAM_METADATA["title"],
-            author=lambda: plugin.get_author() or DEFAULT_STREAM_METADATA["author"],
-            category=lambda: plugin.get_category() or DEFAULT_STREAM_METADATA["category"],
-            game=lambda: plugin.get_category() or DEFAULT_STREAM_METADATA["game"],
-            url=plugin.url
-        )
-    else:
-        title = args.url
-    return title
-
-
 def iter_http_requests(server, player):
     """Repeatedly accept HTTP connections on a server.
 
@@ -167,7 +153,7 @@ def iter_http_requests(server, player):
             continue
 
 
-def output_stream_http(plugin, initial_streams, external=False, port=0):
+def output_stream_http(plugin, initial_streams, formatter: Formatter, external=False, port=0):
     """Continuously output the stream over HTTP."""
     global output
 
@@ -177,12 +163,14 @@ def output_stream_http(plugin, initial_streams, external=False, port=0):
                          "installed. You must specify the path to a player "
                          "executable with --player.")
 
-        title = create_title(plugin)
         server = create_http_server()
-        player = output = PlayerOutput(args.player, args=args.player_args,
-                                       filename=server.url,
-                                       quiet=not args.verbose_player,
-                                       title=title)
+        player = output = PlayerOutput(
+            args.player,
+            args=args.player_args,
+            filename=server.url,
+            quiet=not args.verbose_player,
+            title=formatter.title(args.title, defaults=DEFAULT_STREAM_METADATA) if args.title else args.url
+        )
 
         try:
             log.info("Starting player: {0}".format(args.player))
@@ -231,7 +219,7 @@ def output_stream_http(plugin, initial_streams, external=False, port=0):
 
         if stream_fd and prebuffer:
             log.debug("Writing stream to player")
-            read_stream(stream_fd, server, prebuffer)
+            read_stream(stream_fd, server, prebuffer, formatter)
 
         server.close(True)
 
@@ -239,16 +227,19 @@ def output_stream_http(plugin, initial_streams, external=False, port=0):
     server.close()
 
 
-def output_stream_passthrough(plugin, stream):
+def output_stream_passthrough(stream, formatter: Formatter):
     """Prepares a filename to be passed to the player."""
     global output
 
-    title = create_title(plugin)
-    filename = '"{0}"'.format(stream_to_url(stream))
-    output = PlayerOutput(args.player, args=args.player_args,
-                          filename=filename, call=True,
-                          quiet=not args.verbose_player,
-                          title=title)
+    filename = f'"{stream_to_url(stream)}"'
+    output = PlayerOutput(
+        args.player,
+        args=args.player_args,
+        filename=filename,
+        call=True,
+        quiet=not args.verbose_player,
+        title=formatter.title(args.title, defaults=DEFAULT_STREAM_METADATA) if args.title else args.url
+    )
 
     try:
         log.info("Starting player: {0}".format(args.player))
@@ -291,7 +282,7 @@ def open_stream(stream):
     return stream_fd, prebuffer
 
 
-def output_stream(plugin, stream):
+def output_stream(stream, formatter: Formatter):
     """Open stream, create output and finally write the stream to output."""
     global output
 
@@ -308,26 +299,24 @@ def output_stream(plugin, stream):
     if not success_open:
         console.exit("Could not open stream {0}, tried {1} times, exiting", stream, args.retry_open)
 
-    output = create_output(plugin)
+    output = create_output(formatter)
 
     try:
         output.open()
     except OSError as err:
         if isinstance(output, PlayerOutput):
-            console.exit("Failed to start player: {0} ({1})",
-                         args.player, err)
+            console.exit(f"Failed to start player: {args.player} ({err})")
         else:
-            console.exit("Failed to open output: {0} ({1})",
-                         args.output, err)
+            console.exit(f"Failed to open output: {output.filename} ({err})")
 
     with closing(output):
         log.debug("Writing stream to output")
-        read_stream(stream_fd, output, prebuffer)
+        read_stream(stream_fd, output, prebuffer, formatter)
 
     return True
 
 
-def read_stream(stream, output, prebuffer, chunk_size=8192):
+def read_stream(stream, output, prebuffer, formatter: Formatter, chunk_size=8192):
     """Reads data from stream and then writes it to the output."""
     is_player = isinstance(output, PlayerOutput)
     is_http = isinstance(output, HTTPServer)
@@ -349,11 +338,15 @@ def read_stream(stream, output, prebuffer, chunk_size=8192):
         iter(partial(stream.read, chunk_size), b"")
     )
     if show_progress:
-        stream_iterator = progress(stream_iterator,
-                                   prefix=os.path.basename(args.output))
+        stream_iterator = progress(
+            stream_iterator,
+            prefix=os.path.basename(output.filename)
+        )
     elif show_record_progress:
-        stream_iterator = progress(stream_iterator,
-                                   prefix=os.path.basename(args.record))
+        stream_iterator = progress(
+            stream_iterator,
+            prefix=os.path.basename(output.record.filename)
+        )
 
     try:
         for data in stream_iterator:
@@ -430,24 +423,29 @@ def handle_stream(plugin, streams, stream_name):
                                   sorted(streams.keys())))
         file_output = args.output or args.stdout
 
+        formatter = Formatter({
+            "url": lambda: args.url,
+            "author": lambda: plugin.get_author(),
+            "category": lambda: plugin.get_category(),
+            "game": lambda: plugin.get_category(),
+            "title": lambda: plugin.get_title(),
+        })
+
         for stream_name in [stream_name] + alt_streams:
             stream = streams[stream_name]
             stream_type = type(stream).shortname()
 
             if stream_type in args.player_passthrough and not file_output:
-                log.info("Opening stream: {0} ({1})".format(stream_name,
-                                                            stream_type))
-                success = output_stream_passthrough(plugin, stream)
+                log.info(f"Opening stream: {stream_name} ({stream_type})")
+                success = output_stream_passthrough(stream, formatter)
             elif args.player_external_http:
-                return output_stream_http(plugin, streams, external=True,
+                return output_stream_http(plugin, streams, formatter, external=True,
                                           port=args.player_external_http_port)
             elif args.player_continuous_http and not file_output:
-                return output_stream_http(plugin, streams)
+                return output_stream_http(plugin, streams, formatter)
             else:
-                log.info("Opening stream: {0} ({1})".format(stream_name,
-                                                            stream_type))
-
-                success = output_stream(plugin, stream)
+                log.info(f"Opening stream: {stream_name} ({stream_type})")
+                success = output_stream(stream, formatter)
 
             if success:
                 break
diff --git a/src/streamlink_cli/utils/__init__.py b/src/streamlink_cli/utils/__init__.py
index a000a9ef..bacfcbcf 100644
--- a/src/streamlink_cli/utils/__init__.py
+++ b/src/streamlink_cli/utils/__init__.py
@@ -1,13 +1,14 @@
 import json
 from contextlib import contextmanager
 
+from streamlink_cli.utils.formatter import Formatter
 from streamlink_cli.utils.http_server import HTTPServer
 from streamlink_cli.utils.player import find_default_player
 from streamlink_cli.utils.progress import progress
 from streamlink_cli.utils.stream import stream_to_url
 
 __all__ = [
-    "HTTPServer", "JSONEncoder",
+    "Formatter", "HTTPServer", "JSONEncoder",
     "find_default_player", "ignored", "progress", "stream_to_url"
 ]
 
diff --git a/src/streamlink_cli/utils/formatter.py b/src/streamlink_cli/utils/formatter.py
new file mode 100644
index 00000000..8fe640d7
--- /dev/null
+++ b/src/streamlink_cli/utils/formatter.py
@@ -0,0 +1,52 @@
+from typing import Callable, Dict, Optional, Type, Union
+
+from streamlink_cli.utils.path import replace_chars
+
+
+class _UNKNOWN:
+    pass
+
+
+class Formatter(dict):
+    def __init__(self, mapping: Dict[str, Callable]):
+        super().__init__()
+        self.mapping = mapping
+        self.cache = dict()
+
+    def __missing__(self, key: str) -> Union[Type[_UNKNOWN], None, str]:
+        if key not in self.mapping:
+            return _UNKNOWN
+
+        if key in self.cache:
+            return self.cache[key]
+
+        value = self.mapping[key]()
+        self.cache[key] = value
+
+        return value
+
+    def _format(self, string: str, wrapper: Callable[[str], str], defaults: Optional[Dict] = None) -> str:
+        return string.format_map(_Wrapper(self, wrapper, defaults))
+
+    def filename(self, filename: str, charmap: Optional[str] = None) -> str:
+        return self._format(filename, lambda s: replace_chars(s, charmap))
+
+    def title(self, title: str, defaults: Optional[Dict] = None) -> str:
+        return self._format(title, lambda s: s, defaults)
+
+
+class _Wrapper(dict):
+    def __init__(self, formatter: Formatter, wrapper: Callable[[str], str], defaults: Optional[Dict[str, str]] = None):
+        super().__init__()
+        self.formatter = formatter
+        self.wrapper = wrapper
+        self.defaults = defaults or {}
+
+    def __missing__(self, key: str) -> str:
+        value = self.formatter[key]
+        if value is _UNKNOWN:
+            value = str(self.defaults.get(key, f"{{{key}}}"))
+        elif value is None:
+            value = str(self.defaults.get(key, ""))
+
+        return self.wrapper(value)
diff --git a/src/streamlink_cli/utils/path.py b/src/streamlink_cli/utils/path.py
new file mode 100644
index 00000000..d794478b
--- /dev/null
+++ b/src/streamlink_cli/utils/path.py
@@ -0,0 +1,33 @@
+import re
+from typing import Optional
+
+from streamlink_cli.compat import is_win32
+
+
+REPLACEMENT = "_"
+
+_UNPRINTABLE = "".join(chr(c) for c in range(32))
+_UNSUPPORTED_POSIX = "/"
+_UNSUPPORTED_WIN32 = "\x7f\"*/:<>?\\|"
+
+RE_CHARS_POSIX = re.compile(f"[{re.escape(_UNPRINTABLE + _UNSUPPORTED_POSIX)}]+")
+RE_CHARS_WIN32 = re.compile(f"[{re.escape(_UNPRINTABLE + _UNSUPPORTED_WIN32)}]+")
+if is_win32:
+    RE_CHARS = RE_CHARS_WIN32
+else:
+    RE_CHARS = RE_CHARS_POSIX
+
+
+def replace_chars(path, charmap: Optional[str] = None, replacement=REPLACEMENT):
+    if charmap is None:
+        pattern = RE_CHARS
+    else:
+        charmap = charmap.lower()
+        if charmap in ("posix", "unix"):
+            pattern = RE_CHARS_POSIX
+        elif charmap in ("windows", "win32"):
+            pattern = RE_CHARS_WIN32
+        else:
+            raise ValueError("Invalid charmap")
+
+    return pattern.sub(replacement, path)


Test patch:
diff --git a/tests/resources/__init__.py b/tests/resources/__init__.py
index 40786e77..83746968 100644
--- a/tests/resources/__init__.py
+++ b/tests/resources/__init__.py
@@ -1,17 +1,18 @@
 import codecs
 import os.path
-import xml.etree.ElementTree as ET
 from contextlib import contextmanager
 from io import BytesIO
 
+from lxml.etree import iterparse
+
 
 __here__ = os.path.abspath(os.path.dirname(__file__))
 
 
 def _parse_xml(data, strip_ns=False):
-    data = bytearray(data, "utf8")
+    data = bytes(data, "utf8")
     try:
-        it = ET.iterparse(BytesIO(data))
+        it = iterparse(BytesIO(data))
         for _, el in it:
             if '}' in el.tag and strip_ns:  # pragma: no branch
                 # strip all namespaces
diff --git a/tests/test_api_validate.py b/tests/test_api_validate.py
index c2a0800e..a8663f7a 100644
--- a/tests/test_api_validate.py
+++ b/tests/test_api_validate.py
@@ -1,11 +1,12 @@
 import re
 import unittest
-from xml.etree.ElementTree import Element
+
+from lxml.etree import Element
 
 from streamlink.plugin.api.validate import (
     all, any, attr, endswith, filter, get, getattr, hasattr,
     length, map, optional, startswith, text, transform, union, union_get, url,
-    validate, xml_element, xml_find, xml_findall, xml_findtext
+    validate, xml_element, xml_find, xml_findall, xml_findtext, xml_xpath, xml_xpath_string
 )
 
 
@@ -94,6 +95,7 @@ class TestPluginAPIValidate(unittest.TestCase):
         assert validate(get("invalidkey"), {"key": "value"}) is None
         assert validate(get("invalidkey", "default"), {"key": "value"}) == "default"
         assert validate(get(3, "default"), [0, 1, 2]) == "default"
+        assert validate(get("attr"), Element("foo", {"attr": "value"})) == "value"
 
         with self.assertRaisesRegex(ValueError, "'NoneType' object is not subscriptable"):
             validate(get("key"), None)
@@ -134,12 +136,33 @@ class TestPluginAPIValidate(unittest.TestCase):
         self.assertRaises(ValueError, invalid_length)
 
     def test_xml_element(self):
-        el = Element("tag", attrib={"key": "value"})
+        el = Element("tag")
+        el.set("key", "value")
         el.text = "test"
+        childA = Element("childA")
+        childB = Element("childB")
+        el.append(childA)
+        el.append(childB)
+
+        upper = transform(str.upper)
+        newelem: Element = validate(xml_element(tag=upper, text=upper, attrib={upper: upper}), el)
+
+        assert newelem.tag == "TAG"
+        assert newelem.text == "TEST"
+        assert newelem.attrib == {"KEY": "VALUE"}
+        assert list(newelem.iterchildren()) == [childA, childB]
+
+        with self.assertRaises(ValueError) as cm:
+            validate(xml_element(tag="invalid"), el)
+        assert str(cm.exception).startswith("Unable to validate XML tag: ")
 
-        assert validate(xml_element("tag"), el).tag == "tag"
-        assert validate(xml_element(text="test"), el).text == "test"
-        assert validate(xml_element(attrib={"key": text}), el).attrib == {"key": "value"}
+        with self.assertRaises(ValueError) as cm:
+            validate(xml_element(text="invalid"), el)
+        assert str(cm.exception).startswith("Unable to validate XML text: ")
+
+        with self.assertRaises(ValueError) as cm:
+            validate(xml_element(attrib={"key": "invalid"}), el)
+        assert str(cm.exception).startswith("Unable to validate XML attributes: ")
 
     def test_xml_find(self):
         el = Element("parent")
@@ -148,6 +171,10 @@ class TestPluginAPIValidate(unittest.TestCase):
 
         assert validate(xml_find("bar"), el).tag == "bar"
 
+        with self.assertRaises(ValueError) as cm:
+            validate(xml_find("baz"), el)
+        assert str(cm.exception) == "XPath 'baz' did not return an element"
+
     def test_xml_findtext(self):
         el = Element("foo")
         el.text = "bar"
@@ -162,6 +189,37 @@ class TestPluginAPIValidate(unittest.TestCase):
 
         assert validate(xml_findall("child"), el) == children
 
+    def test_xml_xpath(self):
+        root = Element("root")
+        foo = Element("foo")
+        bar = Element("bar")
+        baz = Element("baz")
+        root.append(foo)
+        root.append(bar)
+        foo.append(baz)
+
+        assert validate(xml_xpath("./descendant-or-self::node()"), root) == [root, foo, baz, bar], "Returns correct node set"
+        assert validate(xml_xpath("./child::qux"), root) is None, "Returns None when node set is empty"
+        assert validate(xml_xpath("name(.)"), root) == "root", "Returns function values instead of node sets"
+        self.assertRaises(ValueError, validate, xml_xpath("."), "not an Element")
+
+    def test_xml_xpath_string(self):
+        root = Element("root")
+        foo = Element("foo")
+        bar = Element("bar")
+        root.set("attr", "")
+        foo.set("attr", "FOO")
+        bar.set("attr", "BAR")
+        root.append(foo)
+        root.append(bar)
+
+        assert validate(xml_xpath_string("./baz"), root) is None, "Returns None if nothing was found"
+        assert validate(xml_xpath_string("./@attr"), root) is None, "Returns None if string is empty"
+        assert validate(xml_xpath_string("./foo/@attr"), root) == "FOO", "Returns the attr value of foo"
+        assert validate(xml_xpath_string("./bar/@attr"), root) == "BAR", "Returns the attr value of bar"
+        assert validate(xml_xpath_string("count(./*)"), root) == "2", "Wraps arbitrary functions"
+        assert validate(xml_xpath_string("./*/@attr"), root) == "FOO", "Returns the first item of a set of nodes"
+
     def test_attr(self):
         el = Element("foo")
         el.text = "bar"
diff --git a/tests/test_cli_main.py b/tests/test_cli_main.py
index 178a63cf..c43cac4c 100644
--- a/tests/test_cli_main.py
+++ b/tests/test_cli_main.py
@@ -1,7 +1,6 @@
 import datetime
 import os
 import sys
-import tempfile
 import unittest
 from pathlib import Path, PosixPath, WindowsPath
 from unittest.mock import Mock, call, patch
@@ -12,8 +11,10 @@ import streamlink_cli.main
 import tests.resources
 from streamlink.plugin.plugin import Plugin
 from streamlink.session import Streamlink
-from streamlink_cli.compat import DeprecatedPath, is_win32
+from streamlink.stream.stream import Stream
+from streamlink_cli.compat import DeprecatedPath, is_win32, stdout
 from streamlink_cli.main import (
+    Formatter,
     NoPluginError,
     check_file_output,
     create_output,
@@ -27,61 +28,12 @@ from streamlink_cli.main import (
 from streamlink_cli.output import FileOutput, PlayerOutput
 
 
-class FakePlugin:
-    @classmethod
-    def stream_weight(cls, stream):
-        return Plugin.stream_weight(stream)
+class FakePlugin(Plugin):
+    def _get_streams(self):
+        pass  # pragma: no cover
 
 
 class TestCLIMain(unittest.TestCase):
-    def test_check_file_output(self):
-        streamlink_cli.main.console = Mock()
-        self.assertIsInstance(check_file_output("test", False), FileOutput)
-
-    def test_check_file_output_exists(self):
-        tmpfile = tempfile.NamedTemporaryFile()
-        try:
-            streamlink_cli.main.console = console = Mock()
-            streamlink_cli.main.sys.stdin = stdin = Mock()
-            stdin.isatty.return_value = True
-            console.ask.return_value = "y"
-            self.assertTrue(os.path.exists(tmpfile.name))
-            self.assertIsInstance(check_file_output(tmpfile.name, False), FileOutput)
-        finally:
-            tmpfile.close()
-
-    def test_check_file_output_exists_notty(self):
-        tmpfile = tempfile.NamedTemporaryFile()
-        try:
-            streamlink_cli.main.console = Mock()
-            streamlink_cli.main.sys.stdin = stdin = Mock()
-            stdin.isatty.return_value = False
-            self.assertTrue(os.path.exists(tmpfile.name))
-            self.assertRaises(SystemExit, check_file_output, tmpfile.name, False)
-        finally:
-            tmpfile.close()
-
-    def test_check_file_output_exists_force(self):
-        tmpfile = tempfile.NamedTemporaryFile()
-        try:
-            streamlink_cli.main.console = Mock()
-            self.assertTrue(os.path.exists(tmpfile.name))
-            self.assertIsInstance(check_file_output(tmpfile.name, True), FileOutput)
-        finally:
-            tmpfile.close()
-
-    @patch('sys.exit')
-    def test_check_file_output_exists_no(self, sys_exit):
-        tmpfile = tempfile.NamedTemporaryFile()
-        try:
-            streamlink_cli.main.console = console = Mock()
-            console.ask.return_value = "n"
-            self.assertTrue(os.path.exists(tmpfile.name))
-            check_file_output(tmpfile.name, False)
-            sys_exit.assert_called_with()
-        finally:
-            tmpfile.close()
-
     def test_resolve_stream_name(self):
         a = Mock()
         b = Mock()
@@ -148,12 +100,14 @@ class TestCLIMain(unittest.TestCase):
             ])
         )
 
+
+class TestCLIMainJsonAndStreamUrl(unittest.TestCase):
     @patch("streamlink_cli.main.args", stream_url=True, subprocess_cmdline=False)
     @patch("streamlink_cli.main.console", json=True)
     def test_handle_stream_with_json_and_stream_url(self, console, args):
         stream = Mock()
         streams = dict(best=stream)
-        plugin = Mock(FakePlugin(), module="fake", arguments=[], streams=Mock(return_value=streams))
+        plugin = Mock(FakePlugin(""), module="fake", arguments=[], streams=Mock(return_value=streams))
 
         handle_stream(plugin, streams, "best")
         self.assertEqual(console.msg.mock_calls, [])
@@ -179,7 +133,7 @@ class TestCLIMain(unittest.TestCase):
     def test_handle_url_with_json_and_stream_url(self, console, args):
         stream = Mock()
         streams = dict(worst=Mock(), best=stream)
-        plugin = Mock(FakePlugin(), module="fake", arguments=[], streams=Mock(return_value=streams))
+        plugin = Mock(FakePlugin(""), module="fake", arguments=[], streams=Mock(return_value=streams))
 
         with patch("streamlink_cli.main.streamlink", resolve_url=Mock(return_value=plugin)):
             handle_url()
@@ -202,76 +156,211 @@ class TestCLIMain(unittest.TestCase):
             self.assertEqual(console.exit.mock_calls, [call("The stream specified cannot be translated to a URL")])
             console.exit.mock_calls.clear()
 
-    def test_create_output_no_file_output_options(self):
-        streamlink_cli.main.console = Mock()
-        streamlink_cli.main.args = args = Mock()
+
+class TestCLIMainCheckFileOutput(unittest.TestCase):
+    @patch("streamlink_cli.main.os.path.isfile", Mock(return_value=False))
+    def test_check_file_output(self):
+        output = check_file_output("foo", False)
+        self.assertIsInstance(output, FileOutput)
+        self.assertEqual(output.filename, "foo")
+
+    @patch("streamlink_cli.main.os.path.isfile", Mock(return_value=True))
+    def test_check_file_output_exists_force(self):
+        output = check_file_output("foo", True)
+        self.assertIsInstance(output, FileOutput)
+        self.assertEqual(output.filename, "foo")
+
+    @patch("streamlink_cli.main.console", Mock(ask=Mock(return_value="y")))
+    @patch("streamlink_cli.main.os.path.isfile", Mock(return_value=True))
+    @patch("streamlink_cli.main.sys")
+    def test_check_file_output_exists_ask_yes(self, mock_sys: Mock):
+        mock_sys.stdin.isatty.return_value = True
+        output = check_file_output("foo", False)
+        self.assertIsInstance(output, FileOutput)
+        self.assertEqual(output.filename, "foo")
+
+    @patch("streamlink_cli.main.console", Mock(ask=Mock(return_value="N")))
+    @patch("streamlink_cli.main.os.path.isfile", Mock(return_value=True))
+    @patch("streamlink_cli.main.sys")
+    def test_check_file_output_exists_ask_no(self, mock_sys: Mock):
+        mock_sys.stdin.isatty.return_value = True
+        mock_sys.exit.side_effect = SystemExit
+        with self.assertRaises(SystemExit):
+            check_file_output("foo", False)
+
+    @patch("streamlink_cli.main.os.path.isfile", Mock(return_value=True))
+    @patch("streamlink_cli.main.sys")
+    def test_check_file_output_exists_notty(self, mock_sys: Mock):
+        mock_sys.stdin.isatty.return_value = False
+        mock_sys.exit.side_effect = SystemExit
+        with self.assertRaises(SystemExit):
+            check_file_output("foo", False)
+
+
+class TestCLIMainCreateOutput(unittest.TestCase):
+    @patch("streamlink_cli.main.args")
+    @patch("streamlink_cli.main.console", Mock())
+    @patch("streamlink_cli.main.DEFAULT_STREAM_METADATA", {"title": "bar"})
+    def test_create_output_no_file_output_options(self, args: Mock):
+        formatter = Formatter({
+            "author": lambda: "foo"
+        })
         args.output = None
         args.stdout = None
         args.record = None
         args.record_and_pipe = None
         args.title = None
+        args.url = "URL"
         args.player = "mpv"
         args.player_args = ""
-        self.assertIsInstance(create_output(FakePlugin), PlayerOutput)
-
-    def test_create_output_file_output(self):
-        tmpfile = tempfile.NamedTemporaryFile()
-        try:
-            streamlink_cli.main.args = args = Mock()
-            streamlink_cli.main.console = Mock()
-            args.output = tmpfile.name
-            self.assertTrue(os.path.exists(tmpfile.name))
-            self.assertIsInstance(create_output(FakePlugin), FileOutput)
-        finally:
-            tmpfile.close()
-
-    def test_create_output_stdout(self):
-        streamlink_cli.main.console = Mock()
-        streamlink_cli.main.args = args = Mock()
+
+        output = create_output(formatter)
+        self.assertIsInstance(output, PlayerOutput)
+        self.assertEqual(output.title, "URL")
+
+        args.title = "{author} - {title}"
+        output = create_output(formatter)
+        self.assertIsInstance(output, PlayerOutput)
+        self.assertEqual(output.title, "foo - bar")
+
+    @patch("streamlink_cli.main.os.path.isfile")
+    @patch("streamlink_cli.main.args")
+    @patch("streamlink_cli.main.console", Mock())
+    def test_create_output_file_output(self, args: Mock, mock_isfile):
+        formatter = Formatter({})
+        args.output = "foo"
+        args.force = False
+        args.fs_safe_rules = None
+        mock_isfile.return_value = False
+
+        output = create_output(formatter)
+        self.assertIsInstance(output, FileOutput)
+        self.assertEqual(output.filename, "foo")
+        self.assertEqual(output.fd, None)
+        self.assertEqual(output.record, None)
+
+    @patch("streamlink_cli.main.args")
+    @patch("streamlink_cli.main.console", Mock())
+    def test_create_output_stdout(self, args: Mock):
+        formatter = Formatter({})
         args.output = None
         args.stdout = True
-        self.assertIsInstance(create_output(FakePlugin), FileOutput)
-
-    def test_create_output_record_and_pipe(self):
-        tmpfile = tempfile.NamedTemporaryFile()
-        try:
-            streamlink_cli.main.console = Mock()
-            streamlink_cli.main.args = args = Mock()
-            args.output = None
-            args.stdout = None
-            args.record_and_pipe = tmpfile.name
-            self.assertIsInstance(create_output(FakePlugin), FileOutput)
-        finally:
-            tmpfile.close()
-
-    def test_create_output_record(self):
-        tmpfile = tempfile.NamedTemporaryFile()
-        try:
-            streamlink_cli.main.console = Mock()
-            streamlink_cli.main.args = args = Mock()
-            args.output = None
-            args.stdout = None
-            args.record = tmpfile.name
-            args.record_and_pipe = None
-            args.title = None
-            args.player = "mpv"
-            args.player_args = ""
-            args.player_fifo = None
-            self.assertIsInstance(create_output(FakePlugin), PlayerOutput)
-        finally:
-            tmpfile.close()
-
-    def test_create_output_record_and_other_file_output(self):
-        streamlink_cli.main.console = console = Mock()
-        streamlink_cli.main.args = args = Mock()
-        console.exit = Mock()
+
+        output = create_output(formatter)
+        self.assertIsInstance(output, FileOutput)
+        self.assertEqual(output.filename, None)
+        self.assertEqual(output.fd, stdout)
+        self.assertEqual(output.record, None)
+
+        args.output = "-"
+        args.stdout = False
+        output = create_output(formatter)
+        self.assertIsInstance(output, FileOutput)
+        self.assertEqual(output.filename, None)
+        self.assertEqual(output.fd, stdout)
+        self.assertEqual(output.record, None)
+
+    @patch("streamlink_cli.main.os.path.isfile")
+    @patch("streamlink_cli.main.args")
+    @patch("streamlink_cli.main.console", Mock())
+    def test_create_output_record_and_pipe(self, args: Mock, mock_isfile: Mock):
+        formatter = Formatter({})
+        args.output = None
+        args.stdout = None
+        args.record_and_pipe = "foo"
+        args.fs_safe_rules = None
+        mock_isfile.return_value = False
+
+        output = create_output(formatter)
+        self.assertIsInstance(output, FileOutput)
+        self.assertEqual(output.filename, None)
+        self.assertEqual(output.fd, stdout)
+        self.assertIsInstance(output.record, FileOutput)
+        self.assertEqual(output.record.filename, "foo")
+        self.assertEqual(output.record.fd, None)
+        self.assertEqual(output.record.record, None)
+
+    @patch("streamlink_cli.main.args")
+    @patch("streamlink_cli.main.console", Mock())
+    @patch("streamlink_cli.main.DEFAULT_STREAM_METADATA", {"title": "bar"})
+    def test_create_output_record(self, args: Mock):
+        formatter = Formatter({
+            "author": lambda: "foo"
+        })
+        args.output = None
+        args.stdout = None
+        args.record = "foo"
+        args.record_and_pipe = None
+        args.fs_safe_rules = None
+        args.title = None
+        args.url = "URL"
+        args.player = "mpv"
+        args.player_args = ""
+        args.player_fifo = None
+        args.player_http = None
+
+        output = create_output(formatter)
+        self.assertIsInstance(output, PlayerOutput)
+        self.assertEqual(output.title, "URL")
+        self.assertIsInstance(output.record, FileOutput)
+        self.assertEqual(output.record.filename, "foo")
+        self.assertEqual(output.record.fd, None)
+        self.assertEqual(output.record.record, None)
+
+        args.title = "{author} - {title}"
+        output = create_output(formatter)
+        self.assertIsInstance(output, PlayerOutput)
+        self.assertEqual(output.title, "foo - bar")
+        self.assertIsInstance(output.record, FileOutput)
+        self.assertEqual(output.record.filename, "foo")
+        self.assertEqual(output.record.fd, None)
+        self.assertEqual(output.record.record, None)
+
+    @patch("streamlink_cli.main.args")
+    @patch("streamlink_cli.main.console")
+    def test_create_output_record_and_other_file_output(self, console: Mock, args: Mock):
+        formatter = Formatter({})
         args.output = None
         args.stdout = True
         args.record_and_pipe = True
-        create_output(FakePlugin)
+        create_output(formatter)
         console.exit.assert_called_with("Cannot use record options with other file output options.")
 
 
+class TestCLIMainHandleStream(unittest.TestCase):
+    @patch("streamlink_cli.main.output_stream")
+    @patch("streamlink_cli.main.args")
+    @patch("streamlink_cli.main.console")
+    def test_handle_stream_output_stream(self, console: Mock, args: Mock, mock_output_stream: Mock):
+        """
+        Test that the formatter does define the correct variables
+        """
+        console.json = False
+        args.subprocess_cmdline = False
+        args.stream_url = False
+        args.output = False
+        args.stdout = False
+        args.url = "URL"
+        args.player_passthrough = []
+        args.player_external_http = False
+        args.player_continuous_http = False
+        mock_output_stream.return_value = True
+
+        plugin = Mock(FakePlugin(""), get_author=lambda: "AUTHOR", get_category=lambda: "CATEGORY", get_title=lambda: "TITLE")
+        stream = Stream(session=Mock())
+        streams = {"best": stream}
+
+        handle_stream(plugin, streams, "best")
+        self.assertEqual(mock_output_stream.call_count, 1)
+        paramStream, paramFormatter = mock_output_stream.call_args[0]
+        self.assertIs(paramStream, stream)
+        self.assertIsInstance(paramFormatter, Formatter)
+        self.assertEqual(
+            paramFormatter.title("{url} - {author} - {category}/{game} - {title}"),
+            "URL - AUTHOR - CATEGORY/CATEGORY - TITLE"
+        )
+
+
 @patch("streamlink_cli.main.log")
 class TestCLIMainSetupConfigArgs(unittest.TestCase):
     configdir = Path(tests.resources.__path__[0], "cli", "config")
diff --git a/tests/test_cli_utils_formatter.py b/tests/test_cli_utils_formatter.py
new file mode 100644
index 00000000..0a3d40a5
--- /dev/null
+++ b/tests/test_cli_utils_formatter.py
@@ -0,0 +1,47 @@
+import unittest
+from unittest.mock import Mock, call, patch
+
+from streamlink_cli.utils.formatter import Formatter
+
+
+class TestFormatter(unittest.TestCase):
+    def setUp(self):
+        self.prop = Mock(return_value="prop")
+        self.formatter = Formatter({
+            "prop": self.prop,
+            "empty": lambda: "",
+            "none": lambda: None
+        })
+
+    def test_unknown(self):
+        self.assertEqual(self.formatter.title("some {unknown} variable"), "some {unknown} variable")
+        self.assertEqual(self.formatter.title("some {unknown} variable", {"unknown": "known"}), "some known variable")
+        self.assertEqual(self.formatter.cache, dict())
+
+    def test_title(self):
+        self.assertEqual(self.formatter.title("text '{prop}' '{empty}' '{none}'"), "text 'prop' '' ''")
+        self.assertEqual(self.formatter.cache, dict(prop="prop", empty="", none=None))
+        self.assertEqual(self.prop.call_count, 1)
+
+        self.assertEqual(self.formatter.title("text '{prop}' '{empty}' '{none}'"), "text 'prop' '' ''")
+        self.assertEqual(self.prop.call_count, 1)
+
+    @patch("streamlink_cli.utils.formatter.replace_chars")
+    def test_filename(self, mock_replace_chars: Mock):
+        mock_replace_chars.side_effect = lambda s, *_: s.upper()
+
+        self.assertEqual(self.formatter.filename("text '{prop}' '{empty}' '{none}'"), "text 'PROP' '' ''")
+        self.assertEqual(self.formatter.cache, dict(prop="prop", empty="", none=None))
+        self.assertEqual(self.prop.call_count, 1)
+        self.assertEqual(mock_replace_chars.call_args_list, [
+            call("prop", None), call("", None), call("", None)
+        ])
+
+        mock_replace_chars.reset_mock()
+
+        self.assertEqual(self.formatter.filename("text '{prop}' '{empty}' '{none}'", "foo"), "text 'PROP' '' ''")
+        self.assertEqual(self.formatter.cache, dict(prop="prop", empty="", none=None))
+        self.assertEqual(self.prop.call_count, 1)
+        self.assertEqual(mock_replace_chars.call_args_list, [
+            call("prop", "foo"), call("", "foo"), call("", "foo")
+        ])
diff --git a/tests/test_cli_utils_path.py b/tests/test_cli_utils_path.py
new file mode 100644
index 00000000..86a188e2
--- /dev/null
+++ b/tests/test_cli_utils_path.py
@@ -0,0 +1,55 @@
+import pytest
+
+from streamlink_cli.utils.path import replace_chars
+from tests import posix_only, windows_only
+
+
+@pytest.mark.parametrize("char", [i for i in range(32)])
+def test_replace_chars_unprintable(char: int):
+    assert replace_chars(f"foo{chr(char)}{chr(char)}bar") == "foo_bar", "Replaces unprintable characters"
+
+
+@posix_only
+@pytest.mark.parametrize("char", "/".split())
+def test_replace_chars_posix(char: str):
+    assert replace_chars(f"foo{char}{char}bar") == "foo_bar", "Replaces multiple unsupported characters in a row"
+
+
+@windows_only
+@pytest.mark.parametrize("char", "\x7f\"*/:<>?\\|".split())
+def test_replace_chars_windows(char: str):
+    assert replace_chars(f"foo{char}{char}bar") == "foo_bar", "Replaces multiple unsupported characters in a row"
+
+
+@posix_only
+def test_replace_chars_posix_all():
+    assert replace_chars("".join(chr(i) for i in range(32)) + "/") == "_"
+
+
+@windows_only
+def test_replace_chars_windows_all():
+    assert replace_chars("".join(chr(i) for i in range(32)) + "\x7f\"*/:<>?\\|") == "_"
+
+
+@posix_only
+def test_replace_chars_posix_override():
+    all_chars = "".join(chr(i) for i in range(32)) + "\x7f\"*:/<>?\\|"
+    assert replace_chars(all_chars) == "_\x7f\"*:_<>?\\|"
+    assert replace_chars(all_chars, "posix") == "_\x7f\"*:_<>?\\|"
+    assert replace_chars(all_chars, "unix") == "_\x7f\"*:_<>?\\|"
+    assert replace_chars(all_chars, "windows") == "_"
+    assert replace_chars(all_chars, "win32") == "_"
+
+
+@windows_only
+def test_replace_chars_windows_override():
+    all_chars = "".join(chr(i) for i in range(32)) + "\x7f\"*:/<>?\\|"
+    assert replace_chars(all_chars) == "_"
+    assert replace_chars(all_chars, "posix") == "_\x7f\"*:_<>?\\|"
+    assert replace_chars(all_chars, "unix") == "_\x7f\"*:_<>?\\|"
+    assert replace_chars(all_chars, "windows") == "_"
+    assert replace_chars(all_chars, "win32") == "_"
+
+
+def test_replace_chars_replacement():
+    assert replace_chars("\x00", None, "+") == "+"
diff --git a/tests/test_utils.py b/tests/test_utils.py
index baa3dc88..216eadb6 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -2,11 +2,12 @@ import base64
 import os.path
 import sys
 import unittest
-import xml.etree.ElementTree as ET
+
+from lxml.etree import Element
 
 from streamlink.exceptions import PluginError
 from streamlink.plugin.api import validate
-from streamlink.plugin.api.validate import text, xml_element
+from streamlink.plugin.api.validate import xml_element
 from streamlink.utils import (
     absolute_url,
     load_module,
@@ -53,25 +54,25 @@ class TestUtil(unittest.TestCase):
         self.assertRaises(PluginError, parse_json, """{"test: 1}""" * 10)
 
     def test_parse_xml(self):
-        expected = ET.Element("test", {"foo": "bar"})
+        expected = Element("test", {"foo": "bar"})
         actual = parse_xml("""<test foo="bar"/>""", ignore_ns=True)
         self.assertEqual(expected.tag, actual.tag)
         self.assertEqual(expected.attrib, actual.attrib)
 
     def test_parse_xml_ns_ignore(self):
-        expected = ET.Element("test", {"foo": "bar"})
+        expected = Element("test", {"foo": "bar"})
         actual = parse_xml("""<test foo="bar" xmlns="foo:bar"/>""", ignore_ns=True)
         self.assertEqual(expected.tag, actual.tag)
         self.assertEqual(expected.attrib, actual.attrib)
 
     def test_parse_xml_ns_ignore_tab(self):
-        expected = ET.Element("test", {"foo": "bar"})
+        expected = Element("test", {"foo": "bar"})
         actual = parse_xml("""<test	foo="bar"	xmlns="foo:bar"/>""", ignore_ns=True)
         self.assertEqual(expected.tag, actual.tag)
         self.assertEqual(expected.attrib, actual.attrib)
 
     def test_parse_xml_ns(self):
-        expected = ET.Element("{foo:bar}test", {"foo": "bar"})
+        expected = Element("{foo:bar}test", {"foo": "bar"})
         actual = parse_xml("""<h:test foo="bar" xmlns:h="foo:bar"/>""")
         self.assertEqual(expected.tag, actual.tag)
         self.assertEqual(expected.attrib, actual.attrib)
@@ -83,9 +84,9 @@ class TestUtil(unittest.TestCase):
                           parse_xml, "1" * 1000, exception=IOError)
 
     def test_parse_xml_validate(self):
-        expected = ET.Element("test", {"foo": "bar"})
+        expected = Element("test", {"foo": "bar"})
         actual = parse_xml("""<test foo="bar"/>""",
-                           schema=validate.Schema(xml_element(tag="test", attrib={"foo": text})))
+                           schema=validate.Schema(xml_element(tag="test", attrib={"foo": str})))
         self.assertEqual(expected.tag, actual.tag)
         self.assertEqual(expected.attrib, actual.attrib)
 
@@ -94,9 +95,9 @@ class TestUtil(unittest.TestCase):
                           parse_xml, """<test foo="bar &"/>""")
 
     def test_parse_xml_entities(self):
-        expected = ET.Element("test", {"foo": "bar &"})
+        expected = Element("test", {"foo": "bar &"})
         actual = parse_xml("""<test foo="bar &"/>""",
-                           schema=validate.Schema(xml_element(tag="test", attrib={"foo": text})),
+                           schema=validate.Schema(xml_element(tag="test", attrib={"foo": str})),
                            invalid_char_entities=True)
         self.assertEqual(expected.tag, actual.tag)
         self.assertEqual(expected.attrib, actual.attrib)
@@ -104,7 +105,7 @@ class TestUtil(unittest.TestCase):
     def test_parse_qsd(self):
         self.assertEqual(
             {"test": "1", "foo": "bar"},
-            parse_qsd("test=1&foo=bar", schema=validate.Schema({"test": validate.text, "foo": "bar"})))
+            parse_qsd("test=1&foo=bar", schema=validate.Schema({"test": str, "foo": "bar"})))
 
     def test_rtmpparse(self):
         self.assertEqual(