Repository: streamlink/streamlink
Pull request #1903 (2018-07-05 16:17:18)

utils.args: Moved streamlink_cli utils.args into streamlink package
make it usable for plugin args

added tests

---

ref https://github.com/streamlink/streamlink/pull/1902#discussion_r200304653


Patch:
diff --git a/src/streamlink/utils/args.py b/src/streamlink/utils/args.py
new file mode 100644
index 00000000..f0cb2694
--- /dev/null
+++ b/src/streamlink/utils/args.py
@@ -0,0 +1,88 @@
+import argparse
+import re
+
+_filesize_re = re.compile(r'''
+    (?P<size>\d+(\.\d+)?)
+    (?P<modifier>[Kk]|[Mm])?
+    (?:[Bb])?
+''', re.VERBOSE)
+_keyvalue_re = re.compile(r'(?P<key>[^=]+)\s*=\s*(?P<value>.*)')
+
+
+def boolean(value):
+    truths = ['yes', '1', 'true', 'on']
+    falses = ['no', '0', 'false', 'off']
+    if value.lower() not in truths + falses:
+        raise argparse.ArgumentTypeError('{0} was not one of {{{1}}}'.format(
+            value, ', '.join(truths + falses)))
+
+    return value.lower() in truths
+
+
+def comma_list(values):
+    return [val.strip() for val in values.split(',')]
+
+
+def comma_list_filter(acceptable):
+    def func(p):
+        values = comma_list(p)
+        return list(filter(lambda v: v in acceptable, values))
+
+    return func
+
+
+def filesize(value):
+    match = _filesize_re.match(value)
+    if not match:
+        raise ValueError
+
+    size = float(match.group('size'))
+    if not size:
+        raise ValueError
+
+    modifier = match.group('modifier')
+    if modifier in ('M', 'm'):
+        size *= 1024 * 1024
+    elif modifier in ('K', 'k'):
+        size *= 1024
+
+    return num(int, min=0)(size)
+
+
+def keyvalue(value):
+    match = _keyvalue_re.match(value)
+    if not match:
+        raise ValueError
+
+    return match.group('key', 'value')
+
+
+def num(type, min=None, max=None):
+    def func(value):
+        value = type(value)
+
+        if min is not None and not (value > min):
+            raise argparse.ArgumentTypeError(
+                '{0} value must be more than {1} but is {2}'.format(
+                    type.__name__, min, value
+                )
+            )
+
+        if max is not None and not (value <= max):
+            raise argparse.ArgumentTypeError(
+                '{0} value must be at most {1} but is {2}'.format(
+                    type.__name__, max, value
+                )
+            )
+
+        return value
+
+    func.__name__ = type.__name__
+
+    return func
+
+
+__all__ = [
+    'boolean', 'comma_list', 'comma_list_filter', 'filesize', 'keyvalue',
+    'num'
+]
diff --git a/src/streamlink_cli/argparser.py b/src/streamlink_cli/argparser.py
index 921bdd37..de847660 100644
--- a/src/streamlink_cli/argparser.py
+++ b/src/streamlink_cli/argparser.py
@@ -4,18 +4,15 @@ from string import printable
 from textwrap import dedent
 
 from streamlink import logger
+from streamlink.utils.args import (
+    boolean, comma_list, comma_list_filter, filesize, keyvalue, num
+)
 from streamlink.utils.times import hours_minutes_seconds
 from .constants import (
     LIVESTREAMER_VERSION, STREAM_PASSTHROUGH, DEFAULT_PLAYER_ARGUMENTS
 )
 from .utils import find_default_player
 
