#!/usr/bin/python3 -OO # Copyright 2007-2021 The SABnzbd-Team # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. """ tests.test_misc - Testing functions in misc.py """ import datetime import subprocess import sys import tempfile from unittest import mock from sabnzbd import lang from sabnzbd import misc from sabnzbd import newsunpack from sabnzbd.config import ConfigCat from sabnzbd.constants import HIGH_PRIORITY, FORCE_PRIORITY, DEFAULT_PRIORITY, NORMAL_PRIORITY from tests.testhelper import * class TestMisc: @staticmethod def assertTime(offset, age): assert offset == misc.calc_age(age, trans=True) assert offset == misc.calc_age(age, trans=False) def test_timeformat24h(self): assert "%H:%M:%S" == misc.time_format("%H:%M:%S") assert "%H:%M" == misc.time_format("%H:%M") @set_config({"ampm": True}) def test_timeformatampm(self): misc.HAVE_AMPM = True assert "%I:%M:%S %p" == misc.time_format("%H:%M:%S") assert "%I:%M %p" == misc.time_format("%H:%M") def test_calc_age(self): date = datetime.datetime.now() m = date - datetime.timedelta(minutes=1) h = date - datetime.timedelta(hours=1) d = date - datetime.timedelta(days=1) self.assertTime("1m", m) self.assertTime("1h", h) self.assertTime("1d", d) def test_safe_lower(self): assert "all caps" == misc.safe_lower("ALL CAPS") assert "" == misc.safe_lower(None) def test_cmp(self): assert misc.cmp(1, 2) < 0 assert misc.cmp(2, 1) > 0 assert misc.cmp(1, 1) == 0 def test_cat_to_opts(self): # Need to create the Default category, as we would in normal instance # Otherwise it will try to save the config ConfigCat("*", {"pp": 3, "script": "None", "priority": NORMAL_PRIORITY}) assert ("*", 3, "None", NORMAL_PRIORITY) == misc.cat_to_opts("*") assert ("*", 3, "None", NORMAL_PRIORITY) == misc.cat_to_opts("Nonsense") assert ("*", 1, "None", NORMAL_PRIORITY) == misc.cat_to_opts("*", pp=1) assert ("*", 1, "test.py", NORMAL_PRIORITY) == misc.cat_to_opts("*", pp=1, script="test.py") ConfigCat("movies", {"priority": HIGH_PRIORITY, "script": "test.py"}) assert ("movies", 3, "test.py", HIGH_PRIORITY) == misc.cat_to_opts("movies") assert ("movies", 1, "test.py", HIGH_PRIORITY) == misc.cat_to_opts("movies", pp=1) assert ("movies", 1, "not_test.py", HIGH_PRIORITY) == misc.cat_to_opts("movies", pp=1, script="not_test.py") assert ("movies", 3, "test.py", FORCE_PRIORITY) == misc.cat_to_opts("movies", priority=FORCE_PRIORITY) # If the category has DEFAULT_PRIORITY, it should use the priority of the *-category (NORMAL_PRIORITY) # If the script-name is Default for a category, it should use the script of the *-category (None) ConfigCat("software", {"priority": DEFAULT_PRIORITY, "script": "Default"}) assert ("software", 3, "None", NORMAL_PRIORITY) == misc.cat_to_opts("software") assert ("software", 3, "None", FORCE_PRIORITY) == misc.cat_to_opts("software", priority=FORCE_PRIORITY) def test_wildcard_to_re(self): assert "\\\\\\^\\$\\.\\[" == misc.wildcard_to_re("\\^$.[") assert "\\]\\(\\)\\+.\\|\\{\\}.*" == misc.wildcard_to_re("]()+?|{}*") def test_cat_convert(self): # TODO: Make test pass def test_convert_version(self): assert (3010099, False) == misc.convert_version("3.1.0") assert (3010099, False) == misc.convert_version("3.1.0BlaBla") assert (3010001, True) == misc.convert_version("3.1.0Alpha1") assert (3010041, True) == misc.convert_version("3.1.0Beta1") assert (3010081, True) == misc.convert_version("3.1.0RC1") assert (3010194, True) == misc.convert_version("3.1.1RC14") def test_from_units(self): assert -1.0 == misc.from_units("-1") assert 100.0 == misc.from_units("100") assert 1024.0 == misc.from_units("1KB") assert 1048576.0 == misc.from_units("1024KB") assert 1048576.0 == misc.from_units("1024Kb") assert 1048576.0 == misc.from_units("1024kB") assert 1048576.0 == misc.from_units("1MB") assert 1073741824.0 == misc.from_units("1GB") assert 1125899906842624.0 == misc.from_units("1P") def test_to_units(self): assert "1 K" == misc.to_units(1024) assert "1 KBla" == misc.to_units(1024, postfix="Bla") assert "1.0 M" == misc.to_units(1024 * 1024) assert "1.0 M" == misc.to_units(1024 * 1024 + 10) assert "10.0 M" == misc.to_units(1024 * 1024 * 10) assert "100.0 M" == misc.to_units(1024 * 1024 * 100) assert "9.8 G" == misc.to_units(1024 * 1024 * 10000) assert "1024.0 P" == misc.to_units(1024 ** 6) def test_unit_back_and_forth(self): assert 100 == misc.from_units(misc.to_units(100)) assert 1024 == misc.from_units(misc.to_units(1024)) assert 1024 ** 3 == misc.from_units(misc.to_units(1024 ** 3)) def test_caller_name(self): @set_config({"log_level": 0}) def test_wrapper(skip): return misc.caller_name(skip=skip) @set_config({"log_level": 2}) def test_wrapper_2(skip): return misc.caller_name(skip=skip) # No logging on lower-level assert "N/A" == test_wrapper(1) assert "N/A" == test_wrapper(2) assert "N/A" == test_wrapper(3) # Wrappers originate from the set_config-wrapper assert "test_wrapper_2" in test_wrapper_2(1) assert "wrapper_func" in test_wrapper_2(2) def test_split_host(self): assert (None, None) == misc.split_host(None) assert (None, None) == misc.split_host("") assert ("sabnzbd.org", 123) == misc.split_host("sabnzbd.org:123") assert ("sabnzbd.org", None) == misc.split_host("sabnzbd.org") assert ("127.0.0.1", 566) == misc.split_host("127.0.0.1:566") assert ("[::1]", 1234) == misc.split_host("[::1]:1234") assert ("[2001:db8::8080]", None) == misc.split_host("[2001:db8::8080]") @set_config({"cleanup_list": [".exe", ".nzb"]}) def test_on_cleanup_list(self): assert misc.on_cleanup_list("test.exe") assert misc.on_cleanup_list("TEST.EXE") assert misc.on_cleanup_list("longExeFIlanam.EXe") assert not misc.on_cleanup_list("testexe") assert misc.on_cleanup_list("test.nzb") assert not misc.on_cleanup_list("test.nzb", skip_nzb=True) assert not misc.on_cleanup_list("test.exe.lnk") def test_format_time_string(self): assert "0 seconds" == misc.format_time_string(None) assert "0 seconds" == misc.format_time_string("Test") assert "0 seconds" == misc.format_time_string(0) assert "1 sec" == misc.format_time_string(1) assert "10 seconds" == misc.format_time_string(10) assert "1 min" == misc.format_time_string(60) assert "1 hour 1 min 1 sec" == misc.format_time_string(60 * 60 + 60 + 1) assert "1 day 59 seconds" == misc.format_time_string(86400 + 59) assert "2 days 2 hours 2 seconds" == misc.format_time_string(2 * 86400 + 2 * 60 * 60 + 2) def test_format_time_string_locale(self): # Have to set the languages, if it was compiled locale_dir = os.path.join(SAB_BASE_DIR, "..", sabnzbd.constants.DEF_LANGUAGE) if not os.path.exists(locale_dir): pytest.mark.skip("No language files compiled") lang.set_locale_info("SABnzbd", locale_dir) lang.set_language("de") assert "1 Sekunde" == misc.format_time_string(1) assert "10 Sekunden" == misc.format_time_string(10) assert "1 Minuten" == misc.format_time_string(60) assert "1 Stunde 1 Minuten 1 Sekunde" == misc.format_time_string(60 * 60 + 60 + 1) assert "1 Tag 59 Sekunden" == misc.format_time_string(86400 + 59) assert "2 Tage 2 Stunden 2 Sekunden" == misc.format_time_string(2 * 86400 + 2 * 60 * 60 + 2) def test_int_conv(self): assert 0 == misc.int_conv("0") assert 10 == misc.int_conv("10") assert 10 == misc.int_conv(10) assert 10 == misc.int_conv(10.0) assert 0 == misc.int_conv(None) assert 1 == misc.int_conv(True) assert 0 == misc.int_conv(object) def test_create_https_certificates(self): cert_file = "test.cert" key_file = "test.key" assert misc.create_https_certificates(cert_file, key_file) assert os.path.exists(cert_file) assert os.path.exists(key_file) # Remove files os.unlink("test.cert") os.unlink("test.key") @pytest.mark.parametrize( "name, result", [ ("Free.Open.Source.Movie.2001.1080p.WEB-DL.DD5.1.H264-FOSS", False), # Not samples ("Setup.exe", False), ("23.123.hdtv-rofl", False), ("Something.1080p.WEB-DL.DD5.1.H264-EMRG-sample", True), # Samples ("Something.1080p.WEB-DL.DD5.1.H264-EMRG-sample.ogg", True), ("Sumtin_Else_1080p_WEB-DL_DD5.1_H264_proof-EMRG", True), ("Wot.Eva.540i.WEB-DL.aac.H264-Groupie sample.mp4", True), ("file-sample.mkv", True), ("PROOF.JPG", True), ("Bla.s01e02.title.1080p.aac-sample proof.mkv", True), ("Bla.s01e02.title.1080p.aac-proof.mkv", True), ("Bla.s01e02.title.1080p.aac sample proof.mkv", True), ("Bla.s01e02.title.1080p.aac proof.mkv", True), ("Lwtn.s08e26.1080p.web.h264-glhf-sample.par2", True), ("Lwtn.s08e26.1080p.web.h264-glhf-sample.vol001-002.par2", True), ("Look at That 2011 540i WEB-DL.H265-NoSample", False), ], ) def test_is_sample(self, name, result): assert misc.is_sample(name) == result @pytest.mark.parametrize( "name, result", [ ("Not Death Proof (2022) 1080p x264 (DD5.1) BE Subs", False), # Try to trigger some false positives ("Proof.of.Everything.(2042).4320p.x266-4U", False), ("Crime_Scene_S01E13_Free_Sample_For_Sale_480p-OhDear", False), ("Sample That 2011 480p WEB-DL.H265-aMiGo", False), ("NOT A SAMPLE.JPG", False), ], ) def test_is_sample_known_false_positives(self, name, result): """We know these fail, but don't have a better solution for them at the moment.""" assert misc.is_sample(name) != result @pytest.mark.parametrize( "test_input, expected_output", [ (["cmd1", 9, "cmd3"], '"cmd1" "9" "cmd3"'), # sending all commands as valid string (["", "cmd1", "5"], '"" "cmd1" "5"'), # sending blank string (["cmd1", None, "cmd3", "tail -f"], '"cmd1" "" "cmd3" "tail -f"'), # sending None in command (["cmd1", 0, "ps ux"], '"cmd1" "" "ps ux"'), # sending 0 ], ) def test_list_to_cmd(self, test_input, expected_output): """Test to convert list to a cmd.exe-compatible command string""" res = misc.list2cmdline(test_input) # Make sure the output is cmd.exe-compatible assert res == expected_output @pytest.mark.parametrize( "value, result", [ ("1.2.3.4", True), ("255.255.255.255", True), ("0.0.0.0", True), ("10.11.12.13", True), ("127.0.0.1", True), ("400.500.600.700", False), ("blabla", False), ("2001::1", False), ("::1", False), ("::", False), ("example.org", False), (None, False), ("", False), ("3.2.0", False), (-42, False), ("::ffff:192.168.1.100", False), ], ) def test_is_ipv4_addr(self, value, result): assert misc.is_ipv4_addr(value) is result @pytest.mark.parametrize( "value, result", [ ("2001::1", True), ("::1", True), ("[2001::1]", True), ("fdd6:5a2d:3f20:0:14b0:d8f4:ccb9:fab6", True), ("::", True), ("a::b", True), ("1.2.3.4", False), ("255.255.255.255", False), ("0.0.0.0", False), ("10.11.12.13", False), ("127.0.0.1", False), ("400.500.600.700", False), ("blabla", False), (666, False), ("example.org", False), (None, False), ("", False), ("[1.2.3.4]", False), ("2001:1", False), ("2001::[2001::1]", False), ("::ffff:192.168.1.100", True), ], ) def test_is_ipv6_addr(self, value, result): assert misc.is_ipv6_addr(value) is result @pytest.mark.parametrize( "value, result", [ ("::1", True), ("[::1]", True), ("127.0.0.1", True), ("127.255.0.0", True), ("127.1.2.7", True), ("fdd6:5a2d:3f20:0:14b0:d8f4:ccb9:fab6", False), ("::", False), ("a::b", False), ("1.2.3.4", False), ("255.255.255.255", False), ("0.0.0.0", False), ("10.11.12.13", False), ("400.500.600.700", False), ("localhost", False), (-666, False), ("example.org", False), (None, False), ("", False), ("[127.6.6.6]", False), ("2001:1", False), ("2001::[2001::1]", False), ("::ffff:192.168.1.100", False), ("::ffff:1.1.1.1", False), ("::ffff:127.0.0.1", True), ], ) def test_is_loopback_addr(self, value, result): assert misc.is_loopback_addr(value) is result @pytest.mark.parametrize( "value, result", [ ("localhost", True), ("::1", True), ("[::1]", True), ("localhost", True), ("127.0.0.1", True), ("127.255.0.0", True), ("127.1.2.7", True), (".local", False), ("test.local", False), ("fdd6:5a2d:3f20:0:14b0:d8f4:ccb9:fab6", False), ("::", False), ("a::b", False), ("1.2.3.4", False), ("255.255.255.255", False), ("0.0.0.0", False), ("10.11.12.13", False), ("400.500.600.700", False), (-1984, False), ("example.org", False), (None, False), ("", False), ("[127.6.6.6]", False), ("2001:1", False), ("2001::[2001::1]", False), ("::ffff:192.168.1.100", False), ("::ffff:1.1.1.1", False), ("::ffff:127.0.0.1", True), ], ) def test_is_localhost(self, value, result): assert misc.is_localhost(value) is result @pytest.mark.parametrize( "value, result", [ ("10.11.12.13", True), ("172.16.2.81", True), ("192.168.255.255", True), ("169.254.42.42", True), # Link-local ("fd00::ffff", True), # Part of fc00::/7, IPv6 "Unique Local Addresses" ("fe80::a1", True), # IPv6 Link-local ("::1", False), ("localhost", False), ("127.0.0.1", False), ("2001:1337:babe::", False), ("172.32.32.32", False), # Near but not part of 172.16.0.0/12 ("100.64.0.1", False), # Test net ("[2001::1]", False), ("::", False), ("::a:b:c", False), ("1.2.3.4", False), ("255.255.255.255", False), ("0.0.0.0", False), ("127.0.0.1", False), ("400.500.600.700", False), ("blabla", False), (-666, False), ("example.org", False), (None, False), ("", False), ("[1.2.3.4]", False), ("2001:1", False), ("2001::[2001::1]", False), ("::ffff:192.168.1.100", True), ("::ffff:1.1.1.1", False), ("::ffff:127.0.0.1", False), ], ) def test_is_lan_addr(self, value, result): assert misc.is_lan_addr(value) is result @pytest.mark.parametrize( "ip, subnet, result", [ ("2001:c0f:fee::1", "2001:c0f:fee", True), # Old-style range setting ("2001:c0f:fee::1", "2001:c0f:FEE:", True), ("2001:c0f:fee::1", "2001:c0FF:ffee", False), ("2001:c0f:fee::1", "2001:c0ff:ffee:", False), ("2001:C0F:FEE::1", "2001:c0f:fee::/48", True), ("2001:c0f:fee::1", "2001:c0f:fee::/112", True), ("2001:c0f:fee::1", "::/0", True), # Subnet equals the entire IPv6 address space ("2001:c0f:fee::1", "2001:c0:ffee::/48", False), ("2001:c0f:fee::1", "2001:c0ff:ee::/112", False), ("2001:c0f:fEE::1", "2001:c0f:fee:eeee::/48", False), # Invalid subnet ("2001:c0f:Fee::1", "2001:c0f:fee:/64", False), ("2001:c0f:fee::1", "2001:c0f:fee:eeee:3:2:1:0/112", False), ("2001:c0f:fee::1", "2001:c0f:fee::1", True), # Single-IP subnet ("2001:c0f:fee::1", "2001:c0f:fee::1/128", True), ("2001:c0f:fee::1", "2001:c0f:fee::2", False), ("2001:c0f:fee::1", "2001:c0f:fee::2/128", False), ("::1", "::/127", True), ("::1", "2021::/64", False), # Localhost not in subnet ("192.168.43.21", "192.168.43", True), # Old-style subnet setting ("192.168.43.21", "192.168.43.", True), ("192.168.43.21", "192.168.4", False), ("192.168.43.21", "192.168.4.", False), ("10.11.12.13", "10", True), # Bad old-style setting (allowed 100.0.0.0/6, 104.0.0.0/6 and 108.0.0.0/7) ("10.11.12.13", "10.", True), # Correct version of the same (10.0.0.0/8 only) ("108.1.2.3", "10", False), # This used to be allowed with the bad setting! ("108.1.2.3", "10.", False), ("192.168.43.21", "192.168.0.0/16", True), ("192.168.43.21", "192.168.0.0/255.255.255.0", True), ("::ffff:192.168.43.21", "192.168.43.0/24", True), # IPv4-mapped IPv6 ("dual-stack") notation ("::FFff:192.168.43.21", "192.168.43.0/24", True), ("::ffff:192.168.12.34", "192.168.43.0/24", False), ("::ffFF:192.168.12.34", "192.168.43.0/24", False), ("192.168.43.21", "192.168.43.0/26", True), ("200.100.50.25", "0.0.0.0/0", True), # Subnet equals the entire IPv4 address space ("192.168.43.21", "10.0.0.0/8", False), ("192.168.43.21", "192.168.1.0/22", False), ("192.168.43.21", "192.168.43.21/24", False), # Invalid subnet ("192.168.43.21", "192.168.43/24", False), ("192.168.43.21", "192.168.43.0/16", False), ("192.168.43.21", "192.168.43.0/255.252.0.0", False), ("192.168.43.21", "192.168.43.21", True), # Single-IP subnet ("192.168.43.21", "192.168.43.21/32", True), ("192.168.43.21", "192.168.43.21/255.255.255.255", True), ("192.168.43.21", "192.168.43.12", False), ("192.168.43.21", "192.168.43.0/32", False), ("192.168.43.21", "43.21.168.192/255.255.255.255", False), ("127.0.0.1", "127.0.0.0/31", True), ("127.0.1.1", "127.0.0.0/24", False), # Localhost not in subnet ("111.222.33.44", "111:222:33::/96", False), # IPv4/IPv6 mixup ("111:222:33::44", "111.222.0.0/24", False), ("aaaa::1:2:3:4", "f:g:h:i:43:21::/112", False), # Invalid subnet ("4.3.2.1", "654.3.2.1.0/24", False), (None, "1.2.3.4/32", False), # Missing input ("1:a:2:b::", None, False), (None, None, False), ], ) def test_ip_in_subnet(self, ip, subnet, result): misc.ip_in_subnet(ip, subnet) is result @pytest.mark.parametrize( "ip, result", [ ("::ffff:127.0.0.1", "127.0.0.1"), ("::FFFF:127.0.0.1", "127.0.0.1"), ("::ffff:192.168.1.255", "192.168.1.255"), ("::ffff:8.8.8.8", "8.8.8.8"), ("2007::2021", "2007::2021"), ("::ffff:2007:2021", "::ffff:2007:2021"), ("2007::ffff:2021", "2007::ffff:2021"), ("12.34.56.78", "12.34.56.78"), ("foobar", "foobar"), ("0:0:0:0:0:ffff:8.8.4.4", "8.8.4.4"), ("0000:0000:0000:0000:0000:ffff:1.0.0.1", "1.0.0.1"), ("0000::0:ffff:1.1.1.1", "1.1.1.1"), ], ) def test_strip_ipv4_mapped_notation(self, ip, result): misc.strip_ipv4_mapped_notation(ip) == result class TestBuildAndRunCommand: # Path should exist script_path = os.path.join(SAB_BASE_DIR, "test_misc.py") def test_none_check(self): with pytest.raises(IOError): misc.build_and_run_command([None]) @mock.patch("subprocess.Popen") @pytest.mark.skipif(not sys.platform.startswith("win"), reason="Windows tests") def test_win(self, mock_subproc_popen): # Needed for priority and startupinfo check import win32process import win32con misc.build_and_run_command(["test.cmd", "input 1"]) assert mock_subproc_popen.call_args[0][0] == ["test.cmd", "input 1"] assert mock_subproc_popen.call_args[1]["creationflags"] == win32process.NORMAL_PRIORITY_CLASS assert mock_subproc_popen.call_args[1]["startupinfo"].dwFlags == win32process.STARTF_USESHOWWINDOW assert mock_subproc_popen.call_args[1]["startupinfo"].wShowWindow == win32con.SW_HIDE misc.build_and_run_command(["test.py", "input 1"]) assert mock_subproc_popen.call_args[0][0] == ["python.exe", "test.py", "input 1"] assert mock_subproc_popen.call_args[1]["creationflags"] == win32process.NORMAL_PRIORITY_CLASS assert mock_subproc_popen.call_args[1]["startupinfo"].dwFlags == win32process.STARTF_USESHOWWINDOW assert mock_subproc_popen.call_args[1]["startupinfo"].wShowWindow == win32con.SW_HIDE # See: https://github.com/sabnzbd/sabnzbd/issues/1043 misc.build_and_run_command(["UnRar.exe", "\\\\?\\C:\\path\\"]) assert mock_subproc_popen.call_args[0][0] == ["UnRar.exe", "\\\\?\\C:\\path\\"] misc.build_and_run_command(["UnRar.exe", "\\\\?\\C:\\path\\"], flatten_command=True) assert mock_subproc_popen.call_args[0][0] == '"UnRar.exe" "\\\\?\\C:\\path\\"' @mock.patch("sabnzbd.misc.userxbit") @mock.patch("subprocess.Popen") def test_std_override(self, mock_subproc_popen, userxbit): userxbit.return_value = True misc.build_and_run_command([self.script_path], stderr=subprocess.DEVNULL) assert mock_subproc_popen.call_args[1]["stderr"] == subprocess.DEVNULL @set_platform("linux") @set_config({"nice": "--adjustment=-7", "ionice": "-t -n9 -c7"}) @mock.patch("sabnzbd.misc.userxbit") @mock.patch("subprocess.Popen") def test_linux_features(self, mock_subproc_popen, userxbit): # Should break on no-execute permissions userxbit.return_value = False with pytest.raises(IOError): misc.build_and_run_command([self.script_path, "input 1"]) userxbit.return_value = True # Check if python-call is added if not supplied by shebang temp_file_fd, temp_file_path = tempfile.mkstemp(suffix=".py") os.close(temp_file_fd) misc.build_and_run_command([temp_file_path, "input 1"]) assert mock_subproc_popen.call_args[0][0] == [ sys.executable if sys.executable else "python", temp_file_path, "input 1", ] os.remove(temp_file_path) # Have to fake these for it to work newsunpack.IONICE_COMMAND = "ionice" newsunpack.NICE_COMMAND = "nice" userxbit.return_value = True misc.build_and_run_command([self.script_path, "input 1"]) assert mock_subproc_popen.call_args[0][0] == [ "nice", "--adjustment=-7", "ionice", "-t", "-n9", "-c7", self.script_path, "input 1", ]