| # Copyright 2023 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| from __future__ import annotations |
| |
| import argparse |
| import datetime as dt |
| import json |
| import math |
| import pathlib |
| import unittest |
| from typing import Any |
| from urllib import parse as urlparse |
| |
| from typing_extensions import override |
| |
| from crossbench import path as pth |
| from crossbench.parse import (DurationParseError, DurationParser, NumberParser, |
| ObjectParser, PathParser, TimeUnit) |
| from protoc import trace_config_pb2 |
| from tests import test_helper |
| from tests.crossbench.base import CrossbenchFakeFsTestCase |
| |
| |
| class DurationParserTestCase(unittest.TestCase): |
| |
| def test_parse_negative(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| DurationParser.positive_duration(-1) |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_duration("-1") |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_or_zero_duration("-1") |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_or_zero_duration(dt.timedelta(seconds=-1)) |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_duration("-1") |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_duration(dt.timedelta(seconds=-1)) |
| self.assertIn("-1", str(cm.exception)) |
| self.assertEqual(DurationParser.any_duration("-1.5").total_seconds(), -1.5) |
| |
| def test_parse_zero(self): |
| self.assertEqual(DurationParser.any_duration("0").total_seconds(), 0) |
| self.assertEqual(DurationParser.any_duration("0s").total_seconds(), 0) |
| self.assertEqual(DurationParser.any_duration("0.0").total_seconds(), 0) |
| self.assertEqual( |
| DurationParser.positive_or_zero_duration("0.0").total_seconds(), 0) |
| invalid: Any |
| for invalid in (-1, 0, "-1", "0", "invalid", dt.timedelta(0), |
| dt.timedelta(seconds=-1)): |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_duration(invalid) |
| self.assertIn(str(invalid), str(cm.exception)) |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_duration(invalid) |
| self.assertIn(str(invalid), str(cm.exception)) |
| |
| def test_parse_empty(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| DurationParser.positive_duration("") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| DurationParser.any_duration("") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| DurationParser.positive_or_zero_duration("") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| DurationParser.positive_duration("") |
| |
| def test_invalid_suffix(self): |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| DurationParser.positive_duration("100XXX") |
| self.assertIn("Unknown duration format", str(cm.exception)) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| DurationParser.positive_duration("X0XX") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| DurationParser.positive_duration("100X0XX") |
| |
| def test_no_unit(self): |
| self.assertEqual( |
| DurationParser.positive_duration("200"), dt.timedelta(seconds=200)) |
| self.assertEqual( |
| DurationParser.positive_duration(200), dt.timedelta(seconds=200)) |
| self.assertEqual( |
| DurationParser.positive_duration_ms("200"), |
| dt.timedelta(milliseconds=200)) |
| self.assertEqual( |
| DurationParser.positive_duration_ms(200), |
| dt.timedelta(milliseconds=200)) |
| |
| def test_milliseconds(self): |
| self.assertEqual( |
| DurationParser.positive_duration("27.5ms"), |
| dt.timedelta(milliseconds=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration(dt.timedelta(milliseconds=27.5)), |
| dt.timedelta(milliseconds=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 millis"), |
| dt.timedelta(milliseconds=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 milliseconds"), |
| dt.timedelta(milliseconds=27.5)) |
| |
| def test_seconds(self): |
| self.assertEqual( |
| DurationParser.positive_duration("27.5s"), dt.timedelta(seconds=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("1 sec"), dt.timedelta(seconds=1)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 secs"), |
| dt.timedelta(seconds=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("1 second"), dt.timedelta(seconds=1)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 seconds"), |
| dt.timedelta(seconds=27.5)) |
| |
| def test_minutes(self): |
| self.assertEqual( |
| DurationParser.positive_duration("27.5m"), dt.timedelta(minutes=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("1 min"), dt.timedelta(minutes=1)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 mins"), |
| dt.timedelta(minutes=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("1 minute"), dt.timedelta(minutes=1)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 minutes"), |
| dt.timedelta(minutes=27.5)) |
| |
| def test_hours(self): |
| self.assertEqual( |
| DurationParser.positive_duration("27.5h"), dt.timedelta(hours=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("0.1 h"), dt.timedelta(hours=0.1)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 hrs"), dt.timedelta(hours=27.5)) |
| self.assertEqual( |
| DurationParser.positive_duration("1 hour"), dt.timedelta(hours=1)) |
| self.assertEqual( |
| DurationParser.positive_duration("27.5 hours"), |
| dt.timedelta(hours=27.5)) |
| |
| |
| class PathParserTestCase(CrossbenchFakeFsTestCase): |
| |
| def test_existing_path(self): |
| path = pth.LocalPath("foo.txt") |
| self.assertFalse(path.exists()) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.existing_path(str(path)) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.existing_path(path) |
| path.touch() |
| self.assertEqual(path, PathParser.existing_path(str(path))) |
| self.assertEqual(path, PathParser.existing_path(path)) |
| |
| def test_not_existing_path(self): |
| path = pth.LocalPath("foo.txt") |
| self.assertFalse(path.exists()) |
| self.assertEqual(path, PathParser.not_existing_path(str(path))) |
| self.assertEqual(path, PathParser.not_existing_path(path)) |
| path.touch() |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.not_existing_path(str(path)) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.not_existing_path(path) |
| |
| |
| class ObjectParserTestCase(CrossbenchFakeFsTestCase): |
| |
| @override |
| def setUp(self): |
| super().setUp() |
| self._json_test_data = {"int": 1, "array": [1, "2"]} |
| |
| def test_parse_any_str(self): |
| self.assertEqual(ObjectParser.any_str(""), "") |
| self.assertEqual(ObjectParser.any_str("1234"), "1234") |
| |
| def test_parse_any_str_invalid(self): |
| invalid: Any |
| for invalid in (None, 1, [], {}, [1], ["a"], {"a": "a"}): |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| ObjectParser.any_str(invalid) |
| self.assertIn(str(invalid), str(cm.exception)) |
| |
| def test_parse_non_empty_str(self): |
| self.assertEqual(ObjectParser.non_empty_str("a string"), "a string") |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| ObjectParser.non_empty_str("") |
| self.assertIn("empty", str(cm.exception)) |
| |
| def test_parse_str_or_file_contents(self): |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, "empty"): |
| ObjectParser.str_or_file_contents("") |
| self.assertEqual( |
| ObjectParser.str_or_file_contents("some data"), "some data") |
| self.assertEqual(ObjectParser.str_or_file_contents("test.txt"), "test.txt") |
| |
| def test_parse_str_or_file_contents_file(self): |
| path = pathlib.Path("./test.txt") |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, str(path)): |
| ObjectParser.str_or_file_contents(path) |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, str(path)): |
| ObjectParser.str_or_file_contents("./test.txt") |
| self.fs.create_file(path, contents="test file contents") |
| self.assertEqual( |
| ObjectParser.str_or_file_contents(path), "test file contents") |
| self.assertEqual(ObjectParser.str_or_file_contents(str(path)), "test.txt") |
| self.assertEqual( |
| ObjectParser.str_or_file_contents("./test.txt"), "test file contents") |
| |
| def test_parse_httpx_url_str(self): |
| for valid in ("http://foo.com", "https://foo.com", "http://localhost:800"): |
| self.assertEqual(ObjectParser.httpx_url_str(valid), valid) |
| invalid: Any |
| for invalid in ("", "ftp://localhost:32", "http://///"): |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| _ = ObjectParser.httpx_url_str(invalid) |
| self.assertIn(invalid, str(cm.exception)) |
| |
| def test_parse_any_int(self): |
| self.assertEqual(NumberParser.any_int("-123456"), -123456) |
| self.assertEqual(NumberParser.any_int(-123456), -123456) |
| self.assertEqual(NumberParser.any_int(float(-123456)), -123456) |
| self.assertEqual(NumberParser.any_int("-1"), -1) |
| self.assertEqual(NumberParser.any_int(-1), -1) |
| self.assertEqual(NumberParser.any_int(float(-1)), -1) |
| self.assertEqual(NumberParser.any_int("0"), 0) |
| self.assertEqual(NumberParser.any_int(0), 0) |
| self.assertEqual(NumberParser.any_int(float(0)), 0) |
| self.assertEqual(NumberParser.any_int("1"), 1) |
| self.assertEqual(NumberParser.any_int(1), 1) |
| self.assertEqual(NumberParser.any_int("123456"), 123456) |
| self.assertEqual(NumberParser.any_int(123456), 123456) |
| |
| def test_parse_any_int_strict(self): |
| self.assertEqual(NumberParser.any_int(float(0), parse_str=False), 0) |
| self.assertEqual(NumberParser.any_int(1, parse_str=False), 1) |
| |
| def test_parse_any_int_invalid(self): |
| invalid: Any |
| for invalid in ("", "-1.2", -1.2, "1.2", 1.2, "100.001", 100.001, "Nan", |
| math.nan, "inf", math.inf, "-inf", -math.inf, "invalid", |
| None): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = NumberParser.any_int(invalid) |
| |
| def test_parse_any_int_invalid_strict(self): |
| invalid: Any |
| for invalid in ("", "-1.2", -1.2, "1.2", 1.2, "100.001", 100.001, "Nan", |
| math.nan, "inf", math.inf, "-inf", -math.inf, "invalid", |
| None): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = NumberParser.any_int(invalid, parse_str=False) |
| |
| def test_parse_positive_int(self): |
| self.assertEqual(NumberParser.positive_int("1"), 1) |
| self.assertEqual(NumberParser.positive_int(1), 1) |
| self.assertEqual(NumberParser.positive_int("123"), 123) |
| self.assertEqual(NumberParser.positive_int(123), 123) |
| |
| def test_parse_positive_int_invalid(self): |
| invalid: Any |
| for invalid in ("", "0", 0, "-1", -1, "-1.2", -1.2, "1.2", 1.2, "Nan", |
| math.nan, "inf", math.inf, "-inf", -math.inf, "invalid", |
| None): |
| with self.assertRaises( |
| argparse.ArgumentTypeError, msg=f"invalid={repr(invalid)}"): |
| _ = NumberParser.positive_int(invalid) |
| |
| def test_parse_int_range(self): |
| self.assertEqual(NumberParser.int_range(min=0, max=10)("1"), 1) |
| self.assertEqual(NumberParser.int_range(min=0, max=10)(1), 1) |
| self.assertEqual(NumberParser.int_range(min=0, max=200)("123"), 123) |
| self.assertEqual(NumberParser.int_range(min=0, max=200)(123), 123) |
| self.assertEqual(NumberParser.int_range(min=-100, max=200)("-12"), -12) |
| self.assertEqual(NumberParser.int_range(min=-100, max=200)(-12), -12) |
| |
| def test_parse_int_range_invalid(self): |
| with self.assertRaises(AssertionError): |
| NumberParser.int_range(1, 1) |
| with self.assertRaises(AssertionError): |
| NumberParser.int_range(10, 1) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| NumberParser.int_range(-1, 10)(-2) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| NumberParser.int_range(-1, 10)(11) |
| |
| def test_parse_positive_int_invalid_strict(self): |
| invalid: Any |
| for invalid in ("", "0", 0, "1", "-1", -1, float(-1), "-1.2", -1.2, "1.2", |
| 1.2, "Nan", math.nan, "inf", math.inf, "-inf", -math.inf, |
| "invalid", None): |
| with self.assertRaises( |
| argparse.ArgumentTypeError, msg=f"invalid={repr(invalid)}"): |
| _ = NumberParser.positive_int(invalid, parse_str=False) |
| |
| def test_parse_positive_zero_int(self): |
| self.assertEqual(NumberParser.positive_zero_int("1"), 1) |
| self.assertEqual(NumberParser.positive_zero_int(1), 1) |
| self.assertEqual(NumberParser.positive_zero_int(float(1)), 1) |
| self.assertEqual(NumberParser.positive_zero_int("0"), 0) |
| self.assertEqual(NumberParser.positive_zero_int(0), 0) |
| |
| def test_parse_positive_zero_int_invalid(self): |
| invalid: Any |
| for invalid in ("", "-1", -1, "-1.2", -1.2, "1.2", 1.2, "NaN", math.nan, |
| "inf", math.inf, "-inf", -math.inf, "invalid", None): |
| with self.assertRaises( |
| argparse.ArgumentTypeError, msg=f"invalid={repr(invalid)}"): |
| _ = NumberParser.positive_zero_int(invalid) |
| |
| def test_parse_any_float(self): |
| self.assertEqual(NumberParser.any_float("-1.2"), -1.2) |
| self.assertEqual(NumberParser.any_float(-1.2), -1.2) |
| self.assertEqual(NumberParser.any_float("-1"), -1.0) |
| self.assertEqual(NumberParser.any_float(-1), -1.0) |
| self.assertEqual(NumberParser.any_float("0"), 0.0) |
| self.assertEqual(NumberParser.any_float(0), 0.0) |
| self.assertEqual(NumberParser.any_float("0.0"), 0.0) |
| self.assertEqual(NumberParser.any_float(0.0), 0.0) |
| self.assertEqual(NumberParser.any_float("0.1"), 0.1) |
| self.assertEqual(NumberParser.any_float(0.1), 0.1) |
| |
| def test_parse_float_invalid(self): |
| invalid: Any |
| for invalid in ("", "abc", "NaN", "inf", "-inf", "invalid"): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = NumberParser.positive_zero_float(invalid) |
| |
| def test_parse_positive_zero_float(self): |
| self.assertEqual(NumberParser.positive_zero_float("1"), 1.0) |
| self.assertEqual(NumberParser.positive_zero_float("0"), 0.0) |
| self.assertEqual(NumberParser.positive_zero_float("0.0"), 0.0) |
| self.assertEqual(NumberParser.positive_zero_float("1.23"), 1.23) |
| |
| def test_parse_positive_zero_float_invalid(self): |
| invalid: Any |
| for invalid in ("", "-1", "-1.2", "NaN", "inf", "-inf", "invalid"): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = NumberParser.positive_zero_float(invalid) |
| |
| def test_parse_float_range(self): |
| self.assertEqual(NumberParser.float_range(min=1, max=2)("1"), 1.0) |
| self.assertEqual(NumberParser.float_range(min=0, max=1)(1), 1.0) |
| self.assertEqual(NumberParser.float_range(min=0, max=1)("0"), 0.0) |
| self.assertEqual(NumberParser.float_range(min=0, max=1)(0), 0.0) |
| self.assertEqual(NumberParser.float_range(min=0, max=1)("0.0"), 0.0) |
| self.assertEqual(NumberParser.float_range(min=0, max=1)(0.0), 0.0) |
| self.assertEqual(NumberParser.float_range(min=0, max=11)("1.23"), 1.23) |
| self.assertEqual(NumberParser.float_range(min=0, max=11)(1.23), 1.23) |
| self.assertEqual(NumberParser.float_range(min=-2, max=11)("-1.1"), -1.1) |
| self.assertEqual(NumberParser.float_range(min=-2, max=11)(-1.1), -1.1) |
| |
| def test_parse_float_range_invalid(self): |
| with self.assertRaises(AssertionError): |
| NumberParser.float_range(1, 1) |
| with self.assertRaises(AssertionError): |
| NumberParser.float_range(10, 1.0) |
| with self.assertRaises(AssertionError): |
| NumberParser.float_range(-10.1, -11.0) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| NumberParser.int_range(-1.1, 10.1)(-1.2) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| NumberParser.int_range(-1.1, 10.1)(10.2) |
| |
| def test_parse_port_number(self): |
| self.assertEqual(NumberParser.port_number(1), 1) |
| self.assertEqual(NumberParser.port_number("1"), 1) |
| self.assertEqual(NumberParser.port_number(440), 440) |
| self.assertEqual(NumberParser.port_number("440"), 440) |
| self.assertEqual(NumberParser.port_number(65535), 65535) |
| self.assertEqual(NumberParser.port_number("65535"), 65535) |
| |
| def test_parse_port_number_zer(self): |
| self.assertEqual(NumberParser.port_number_zero(0), 0) |
| self.assertEqual(NumberParser.port_number_zero("0"), 0) |
| self.assertEqual(NumberParser.port_number(1), 1) |
| self.assertEqual(NumberParser.port_number("1"), 1) |
| self.assertEqual(NumberParser.port_number_zero(440), 440) |
| self.assertEqual(NumberParser.port_number_zero("440"), 440) |
| self.assertEqual(NumberParser.port_number_zero(65535), 65535) |
| self.assertEqual(NumberParser.port_number_zero("65535"), 65535) |
| |
| def test_parse_port_number_invalid(self): |
| invalid: Any |
| for invalid in ("", "-1", "-1.2", "6553500", 6553500, "inf", "-inf", |
| "invalid", 0, "0"): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = NumberParser.port_number(invalid) |
| |
| def test_parse_port_number_invalid_zero(self): |
| invalid: Any |
| for invalid in ( |
| "", |
| "-1", |
| "-1.2", |
| "6553500", |
| "inf", |
| "-inf", |
| "invalid", |
| ): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = NumberParser.port_number_zero(invalid) |
| |
| def _json_file_test_helper(self, parser) -> Any: |
| with self.assertRaises(argparse.ArgumentTypeError): |
| parser("file") |
| |
| path = pathlib.Path("file.json") |
| self.assertFalse(path.exists()) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| parser(path) |
| |
| path.touch() |
| with self.assertRaises(argparse.ArgumentTypeError): |
| parser(path) |
| |
| with path.open("w", encoding="utf-8") as f: |
| f.write("{invalid json data") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| parser(path) |
| # Test very long lines too. |
| with path.open("w", encoding="utf-8") as f: |
| f.write("{\n invalid json data" + "." * 100) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| parser(path) |
| |
| with path.open("w", encoding="utf-8") as f: |
| f.write("""{ |
| 'a': {}, |
| 'c': }} |
| """) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| parser(path) |
| |
| with path.open("w", encoding="utf-8") as f: |
| json.dump(self._json_test_data, f) |
| str_result = parser(str(path)) |
| path_result = parser(path) |
| self.assertEqual(str_result, path_result) |
| return str_result |
| |
| def test_parse_json_file(self): |
| result = self._json_file_test_helper(ObjectParser.json_file) |
| self.assertDictEqual(self._json_test_data, result) |
| |
| def test_parse_json_file_path(self): |
| result = self._json_file_test_helper(PathParser.json_file_path) |
| self.assertEqual(pathlib.Path("file.json"), result) |
| |
| def test_parse_hjson_file_path(self): |
| result = self._json_file_test_helper(PathParser.hjson_file_path) |
| self.assertEqual(pathlib.Path("file.json"), result) |
| |
| def test_parse_inline_hjson(self): |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, "empty"): |
| ObjectParser.inline_hjson("") |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, "braces"): |
| ObjectParser.inline_hjson("{") |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, "braces"): |
| ObjectParser.inline_hjson("[]") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.inline_hjson("{invalid json}") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.inline_hjson("{'asdfas':'asdf}") |
| self.assertDictEqual( |
| self._json_test_data, |
| ObjectParser.inline_hjson(json.dumps(self._json_test_data))) |
| |
| def test_parse_dir_path(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.dir_path("") |
| file = pathlib.Path("file") |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| PathParser.dir_path(file) |
| self.assertIn("does not exist", str(cm.exception)) |
| file.touch() |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| PathParser.dir_path(file) |
| self.assertIn("not a folder", str(cm.exception)) |
| folder = pathlib.Path("folder") |
| folder.mkdir() |
| self.assertEqual(folder, PathParser.dir_path(folder)) |
| self.assertEqual(folder, PathParser.dir_path(str(folder))) |
| |
| def test_parse_non_empty_dir_path(self): |
| folder = pathlib.Path("folder") |
| folder.mkdir() |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| PathParser.non_empty_dir_path(folder) |
| self.assertIn("empty", str(cm.exception)) |
| (folder / "foo").touch() |
| self.assertEqual(folder, PathParser.non_empty_dir_path(folder)) |
| self.assertEqual(folder, PathParser.non_empty_dir_path(str(folder))) |
| |
| def test_parse_non_empty_file_path(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.non_empty_file_path("") |
| folder = pathlib.Path("folder") |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| PathParser.non_empty_file_path(folder) |
| self.assertIn("does not exist", str(cm.exception)) |
| folder.mkdir() |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| PathParser.non_empty_file_path(folder) |
| self.assertIn("not a file", str(cm.exception)) |
| file = pathlib.Path("file") |
| file.touch() |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| self.assertEqual(file, PathParser.non_empty_file_path(file)) |
| self.assertIn("is an empty file", str(cm.exception)) |
| |
| with file.open("w", encoding="utf-8") as f: |
| f.write("fooo") |
| self.assertEqual(file, PathParser.non_empty_file_path(file)) |
| |
| def test_parse_existing_file_path(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.existing_file_path("") |
| folder = pathlib.Path("folder") |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| PathParser.existing_file_path(folder) |
| self.assertIn("does not exist", str(cm.exception)) |
| folder.mkdir() |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| PathParser.existing_file_path(folder) |
| self.assertIn("not a file", str(cm.exception)) |
| file = pathlib.Path("file") |
| file.touch() |
| self.assertEqual(file, PathParser.existing_file_path(file)) |
| |
| def test_parse_path(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.path("") |
| folder = pathlib.Path("folder") |
| folder.mkdir() |
| self.assertEqual(folder, PathParser.path(folder)) |
| file = pathlib.Path("file") |
| file.touch() |
| self.assertEqual(file, PathParser.path(file)) |
| |
| def test_parse_any_path_invalid(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.any_path("") |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.any_path(None) |
| |
| def test_parse_any_path(self): |
| folder = pathlib.Path("folder") |
| folder_pure = pathlib.PurePath(folder) |
| self.assertEqual(folder_pure, PathParser.any_path(folder)) |
| folder.mkdir() |
| self.assertEqual(folder_pure, PathParser.any_path(folder)) |
| file = pathlib.Path("file") |
| file_pure = pathlib.PurePath(file) |
| self.assertEqual(file_pure, PathParser.any_path(file)) |
| file.touch() |
| self.assertEqual(file_pure, PathParser.any_path(file)) |
| |
| def test_parse_optional_any_path_invalid(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| PathParser.optional_any_path("") |
| |
| def test_parse_optional_any_path(self): |
| self.assertIsNone(PathParser.optional_any_path(None)) |
| folder = pathlib.Path("folder") |
| folder_pure = pathlib.PurePath(folder) |
| self.assertEqual(folder_pure, PathParser.optional_any_path(folder)) |
| |
| def test_parse_bool_success(self): |
| self.assertIs(ObjectParser.bool("true"), True) |
| self.assertIs(ObjectParser.bool("True"), True) |
| self.assertIs(ObjectParser.bool(True), True) |
| self.assertIs(ObjectParser.bool("false"), False) |
| self.assertIs(ObjectParser.bool("False"), False) |
| self.assertIs(ObjectParser.bool(False), False) |
| |
| def test_parse_bool_success_strict(self): |
| self.assertIs(ObjectParser.bool(True, strict=True), True) |
| self.assertIs(ObjectParser.bool(False, strict=True), False) |
| |
| def test_parse_bool_invalid(self): |
| invalid: Any |
| for invalid in (1, 0, "1", "0", "", None, [], tuple()): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.bool(invalid) |
| ObjectParser.bool(invalid, strict=True) |
| |
| def test_parse_bool_invalid_strict(self): |
| invalid: Any |
| for invalid in (None, "False", "false", "True", "true"): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.bool(invalid, strict=True) |
| |
| def test_parse_optional_bool(self): |
| self.assertIsNone(ObjectParser.optional_bool(None)) |
| self.assertIs(ObjectParser.optional_bool("true"), True) |
| self.assertIs(ObjectParser.optional_bool("false"), False) |
| |
| def test_parse_optional_bool_invalid(self): |
| invalid: Any |
| for invalid in (1, 0, "1", "0", "", [], tuple()): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.optional_bool(invalid) |
| ObjectParser.optional_bool(invalid, strict=True) |
| |
| def test_parse_optional_bool_invalid_strict(self): |
| invalid: Any |
| for invalid in ("False", "false", "True", "true"): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.optional_bool(invalid, strict=True) |
| |
| def test_parse_sh_cmd(self): |
| self.assertListEqual(ObjectParser.sh_cmd("ls -al ."), ["ls", "-al", "."]) |
| self.assertListEqual(ObjectParser.sh_cmd("ls -al '.'"), ["ls", "-al", "."]) |
| self.assertListEqual( |
| ObjectParser.sh_cmd(";ls -al '.'"), [";ls", "-al", "."]) |
| self.assertListEqual( |
| ObjectParser.sh_cmd(("ls", "-al", ".")), ["ls", "-al", "."]) |
| |
| def test_parse_sh_cmd_invalid(self): |
| invalid: Any |
| for invalid in (1, "", None, [], "ls -al \"."): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.sh_cmd(invalid) |
| |
| def test_parse_dict_invalid(self): |
| invalid: Any |
| for invalid in (1, 0, "1", "0", "", None, [], tuple()): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.dict(invalid) |
| |
| def test_parse_dict(self): |
| self.assertDictEqual(ObjectParser.dict({}), {}) |
| self.assertDictEqual(ObjectParser.dict({"A": 2}), {"A": 2}) |
| |
| def test_parse_non_empty_dict_invalid(self): |
| invalid: Any |
| for invalid in (1, 0, "1", "0", "", None, [], tuple(), {}): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.non_empty_dict(invalid) |
| |
| def test_parse_non_empty_dict(self): |
| result = ObjectParser.non_empty_dict({"a": 1}) |
| self.assertDictEqual(result, {"a": 1}) |
| |
| def test_parse_unique_sequence(self): |
| self.assertListEqual(ObjectParser.unique_sequence([]), []) |
| self.assertTupleEqual(ObjectParser.unique_sequence(tuple()), tuple()) |
| self.assertListEqual(ObjectParser.unique_sequence([1, 2, 3]), [1, 2, 3]) |
| self.assertTupleEqual(ObjectParser.unique_sequence((1, 2, 3)), (1, 2, 3)) |
| |
| def test_parse_unique_sequence_invalid(self): |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| ObjectParser.unique_sequence([1, 1, 2, 2, 2, 3, 5, 5]) |
| self.assertIn("duplicates", str(cm.exception)) |
| self.assertIn("1, 2, 5", str(cm.exception)) |
| |
| def test_parse_unique_sequence_custom_exception(self): |
| |
| class CustomException(Exception): |
| pass |
| |
| with self.assertRaises(CustomException): |
| ObjectParser.unique_sequence([1, 1], error_cls=CustomException) |
| |
| def test_parse_unique_sequence_custom_name(self): |
| with self.assertRaises(argparse.ArgumentTypeError) as cm: |
| ObjectParser.unique_sequence([1, 1], name="custom test name") |
| self.assertIn("custom test name", str(cm.exception)) |
| |
| def test_parse_sequence(self): |
| self.assertListEqual(ObjectParser.sequence([]), []) |
| self.assertListEqual(ObjectParser.sequence([1, 2]), [1, 2]) |
| self.assertTupleEqual(ObjectParser.sequence(tuple()), tuple()) |
| self.assertTupleEqual(ObjectParser.sequence((1, 2)), (1, 2)) |
| |
| def test_parse_sequence_invalid(self): |
| invalid: Any |
| for invalid in ("", "1", 1, {}, {"a": 1}, set(), set((1, 2))): |
| with self.subTest(invalid=invalid): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.sequence(invalid) |
| |
| def test_parse_non_empty_sequence(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.non_empty_sequence([]) |
| self.assertListEqual(ObjectParser.non_empty_sequence([1, 2]), [1, 2]) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.non_empty_sequence(tuple()) |
| self.assertTupleEqual(ObjectParser.non_empty_sequence((1, 2)), (1, 2)) |
| |
| def test_parse_non_empty_sequence_invalid(self): |
| invalid: Any |
| for invalid in ("", "1", 1, {}, {"a": 1}, set(), set((1, 2)), (), []): |
| with self.subTest(invalid=invalid): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.non_empty_sequence(invalid) |
| |
| def test_parse_fuzzy_url_invalid(self): |
| invalid = ("foo", "x64/123", "x64/www.google.com") |
| for url in invalid: |
| with self.subTest(url=url): |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, url): |
| ObjectParser.fuzzy_url(url) |
| |
| def test_parse_fuzzy_url(self): |
| expected = ( |
| ("/foo/bar", "file:///foo/bar"), |
| ("C:/foo/bar", "file://C:/foo/bar"), |
| ("1234.com", "https://1234.com"), |
| ("http://1234.com", "http://1234.com"), |
| ("http://127.0.0.1", "http://127.0.0.1"), |
| ("http://localhost", "http://localhost"), |
| ("127.0.0.1", "https://127.0.0.1"), |
| ("localhost", "https://localhost"), |
| ("test.com", "https://test.com"), |
| ("test.com/", "https://test.com/"), |
| ("test.com/1234", "https://test.com/1234"), |
| ("test.com/bar", "https://test.com/bar"), |
| ("test.com/bar?x=1", "https://test.com/bar?x=1"), |
| ("test.com:1234", "https://test.com:1234"), |
| ("test.com:1234/", "https://test.com:1234/"), |
| ("test.com:1234/56", "https://test.com:1234/56"), |
| ("test.com:1234/56/", "https://test.com:1234/56/"), |
| ("test.com:1234/bar", "https://test.com:1234/bar"), |
| ("test.com:1234/bar?x=1", "https://test.com:1234/bar?x=1"), |
| ("localhost:8123", "https://localhost:8123"), |
| ("localhost:8123", "https://localhost:8123"), |
| ("localhost:8123/", "https://localhost:8123/"), |
| ("localhost:8123/77", "https://localhost:8123/77"), |
| ("localhost:8123/77/", "https://localhost:8123/77/"), |
| ("localhost:8123/bar", "https://localhost:8123/bar"), |
| ("localhost:8123/bar?x=1", "https://localhost:8123/bar?x=1"), |
| ("data:text/html,this is some data", |
| "data:text/html,this is some data"), |
| ("chrome://extensions", "chrome://extensions"), |
| ) |
| for url, result in expected: |
| with self.subTest(url=url): |
| self.assertEqual(ObjectParser.fuzzy_url_str(url), result) |
| parsed = ObjectParser.fuzzy_url(url) |
| self.assertEqual(urlparse.urlunparse(parsed), result) |
| |
| def test_parse_fuzzy_url_default_scheme(self): |
| expected = ("test.com", "test.com/", "test.com/bar", "test.com/bar?x=1", |
| "test.com:1234", "test.com:1234/", "test.com:1234/bar", |
| "test.com:1234/bar?x=1", "localhost:8123", "localhost:8123/", |
| "localhost:8123/bar", "localhost:8123/bar?x1") |
| for url in expected: |
| with self.subTest(url=url): |
| result_default = f"https://{url}" |
| self.assertEqual(ObjectParser.fuzzy_url_str(url), result_default) |
| parsed = ObjectParser.fuzzy_url(url) |
| self.assertEqual(urlparse.urlunparse(parsed), result_default) |
| result_custom = f"ftp://{url}" |
| self.assertEqual( |
| ObjectParser.fuzzy_url_str(url, default_scheme="ftp"), |
| result_custom) |
| parsed = ObjectParser.fuzzy_url(url, default_scheme="ftp") |
| self.assertEqual(urlparse.urlunparse(parsed), result_custom) |
| |
| def test_parse_url(self): |
| expected = ( |
| ("file:///foo/bar", "file:///foo/bar"), |
| ("about:blank", "about:blank"), |
| ("http://test.com/bar", "http://test.com/bar"), |
| ("https://test.com/bar", "https://test.com/bar"), |
| ("http://test.com", "http://test.com"), |
| ("https://test.com/", "https://test.com/"), |
| ("http://test.com/bar", "http://test.com/bar"), |
| ("https://test.com/bar?x=1", "https://test.com/bar?x=1"), |
| ("http://test.com:1234", "http://test.com:1234"), |
| ("https://test.com:1234/", "https://test.com:1234/"), |
| ("http://test.com:1234/bar", "http://test.com:1234/bar"), |
| ("https://test.com:1234/bar?x=1", "https://test.com:1234/bar?x=1"), |
| ("http://localhost:8123", "http://localhost:8123"), |
| ("https://localhost:8123/", "https://localhost:8123/"), |
| ("http://localhost:8123/bar", "http://localhost:8123/bar"), |
| ("https://localhost:8123/bar?x=1", "https://localhost:8123/bar?x=1"), |
| ("data:text/html,this is some data", |
| "data:text/html,this is some data"), |
| ) |
| for url, result in expected: |
| with self.subTest(url=url): |
| self.assertEqual(ObjectParser.url_str(url), result) |
| self.assertEqual(ObjectParser.fuzzy_url_str(url), result) |
| parsed = ObjectParser.url(url) |
| self.assertEqual(urlparse.urlunparse(parsed), result) |
| parsed_fuzzy = ObjectParser.fuzzy_url(url) |
| self.assertEqual(urlparse.urlunparse(parsed_fuzzy), result) |
| |
| def test_parse_url_invalid(self): |
| invalid: Any |
| for invalid in (None, "", {}, "http:// foo .com/bar", "htt p://foo.com", |
| "http://foo.com:-123/bar"): |
| with self.subTest(invalid=invalid): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.url(invalid) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.url_str(invalid) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.httpx_url_str(invalid) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.fuzzy_url_str(invalid) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.fuzzy_url(invalid) |
| |
| def test_parse_httpx_url_str_invalid(self): |
| invalid: Any |
| for invalid in ("ftp://foo.com:123/bar", "ssh://test.com"): |
| with self.subTest(invalid=invalid): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.httpx_url_str(invalid) |
| |
| def test_parse_url_scheme(self): |
| url = "ftp://foo.com" |
| parsed = ObjectParser.url(url) |
| self.assertEqual(urlparse.urlunparse(parsed), url) |
| with self.assertRaises(argparse.ArgumentTypeError): |
| _ = ObjectParser.url(url, schemes=("https",)) |
| parsed = ObjectParser.url( |
| url, schemes=( |
| "https", |
| "ftp", |
| )) |
| self.assertEqual(urlparse.urlunparse(parsed), url) |
| |
| def test_parse_regexp(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.regexp("\\") |
| pattern = ObjectParser.regexp("^abc$") |
| self.assertEqual(pattern.pattern, "^abc$") |
| |
| def test_bytes_or_file_contents_invalid(self): |
| with self.assertRaises(argparse.ArgumentTypeError): |
| ObjectParser.bytes_or_file_contents(None) |
| |
| def test_bytes_or_file_contents(self): |
| result: bytes = ObjectParser.bytes_or_file_contents("some data") |
| self.assertEqual(result, b"some data") |
| |
| file = pathlib.Path("file") |
| with file.open("w", encoding="utf-8") as f: |
| f.write("some data") |
| result = ObjectParser.bytes_or_file_contents(file) |
| self.assertEqual(result, b"some data") |
| |
| def test_proto_or_file_textproto(self): |
| text_proto_file = pathlib.Path("trace_config.textproto") |
| text_config: str = """ |
| buffers: { |
| size_kb: 123456 |
| fill_policy: DISCARD |
| } |
| """ |
| parser = ObjectParser.proto_or_file(trace_config_pb2.TraceConfig) |
| proto_instance: trace_config_pb2.TraceConfig = parser(text_config) |
| self.assertEqual(len(proto_instance.buffers), 1) |
| self.assertEqual(proto_instance.buffers[0].size_kb, 123456) |
| |
| with text_proto_file.open("w", encoding="utf-8") as f: |
| f.write(text_config) |
| proto_instance_2 = parser(text_proto_file) |
| self.assertEqual(len(proto_instance_2.buffers), 1) |
| self.assertEqual(proto_instance_2.buffers[0].size_kb, 123456) |
| self.assertEqual(proto_instance, proto_instance_2) |
| |
| def test_proto_or_file_invalid(self): |
| text_proto_file = pathlib.Path("trace_config.textproto") |
| text_config: str = """ |
| buffers-invalid: { |
| size_kb: 123456 |
| fill_policy: DISCARD |
| } |
| """ |
| parser = ObjectParser.proto_or_file(trace_config_pb2.TraceConfig) |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, "buffers-invalid"): |
| parser(text_config) |
| |
| with text_proto_file.open("w", encoding="utf-8") as f: |
| f.write(text_config) |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, "buffers-invalid"): |
| parser(text_proto_file) |
| |
| def test_proto_or_file_binary(self): |
| text_proto_file = pathlib.Path("trace_config.textproto") |
| binary_proto_file = pathlib.Path("trace_config.proto") |
| text_config: str = """ |
| buffers: { |
| size_kb: 123456 |
| fill_policy: DISCARD |
| } |
| """ |
| with text_proto_file.open("w", encoding="utf-8") as f: |
| f.write(text_config) |
| parser = ObjectParser.proto_or_file(trace_config_pb2.TraceConfig) |
| proto_instance = parser(text_proto_file) |
| with binary_proto_file.open("wb") as f: |
| f.write(proto_instance.SerializeToString()) |
| |
| proto_instance_2 = parser(binary_proto_file) |
| self.assertEqual(len(proto_instance_2.buffers), 1) |
| self.assertEqual(proto_instance_2.buffers[0].size_kb, 123456) |
| self.assertEqual(proto_instance, proto_instance_2) |
| |
| def test_proto_or_file_binary_invalid_format(self): |
| binary_proto_file = pathlib.Path("trace_config.proto") |
| self.fs.create_file(binary_proto_file, contents="invalid data") |
| parser = ObjectParser.proto_or_file(trace_config_pb2.TraceConfig) |
| with self.assertRaisesRegex(argparse.ArgumentTypeError, "TraceConfig"): |
| parser(binary_proto_file) |
| |
| |
| class TimeUnitTestCase(unittest.TestCase): |
| |
| def test_parse_microseconds(self): |
| for unit in ("us", "micros", "microseconds"): |
| self.assertIs(TimeUnit.parse(unit), TimeUnit.MICROSECOND) |
| |
| def test_parse_milliseconds(self): |
| for unit in ("ms", "millis", "milliseconds"): |
| self.assertIs(TimeUnit.parse(unit), TimeUnit.MILLISECOND) |
| |
| def test_parse_seconds(self): |
| for unit in ("s", "sec", "secs", "second", "seconds"): |
| self.assertIs(TimeUnit.parse(unit), TimeUnit.SECOND) |
| |
| def test_parse_minutes(self): |
| for unit in ("m", "min", "mins", "minute", "minutes"): |
| self.assertIs(TimeUnit.parse(unit), TimeUnit.MINUTE) |
| |
| def test_parse_hours(self): |
| for unit in ("h", "hrs", "hour", "hours"): |
| self.assertIs(TimeUnit.parse(unit), TimeUnit.HOUR) |
| |
| def test_parse_days(self): |
| for unit in ("d", "day", "days"): |
| self.assertIs(TimeUnit.parse(unit), TimeUnit.DAY) |
| |
| def test_parse_weeks(self): |
| for unit in ("w", "week", "weeks"): |
| self.assertIs(TimeUnit.parse(unit), TimeUnit.WEEK) |
| |
| def test_parse_invalid(self): |
| invalid: Any |
| for invalid in ("months", "yy", "ww", "i"): |
| with self.assertRaises(DurationParseError): |
| TimeUnit.parse(invalid) |
| |
| def test_to_timedelta(self): |
| self.assertEqual( |
| TimeUnit.MICROSECOND.timedelta(123), dt.timedelta(microseconds=123)) |
| self.assertEqual( |
| TimeUnit.MILLISECOND.timedelta(123), dt.timedelta(milliseconds=123)) |
| self.assertEqual(TimeUnit.SECOND.timedelta(123), dt.timedelta(seconds=123)) |
| self.assertEqual(TimeUnit.MINUTE.timedelta(123), dt.timedelta(minutes=123)) |
| self.assertEqual(TimeUnit.HOUR.timedelta(123), dt.timedelta(hours=123)) |
| self.assertEqual(TimeUnit.DAY.timedelta(123), dt.timedelta(days=123)) |
| self.assertEqual(TimeUnit.WEEK.timedelta(123), dt.timedelta(weeks=123)) |
| |
| |
| if __name__ == "__main__": |
| test_helper.run_pytest(__file__) |