-_filesize_re = re.compile(r"""
-    (?P<size>\d+(\.\d+)?)
-    (?P<modifier>[Kk]|[Mm])?
-    (?:[Bb])?
-""", re.VERBOSE)
-_keyvalue_re = re.compile(r"(?P<key>[^=]+)\s*=\s*(?P<value>.*)")
 _printable_re = re.compile(r"[{0}]".format(printable))
 _option_re = re.compile(r"""
     (?P<name>[A-z-]+) # A option name, valid characters are A to z and dash.
@@ -68,78 +65,6 @@ class HelpFormatter(argparse.RawDescriptionHelpFormatter):
         return text.splitlines()
 
 
-def comma_list(values):
-    return [val.strip() for val in values.split(",")]
-
-
-def comma_list_filter(acceptable):
-    def func(p):
-        values = comma_list(p)
-        return list(filter(lambda v: v in acceptable, values))
-
-    return func
-
-
-def num(type, min=None, max=None):
-    def func(value):
-        value = type(value)
-
-        if min is not None and not (value > min):
-            raise argparse.ArgumentTypeError(
-                "{0} value must be more than {1} but is {2}".format(
-                    type.__name__, min, value
-                )
-            )
-
-        if max is not None and not (value <= max):
-            raise argparse.ArgumentTypeError(
-                "{0} value must be at most {1} but is {2}".format(
-                    type.__name__, max, value
-                )
-            )
-
-        return value
-
-    func.__name__ = type.__name__
-
-    return func
-
-
-def filesize(value):
-    match = _filesize_re.match(value)
-    if not match:
-        raise ValueError
-
-    size = float(match.group("size"))
-    if not size:
-        raise ValueError
-
-    modifier = match.group("modifier")
-    if modifier in ("M", "m"):
-        size *= 1024 * 1024
-    elif modifier in ("K", "k"):
-        size *= 1024
-
-    return num(int, min=0)(size)
-
-
-def keyvalue(value):
-    match = _keyvalue_re.match(value)
-    if not match:
-        raise ValueError
-
-    return match.group("key", "value")
-
-
-def boolean(value):
-    truths = ["yes", "1", "true", "on"]
-    falses = ["no", "0", "false", "off"]
-    if value.lower() not in truths + falses:
-        raise argparse.ArgumentTypeError("{0} was not one of {{{1}}}".format(value, ', '.join(truths + falses)))
-
-    return value.lower() in truths
-
-
 def build_parser():
     parser = ArgumentParser(
         prog="streamlink",


Test patch:
diff --git a/tests/test_utils_args.py b/tests/test_utils_args.py
new file mode 100644
index 00000000..b1d9e94e
--- /dev/null
+++ b/tests/test_utils_args.py
@@ -0,0 +1,123 @@
+import unittest
+
+from argparse import ArgumentTypeError
+from streamlink.utils.args import (
+    boolean, comma_list, comma_list_filter, filesize, keyvalue, num
+)
+
+
+class TestUtilsArgs(unittest.TestCase):
+
+    def test_boolean_true(self):
+        self.assertEqual(boolean('1'), True)
+        self.assertEqual(boolean('on'), True)
+        self.assertEqual(boolean('true'), True)
+        self.assertEqual(boolean('yes'), True)
+        self.assertEqual(boolean('Yes'), True)
+
+    def test_boolean_false(self):
+        self.assertEqual(boolean('0'), False)
+        self.assertEqual(boolean('false'), False)
+        self.assertEqual(boolean('no'), False)
+        self.assertEqual(boolean('No'), False)
+        self.assertEqual(boolean('off'), False)
+
+    def test_boolean_error(self):
+        with self.assertRaises(ArgumentTypeError):
+            boolean('yesno')
+
+        with self.assertRaises(ArgumentTypeError):
+            boolean('FOO')
+
+        with self.assertRaises(ArgumentTypeError):
+            boolean('2')
+
+    def test_comma_list(self):
+        # (values, result)
+        test_data = [
+            ('foo.bar,example.com', ['foo.bar', 'example.com']),
+            ('/var/run/foo,/var/run/bar', ['/var/run/foo', '/var/run/bar']),
+            ('foo bar,24', ['foo bar', '24']),
+            ('hls', ['hls']),
+        ]
+
+        for _v, _r in test_data:
+            self.assertEqual(comma_list(_v), _r)
+
+    def test_comma_list_filter(self):
+        # (acceptable, values, result)
+        test_data = [
+            (['foo', 'bar', 'com'], 'foo,bar,example.com', ['foo', 'bar']),
+            (['/var/run/foo', 'FO'], '/var/run/foo,/var/run/bar',
+             ['/var/run/foo']),
+            (['hls', 'hls5', 'dash'], 'hls,hls5', ['hls', 'hls5']),
+            (['EU', 'RU'], 'DE,FR,RU,US', ['RU']),
+        ]
+
+        for _a, _v, _r in test_data:
+            func = comma_list_filter(_a)
+            self.assertEqual(func(_v), _r)
+
+    def test_filesize(self):
+        self.assertEqual(filesize('2000'), 2000)
+        self.assertEqual(filesize('11KB'), 1024 * 11)
+        self.assertEqual(filesize('12MB'), 1024 * 1024 * 12)
+        self.assertEqual(filesize('1KB'), 1024)
+        self.assertEqual(filesize('1MB'), 1024 * 1024)
+        self.assertEqual(filesize('2KB'), 1024 * 2)
+
+    def test_filesize_error(self):
+        with self.assertRaises(ValueError):
+            filesize('FOO')
+
+        with self.assertRaises(ValueError):
+            filesize('0.00000')
+
+    def test_keyvalue(self):
+        # (value, result)
+        test_data = [
+            ('X-Forwarded-For=127.0.0.1', ('X-Forwarded-For', '127.0.0.1')),
+            ('Referer=https://foo.bar', ('Referer', 'https://foo.bar')),
+            (
+                'User-Agent=Mozilla/5.0 (X11; Linux x86_64; rv:60.0)'
+                ' Gecko/20100101 Firefox/60.0',
+                ('User-Agent', 'Mozilla/5.0 (X11; Linux x86_64; rv:60.0) '
+                 'Gecko/20100101 Firefox/60.0')
+            ),
+            ('domain=example.com', ('domain', 'example.com')),
+        ]
+
+        for _v, _r in test_data:
+            self.assertEqual(keyvalue(_v), _r)
+
+    def test_keyvalue_error(self):
+        with self.assertRaises(ValueError):
+            keyvalue('127.0.0.1')
+
+    def test_num(self):
+        # (value, func, result)
+        test_data = [
+            ('33', num(int, 5, 120), 33),
+            ('234', num(int, min=10), 234),
+            ('50.222', num(float, 10, 120), 50.222),
+        ]
+
+        for _v, _f, _r in test_data:
+            self.assertEqual(_f(_v), _r)
+
+    def test_num_error(self):
+        with self.assertRaises(ArgumentTypeError):
+            func = num(int, 5, 10)
+            func('3')
+
+        with self.assertRaises(ArgumentTypeError):
+            func = num(int, max=11)
+            func('12')
+
+        with self.assertRaises(ArgumentTypeError):
+            func = num(int, min=15)
+            func('8')
+
+        with self.assertRaises(ArgumentTypeError):
+            func = num(float, 10, 20)
+            func('40.222')