From 9c3de53a3b0ddccd003bcb4852b76459d6f199fc Mon Sep 17 00:00:00 2001 From: A_D Date: Mon, 23 Nov 2020 08:17:16 +0200 Subject: [PATCH] Added config tests --- tests/_old_config.py | 500 +++++++++++++++++++++++++++++++++++++++++++ tests/test_config.py | 187 ++++++++++++++++ 2 files changed, 687 insertions(+) create mode 100644 tests/_old_config.py create mode 100644 tests/test_config.py diff --git a/tests/_old_config.py b/tests/_old_config.py new file mode 100644 index 00000000..197ca9f8 --- /dev/null +++ b/tests/_old_config.py @@ -0,0 +1,500 @@ +import numbers +import sys +import warnings +from configparser import NoOptionError +from os import getenv, mkdir, pardir, makedirs +from os.path import dirname, expanduser, isdir, join, normpath +from sys import platform +from typing import TYPE_CHECKING, Optional, Union + +from config import applongname, appname, update_interval +from EDMCLogging import get_main_logger + +logger = get_main_logger() + +if platform == 'darwin': + from Foundation import ( # type: ignore + NSApplicationSupportDirectory, NSBundle, NSDocumentDirectory, NSSearchPathForDirectoriesInDomains, + NSUserDefaults, NSUserDomainMask + ) + +elif platform == 'win32': + import ctypes + import uuid + from ctypes.wintypes import DWORD, HANDLE, HKEY, LONG, LPCVOID, LPCWSTR + if TYPE_CHECKING: + import ctypes.windll # type: ignore + + FOLDERID_Documents = uuid.UUID('{FDD39AD0-238F-46AF-ADB4-6C85480369C7}') + FOLDERID_LocalAppData = uuid.UUID('{F1B32785-6FBA-4FCF-9D55-7B8E7F157091}') + FOLDERID_Profile = uuid.UUID('{5E6C858F-0E22-4760-9AFE-EA3317B67173}') + FOLDERID_SavedGames = uuid.UUID('{4C5C32FF-BB9D-43b0-B5B4-2D72E54EAAA4}') + + SHGetKnownFolderPath = ctypes.windll.shell32.SHGetKnownFolderPath + SHGetKnownFolderPath.argtypes = [ctypes.c_char_p, DWORD, HANDLE, ctypes.POINTER(ctypes.c_wchar_p)] + + CoTaskMemFree = ctypes.windll.ole32.CoTaskMemFree + CoTaskMemFree.argtypes = [ctypes.c_void_p] + + # winreg in Python <= 3.7.4 handles REG_MULTI_SZ incorrectly, so do this instead. https://bugs.python.org/issue32587 + HKEY_CURRENT_USER = 0x80000001 + KEY_ALL_ACCESS = 0x000F003F + REG_CREATED_NEW_KEY = 0x00000001 + REG_OPENED_EXISTING_KEY = 0x00000002 + REG_SZ = 1 + REG_DWORD = 4 + REG_MULTI_SZ = 7 + + REG_RESERVED_ALWAYS_ZERO = 0 + + RegCreateKeyEx = ctypes.windll.advapi32.RegCreateKeyExW + RegCreateKeyEx.restype = LONG + RegCreateKeyEx.argtypes = [ + HKEY, LPCWSTR, DWORD, LPCVOID, DWORD, DWORD, LPCVOID, ctypes.POINTER(HKEY), ctypes.POINTER(DWORD) + ] + + RegOpenKeyEx = ctypes.windll.advapi32.RegOpenKeyExW + RegOpenKeyEx.restype = LONG + RegOpenKeyEx.argtypes = [HKEY, LPCWSTR, DWORD, DWORD, ctypes.POINTER(HKEY)] + + RegCloseKey = ctypes.windll.advapi32.RegCloseKey + RegCloseKey.restype = LONG + RegCloseKey.argtypes = [HKEY] + + RegQueryValueEx = ctypes.windll.advapi32.RegQueryValueExW + RegQueryValueEx.restype = LONG + RegQueryValueEx.argtypes = [HKEY, LPCWSTR, LPCVOID, ctypes.POINTER(DWORD), LPCVOID, ctypes.POINTER(DWORD)] + + RegSetValueEx = ctypes.windll.advapi32.RegSetValueExW + RegSetValueEx.restype = LONG + RegSetValueEx.argtypes = [HKEY, LPCWSTR, LPCVOID, DWORD, LPCVOID, DWORD] + + RegCopyTree = ctypes.windll.advapi32.RegCopyTreeW + RegCopyTree.restype = LONG + RegCopyTree.argtypes = [HKEY, LPCWSTR, HKEY] + + RegDeleteKey = ctypes.windll.advapi32.RegDeleteTreeW + RegDeleteKey.restype = LONG + RegDeleteKey.argtypes = [HKEY, LPCWSTR] + + RegDeleteValue = ctypes.windll.advapi32.RegDeleteValueW + RegDeleteValue.restype = LONG + RegDeleteValue.argtypes = [HKEY, LPCWSTR] + + def known_folder_path(guid: uuid.UUID) -> Optional[str]: + """Look up a Windows GUID to actual folder path name.""" + buf = ctypes.c_wchar_p() + if SHGetKnownFolderPath(ctypes.create_string_buffer(guid.bytes_le), 0, 0, ctypes.byref(buf)): + return None + retval = buf.value # copy data + CoTaskMemFree(buf) # and free original + return retval + +elif platform == 'linux': + import codecs + from configparser import RawConfigParser + + +class OldConfig(): + """Object that holds all configuration data.""" + + OUT_MKT_EDDN = 1 + # OUT_MKT_BPC = 2 # No longer supported + OUT_MKT_TD = 4 + OUT_MKT_CSV = 8 + OUT_SHIP = 16 + # OUT_SHIP_EDS = 16 # Replaced by OUT_SHIP + # OUT_SYS_FILE = 32 # No longer supported + # OUT_STAT = 64 # No longer available + # OUT_SHIP_CORIOLIS = 128 # Replaced by OUT_SHIP + OUT_STATION_ANY = OUT_MKT_EDDN | OUT_MKT_TD | OUT_MKT_CSV + # OUT_SYS_EDSM = 256 # Now a plugin + # OUT_SYS_AUTO = 512 # Now always automatic + OUT_MKT_MANUAL = 1024 + OUT_SYS_EDDN = 2048 + OUT_SYS_DELAY = 4096 + + if platform == 'darwin': # noqa: C901 # It's gating *all* the functions + + def __init__(self): + self.app_dir = join( + NSSearchPathForDirectoriesInDomains(NSApplicationSupportDirectory, NSUserDomainMask, True)[0], appname + ) + if not isdir(self.app_dir): + mkdir(self.app_dir) + + self.plugin_dir = join(self.app_dir, 'plugins') + if not isdir(self.plugin_dir): + mkdir(self.plugin_dir) + + if getattr(sys, 'frozen', False): + self.internal_plugin_dir = normpath(join(dirname(sys.executable), pardir, 'Library', 'plugins')) + self.respath = normpath(join(dirname(sys.executable), pardir, 'Resources')) + self.identifier = NSBundle.mainBundle().bundleIdentifier() + + else: + self.internal_plugin_dir = join(dirname(__file__), 'plugins') + self.respath = dirname(__file__) + # Don't use Python's settings if interactive + self.identifier = f'uk.org.marginal.{appname.lower()}' + NSBundle.mainBundle().infoDictionary()['CFBundleIdentifier'] = self.identifier + + self.default_journal_dir = join( + NSSearchPathForDirectoriesInDomains(NSApplicationSupportDirectory, NSUserDomainMask, True)[0], + 'Frontier Developments', + 'Elite Dangerous' + ) + self.home = expanduser('~') + + self.defaults = NSUserDefaults.standardUserDefaults() + self.settings = dict(self.defaults.persistentDomainForName_(self.identifier) or {}) # make writeable + + # Check out_dir exists + if not self.get('outdir') or not isdir(str(self.get('outdir'))): + self.set('outdir', NSSearchPathForDirectoriesInDomains(NSDocumentDirectory, NSUserDomainMask, True)[0]) + + def get(self, key: str, default: Union[None, list, str] = None) -> Union[None, list, str]: + """Look up a string configuration value.""" + val = self.settings.get(key) + if val is None: + return default + + elif isinstance(val, str): + return str(val) + + elif isinstance(val, list): + return list(val) # make writeable + + else: + return default + + def getint(self, key: str, default: int = 0) -> int: + """Look up an integer configuration value.""" + try: + return int(self.settings.get(key, default)) # should already be int, but check by casting + + except ValueError as e: + logger.error(f"Failed to int({key=})", exc_info=e) + return default + + except Exception as e: + logger.debug('The exception type is ...', exc_info=e) + return default + + def set(self, key: str, val: Union[int, str, list]) -> None: + """Set value on the specified configuration key.""" + self.settings[key] = val + + def delete(self, key: str) -> None: + """Delete the specified configuration key.""" + self.settings.pop(key, None) + + def save(self) -> None: + """Save current configuration to disk.""" + self.defaults.setPersistentDomain_forName_(self.settings, self.identifier) + self.defaults.synchronize() + + def close(self) -> None: + """Close the configuration.""" + self.save() + self.defaults = None + + elif platform == 'win32': + + def __init__(self): + self.app_dir = join(known_folder_path(FOLDERID_LocalAppData), appname) + if not isdir(self.app_dir): + mkdir(self.app_dir) + + self.plugin_dir = join(self.app_dir, 'plugins') + if not isdir(self.plugin_dir): + mkdir(self.plugin_dir) + + if getattr(sys, 'frozen', False): + self.internal_plugin_dir = join(dirname(sys.executable), 'plugins') + self.respath = dirname(sys.executable) + + else: + self.internal_plugin_dir = join(dirname(__file__), 'plugins') + self.respath = dirname(__file__) + + # expanduser in Python 2 on Windows doesn't handle non-ASCII - http://bugs.python.org/issue13207 + self.home = known_folder_path(FOLDERID_Profile) or r'\\' + + journaldir = known_folder_path(FOLDERID_SavedGames) + if journaldir: + self.default_journal_dir = join(journaldir, 'Frontier Developments', 'Elite Dangerous') + + else: + self.default_journal_dir = None + + self.identifier = applongname + self.hkey = HKEY() + disposition = DWORD() + if RegCreateKeyEx( + HKEY_CURRENT_USER, + r'Software\Marginal\EDMarketConnector', + 0, + None, + 0, + KEY_ALL_ACCESS, + None, + ctypes.byref(self.hkey), + ctypes.byref(disposition) + ): + raise Exception() + + # set WinSparkle defaults - https://github.com/vslavik/winsparkle/wiki/Registry-Settings + edcdhkey = HKEY() + if RegCreateKeyEx( + HKEY_CURRENT_USER, + r'Software\EDCD\EDMarketConnector', + 0, + None, + 0, + KEY_ALL_ACCESS, + None, + ctypes.byref(edcdhkey), + ctypes.byref(disposition) + ): + raise Exception() + + sparklekey = HKEY() + if not RegCreateKeyEx( + edcdhkey, + 'WinSparkle', + 0, + None, + 0, + KEY_ALL_ACCESS, + None, + ctypes.byref(sparklekey), + ctypes.byref(disposition) + ): + if disposition.value == REG_CREATED_NEW_KEY: + buf = ctypes.create_unicode_buffer('1') + RegSetValueEx(sparklekey, 'CheckForUpdates', 0, 1, buf, len(buf) * 2) + + buf = ctypes.create_unicode_buffer(str(update_interval)) + RegSetValueEx(sparklekey, 'UpdateInterval', 0, 1, buf, len(buf) * 2) + RegCloseKey(sparklekey) + + if not self.get('outdir') or not isdir(self.get('outdir')): + self.set('outdir', known_folder_path(FOLDERID_Documents) or self.home) + + def get(self, key: str, default: Union[None, list, str] = None) -> Union[None, list, str]: + """Look up a string configuration value.""" + key_type = DWORD() + key_size = DWORD() + # Only strings are handled here. + if ( + RegQueryValueEx( + self.hkey, + key, + 0, + ctypes.byref(key_type), + None, + ctypes.byref(key_size) + ) + or key_type.value not in [REG_SZ, REG_MULTI_SZ] + ): + return default + + buf = ctypes.create_unicode_buffer(int(key_size.value / 2)) + if RegQueryValueEx(self.hkey, key, 0, ctypes.byref(key_type), buf, ctypes.byref(key_size)): + return default + + elif key_type.value == REG_MULTI_SZ: + return list(ctypes.wstring_at(buf, len(buf)-2).split('\x00')) + + else: + return str(buf.value) + + def getint(self, key: str, default: int = 0) -> int: + """Look up an integer configuration value.""" + key_type = DWORD() + key_size = DWORD(4) + key_val = DWORD() + if ( + RegQueryValueEx( + self.hkey, + key, + 0, + ctypes.byref(key_type), + ctypes.byref(key_val), + ctypes.byref(key_size) + ) + or key_type.value != REG_DWORD + ): + return default + + else: + return key_val.value + + def set(self, key: str, val: Union[int, str, list]) -> None: + """Set value on the specified configuration key.""" + if isinstance(val, str): + buf = ctypes.create_unicode_buffer(val) + RegSetValueEx(self.hkey, key, 0, REG_SZ, buf, len(buf)*2) + + elif isinstance(val, numbers.Integral): + RegSetValueEx(self.hkey, key, 0, REG_DWORD, ctypes.byref(DWORD(val)), 4) + + elif isinstance(val, list): + # null terminated non-empty strings + string_val = '\x00'.join([str(x) or ' ' for x in val] + ['']) + buf = ctypes.create_unicode_buffer(string_val) + RegSetValueEx(self.hkey, key, 0, REG_MULTI_SZ, buf, len(buf)*2) + + else: + raise NotImplementedError() + + def delete(self, key: str) -> None: + """Delete the specified configuration key.""" + RegDeleteValue(self.hkey, key) + + def save(self) -> None: + """Save current configuration to disk.""" + pass # Redundant since registry keys are written immediately + + def close(self) -> None: + """Close the configuration.""" + RegCloseKey(self.hkey) + self.hkey = None + + elif platform == 'linux': + SECTION = 'config' + + def __init__(self): + + # http://standards.freedesktop.org/basedir-spec/latest/ar01s03.html + self.app_dir = join(getenv('XDG_DATA_HOME', expanduser('~/.local/share')), appname) + if not isdir(self.app_dir): + makedirs(self.app_dir) + + self.plugin_dir = join(self.app_dir, 'plugins') + if not isdir(self.plugin_dir): + mkdir(self.plugin_dir) + + self.internal_plugin_dir = join(dirname(__file__), 'plugins') + self.default_journal_dir = None + self.home = expanduser('~') + self.respath = dirname(__file__) + self.identifier = f'uk.org.marginal.{appname.lower()}' + + self.filename = join(getenv('XDG_CONFIG_HOME', expanduser('~/.config')), appname, f'{appname}.ini') + if not isdir(dirname(self.filename)): + makedirs(dirname(self.filename)) + + self.config = RawConfigParser(comment_prefixes=('#',)) + try: + with codecs.open(self.filename, 'r') as h: + self.config.read_file(h) + + except Exception as e: + logger.debug('Reading config failed, assuming we\'re making a new one...', exc_info=e) + self.config.add_section(self.SECTION) + + if not self.get('outdir') or not isdir(self.get('outdir')): + self.set('outdir', expanduser('~')) + + def get(self, key: str, default: Union[None, list, str] = None) -> Union[None, list, str]: + """Look up a string configuration value.""" + try: + val = self.config.get(self.SECTION, key) + if '\n' in val: # list + # ConfigParser drops the last entry if blank, + # so we add a spurious ';' entry in set() and remove it here + assert val.split('\n')[-1] == ';', val.split('\n') + return [self._unescape(x) for x in val.split('\n')[:-1]] + else: + return self._unescape(val) + + except NoOptionError: + logger.debug(f'attempted to get key {key} that does not exist') + return default + + except Exception as e: + logger.debug('And the exception type is...', exc_info=e) + return default + + def getint(self, key: str, default: int = 0) -> int: + """Look up an integer configuration value.""" + try: + return self.config.getint(self.SECTION, key) + + except ValueError as e: + logger.error(f"Failed to int({key=})", exc_info=e) + + except NoOptionError: + logger.debug(f'attempted to get key {key} that does not exist') + + except Exception: + logger.exception(f'unexpected exception while attempting to access {key}') + + return default + + def set(self, key: str, val: Union[int, str, list]) -> None: + """Set value on the specified configuration key.""" + if isinstance(val, bool): + self.config.set(self.SECTION, key, val and '1' or '0') + + elif isinstance(val, str) or isinstance(val, numbers.Integral): + self.config.set(self.SECTION, key, self._escape(val)) + + elif isinstance(val, list): + self.config.set(self.SECTION, key, '\n'.join([self._escape(x) for x in val] + [';'])) + + else: + raise NotImplementedError() + + def delete(self, key: str) -> None: + """Delete the specified configuration key.""" + self.config.remove_option(self.SECTION, key) + + def save(self) -> None: + """Save current configuration to disk.""" + with codecs.open(self.filename, 'w', 'utf-8') as h: + self.config.write(h) + + def close(self) -> None: + """Close the configuration.""" + self.save() + self.config = None + + def _escape(self, val: str) -> str: + """Escape a string for storage.""" + return str(val).replace('\\', '\\\\').replace('\n', '\\n').replace(';', '\\;') + + def _unescape(self, val: str) -> str: + """Un-escape a string from storage.""" + chars = list(val) + i = 0 + while i < len(chars): + if chars[i] == '\\': + chars.pop(i) + if chars[i] == 'n': + chars[i] = '\n' + i += 1 + return ''.join(chars) + + else: + def __init__(self): + raise NotImplementedError('Implement me') + + # Common + + def get_password(self, account: str) -> None: + """Legacy password retrieval.""" + warnings.warn("password subsystem is no longer supported", DeprecationWarning) + + def set_password(self, account: str, password: str) -> None: + """Legacy password setting.""" + warnings.warn("password subsystem is no longer supported", DeprecationWarning) + + def delete_password(self, account: str) -> None: + """Legacy password deletion.""" + warnings.warn("password subsystem is no longer supported", DeprecationWarning) + + +old_config = OldConfig() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 00000000..c61c70d0 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,187 @@ +"""Test the config system.""" +from __future__ import annotations + +import contextlib +import itertools +import pathlib +import random +import string +import sys +from typing import Any, List, cast + +import pytest +from pytest import mark + +# isort: split + +sys.path += [str(pathlib.Path(__file__).parent.parent.resolve()), "."] +print(sys.path) + +from _old_config import old_config # noqa: E402 + +from config import LinuxConfig, config # noqa: E402 + + +def _fuzz_list(length: int) -> List[str]: + out = [] + for _ in range(length): + out.append(_fuzz_generators[str](random.randint(0, 1337))) + + return cast(List[str], out) + + +_fuzz_generators = { # Type annotating this would be a nightmare. + int: lambda i: random.randint(min(0, i), max(0, i)), + # This doesn't cover unicode, or random bytes. Use them at your own peril + str: lambda l: "".join(random.choice(string.ascii_letters + string.digits + '\r\n') for _ in range(l)), + bool: lambda _: bool(random.choice((True, False))), + list: _fuzz_list, +} + + +def _get_fuzz(_type: Any, num_values=50, value_length=(0, 10)) -> list: + return [_fuzz_generators[_type](random.randint(*value_length)) for _ in range(num_values)] + + +int_tests = [0, 1, 2, 3, (1 << 32)-1, -1337] + +string_tests = [ + "test", "", "this\nis\na\ntest", "orange sidewinder", "needs \\ more backslashes\\", "\\; \\n", r"\\\\ \\\\; \\\\n", + r"entry with escapes \\ \\; \\n" +] + +list_tests = [ + ["test"], + ["multiple", "entries"], + ["multiple", "entries", "with", "", "blanks"], + ["entry", "that ends", "in a", ""], + ["entry with \n", "newlines\nin", "weird\nplaces"], + [r"entry with escapes \\ \\; \\n"], + [r"\\\\ \\\\; \\\\n"] +] +bool_tests = [True, False] + +big_int = 1 << 64 + + +def _make_params(args: List[Any], id_name: str = 'random_test_{i}'): + return [pytest.param(x, id=id_name.format(i=i)) for i, x in enumerate(args)] + + +def _build_test_list(static_data, random_data, random_id_name='random_test_{i}'): + return itertools.chain(static_data, _make_params(random_data, id_name=random_id_name)) + + +class TestNewConfig: + """Test the new config with an array of hand picked and random data.""" + + @mark.parametrize("i", _build_test_list(int_tests, _get_fuzz(int, value_length=(-big_int, big_int)))) + def test_ints(self, i): + """Save int and then unpack it again.""" + name = f"int_test_{i}" + config.set(name, i) + assert i == config.get_int(name) + config.delete(name) + + @mark.parametrize("string", _build_test_list(string_tests, _get_fuzz(str, value_length=(0, 512)))) + def test_string(self, string: str): + """Save a string and then ask for it back.""" + name = f'str_test_{hash(string)}' + config.set(name, string) + assert string == config.get_str(name) + config.delete(name) + + @mark.parametrize("lst", _build_test_list(list_tests, _get_fuzz(list))) + def test_list(self, lst: List[str]): + """Save a list and then ask for it back.""" + name = f'list_test_{ hash("".join(lst)) }' + config.set(name, lst) + assert lst == config.get_list(name) + + config.delete(name) + + @mark.parametrize('b', bool_tests) + def test_bool(self, b): + """Save a bool and ask for it back.""" + name = str(b) + config.set(name, b) + assert b == config.get_bool(name) + config.delete(name) + + +class TestOldNewConfig: + """Tests going through the old config and out the new config.""" + + KEY_PREFIX = 'oldnew_' + + def teardown_method(self): + """ + Teardown for all config tests to save out configs. + + It is expected that tests have cleaned up whatever keys they have aded. + This is here to ensure that a save out happens for configs that need it. + """ + old_config.save() + config.save() + + def __update_linuxconfig(self): + """On linux config uses ConfigParser, which doesn't update from disk changes. Force the update here.""" + if isinstance(config, LinuxConfig) and config.config is not None: + config.config.read(config.filename) + + @mark.parametrize("i", _build_test_list(int_tests, _get_fuzz(int, 50, (-(1 << 32), 1 << 32)))) + def test_int(self, i: int) -> None: + """Save an int though the old config, recall it using the new config.""" + name = self.KEY_PREFIX + f'int_{i}' + old_config.set(name, i) + old_config.save() + + self.__update_linuxconfig() + + res = config.get_int(name) + with contextlib.ExitStack() as stack: + stack.callback(config.delete, name) + stack.callback(old_config.delete, name) + assert res == i + + @mark.parametrize("string", _build_test_list(string_tests, _get_fuzz(str, value_length=(0, 512)))) + def test_string(self, string: str) -> None: + string = string.replace("\r", "") # The old config does _not_ support \r in its entries. We do. + name = self.KEY_PREFIX + f'string_{hash(string)}' + old_config.set(name, string) + old_config.save() + + self.__update_linuxconfig() + + res = config.get_str(name) + with contextlib.ExitStack() as stack: + stack.callback(config.delete, name) + stack.callback(old_config.delete, name) + assert res == string + + @mark.parametrize("lst", _build_test_list(list_tests, _get_fuzz(list))) + def test_list(self, lst): + lst = [x.replace("\r", "") for x in lst] + name = self.KEY_PREFIX + f'list_test_{ hash("".join(lst)) }' + old_config.set(name, lst) + old_config.save() + + self.__update_linuxconfig() + + res = config.get_list(name) + with contextlib.ExitStack() as stack: + stack.callback(config.delete, name) + stack.callback(old_config.delete, name) + assert res == lst + + @mark.skipif(sys.platform == 'win32', reason="Old Windows config does not support bool types") + @mark.parametrize("b", bool_tests) + def test_bool(self, b): + name = str(b) + old_config.set(name, b) + old_config.save() + self.__update_linuxconfig() + with contextlib.ExitStack() as stack: + stack.callback(config.delete, name) + stack.callback(old_config.delete, name) + assert config.get_bool(name) == b