mirror of
https://github.com/EDCD/EDMarketConnector.git
synced 2026-11-14 02:32:44 +03:00
Added config tests
This commit is contained in:
A_D authored and 
Athanasius committed
1 parent
c72c1be4da
commit
9c3de53a3b
2 files changed
+687
No files matched your search
@@ -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()
|
||||
@@ -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
|
||||
Reference in new issue
Block a user