diff --git a/service/spoc.conf b/service/spoc.conf deleted file mode 100644 index 415eba2..0000000 --- a/service/spoc.conf +++ /dev/null @@ -1,3 +0,0 @@ -[spoc] -data-dir = /var/lib/spoc/ -repo-url = https://repo.spotter.cz/spoc/ diff --git a/setup.cfg b/setup.cfg index 3b8ee3b..d3b13fc 100644 --- a/setup.cfg +++ b/setup.cfg @@ -22,6 +22,7 @@ classifiers = [options] packages = find: package_dir = = src +py_modules = spoc_cli python_requires = >= 3.5 install_requires = requests @@ -30,7 +31,7 @@ where = src [options.entry_points] console_scripts = - spoc = spoc:main + spoc = spoc_cli:main [tool:pytest] testpaths = tests diff --git a/src/spoc/__init__.py b/src/spoc/__init__.py index 0c40f9c..38a634b 100644 --- a/src/spoc/__init__.py +++ b/src/spoc/__init__.py @@ -1,5 +1,3 @@ -import argparse -import os from pkg_resources import parse_version from . import app @@ -9,144 +7,98 @@ from . import podman from . import repo from .flock import locked -def print_lock(pid): - with open(os.path.join('/proc', pid, 'cmdline')) as f: - cmdline = f.read().replace('\0', ' ').strip() - print(f'Waiting for lock currently held by process {pid} - {cmdline}') -@locked(config.LOCK_FILE, print_lock) -def listing(list_type): - if list_type == 'installed': - apps = podman.get_apps() - elif list_type == 'online': - apps = {app:definition['version'] for app,definition in repo.get_apps().items()} - elif list_type == 'updates': - online_apps = {app:definition['version'] for app,definition in repo.get_apps().items()} - apps = {app:f'{version} -> {online_apps[app]}' for app,version in podman.get_apps().items() - if app in online_apps - and parse_version(online_apps[app]) > parse_version(version)} - else: - apps = {} - for app_name, app_version in sorted(apps.items()): - print(app_name, app_version) +class AppError(Exception): + def __init__(self, app_name): + super().__init__(app_name) + self.app_name = app_name -@locked(config.LOCK_FILE, print_lock) +class AppAlreadyInstalledError(AppError): + pass + +class AppNotInstalledError(AppError): + pass + +class AppNotInRepoError(AppError): + pass + +class AppNotUpdateableError(AppError): + pass + + +def list_installed(): + return dict(sorted(podman.get_apps().items())) + +def list_online(): + return {app:definition['version'] for app,definition in sorted(repo.get_apps().items())} + +def list_updates(): + online_apps = {app:definition['version'] for app,definition in repo.get_apps().items()} + apps = {app:f'{version} -> {online_apps[app]}' for app,version in podman.get_apps().items() + if app in online_apps + and parse_version(online_apps[app]) > parse_version(version)} + return dict(sorted(apps.items())) + +@locked() def install(app_name): + if app_name in podman.get_apps(): + raise AppAlreadyInstalledError(app_name) + if app_name not in repo.get_apps(): + raise AppNotInRepoError(app_name) app.install(app_name) -@locked(config.LOCK_FILE, print_lock) +@locked() def update(app_name): + if app_name not in podman.get_apps(): + raise AppNotInstalledError(app_name) + if app_name not in list_updates(): + raise AppNotUpdateableError(app_name) app.update(app_name) -@locked(config.LOCK_FILE, print_lock) +@locked() def uninstall(app_name): + if app_name not in podman.get_apps(): + raise AppNotInstalledError(app_name) app.uninstall(app_name) -@locked(config.LOCK_FILE, print_lock) +@locked() def start(app_name): + if app_name not in podman.get_apps(): + raise AppNotInstalledError(app_name) podman.start_pod(app_name) -@locked(config.LOCK_FILE, print_lock) +@locked() def stop(app_name): + if app_name not in podman.get_apps(): + raise AppNotInstalledError(app_name) podman.stop_pod(app_name) -@locked(config.LOCK_FILE, print_lock) def status(app_name): - app_status = podman.get_pod_status(app_name) - print(app_status) + if app_name not in podman.get_apps(): + raise AppNotInstalledError(app_name) + return podman.get_pod_status(app_name) -@locked(config.LOCK_FILE, print_lock) -def set_autostart(app_name, value): - enabled = value.lower() in ('1', 'on', 'enable', 'true') +@locked() +def set_autostart(app_name, enabled): + if app_name not in podman.get_apps(): + raise AppNotInstalledError(app_name) autostart.set_app(app_name, enabled) -@locked(config.LOCK_FILE, print_lock) +@locked() def start_autostarted(): for app_name in autostart.get_apps(): podman.start_pod(app_name) -@locked(config.LOCK_FILE, print_lock) +@locked() def stop_all(): for app_name in podman.get_apps(): podman.stop_pod(app_name) -@locked(config.LOCK_FILE, print_lock) +@locked() +def login(host, username, password): + config.write_auth(host, username, password) + repo.load(force=True) + +@locked() def prune(): podman.prune() - -def parse_args(args=None): - parser = argparse.ArgumentParser(description='SPOC application manager') - parser.set_defaults(action=None) - subparsers = parser.add_subparsers() - - parser_list = subparsers.add_parser('list') - parser_list.set_defaults(action=listing) - parser_list.add_argument('type', choices=('installed', 'online', 'updates'), - default='installed', const='installed', nargs='?', - help='Selected repository or application criteria') - - parser_install = subparsers.add_parser('install') - parser_install.set_defaults(action=install) - parser_install.add_argument('app', help='Name of the application to install') - - parser_update = subparsers.add_parser('update') - parser_update.set_defaults(action=update) - parser_update.add_argument('app', help='Name of the application to update') - - parser_uninstall = subparsers.add_parser('uninstall') - parser_uninstall.set_defaults(action=uninstall) - parser_uninstall.add_argument('app', help='Name of the application to uninstall') - - parser_start = subparsers.add_parser('start') - parser_start.set_defaults(action=start) - parser_start.add_argument('app', help='Name of the application to start') - - parser_stop = subparsers.add_parser('stop') - parser_stop.set_defaults(action=stop) - parser_stop.add_argument('app', help='Name of the application to stop') - - parser_status = subparsers.add_parser('status') - parser_status.set_defaults(action=status) - parser_status.add_argument('app', nargs='?', help='Name of the application to check') - - parser_autostart = subparsers.add_parser('autostart') - parser_autostart.set_defaults(action=set_autostart) - parser_autostart.add_argument('app', help='Name of the application to be automatically started') - parser_autostart.add_argument('value', choices=('1', 'on', 'enable', 'true', '0', 'off', 'disable', 'false'), help='Set or unset the applications to be automatically started after the host boots up') - - parser_start_autostarted = subparsers.add_parser('start-autostarted') - parser_start_autostarted.set_defaults(action=start_autostarted) - - parser_stop_all = subparsers.add_parser('stop-all') - parser_stop_all.set_defaults(action=stop_all) - - parser_prune = subparsers.add_parser('prune') - parser_prune.set_defaults(action=prune) - - return parser.parse_args(args) - - -def main(): - args = parse_args() - if args.action is listing: - listing(args.type) - elif args.action is install: - install(args.app) - elif args.action is update: - update(args.app) - elif args.action is uninstall: - uninstall(args.app) - elif args.action is start: - start(args.app) - elif args.action is stop: - stop(args.app) - elif args.action is status: - status(args.app) - elif args.action is set_autostart: - set_autostart(args.app, args.value) - elif args.action is start_autostarted: - start_autostarted() - elif args.action is stop_all: - stop_all() - elif args.action is prune: - prune() diff --git a/src/spoc/app.py b/src/spoc/app.py index 2972010..bfd2079 100644 --- a/src/spoc/app.py +++ b/src/spoc/app.py @@ -1,5 +1,6 @@ import os +from . import autostart from . import config from . import depsolver from . import podman @@ -44,6 +45,7 @@ class App: self.create_containers(containers) def uninstall(self): + autostart.set_app(self.app_name, False) self.remove_pod() self.remove_env_vars() self.remove_volumes(self.get_existing_volumes()) @@ -56,21 +58,21 @@ class App: podman.remove_pod(self.app_name) def read_env_vars(self): - vars = {} + env_vars = {} try: with open(self.env_file) as f: lines = f.read().splitlines() for line in lines: key,value = line.split('=', 1) - vars[key] = value + env_vars[key] = value except FileNotFoundError: pass - return vars + return env_vars - def write_env_vars(self, vars): + def write_env_vars(self, env_vars): os.makedirs(config.DATA_DIR, exist_ok=True) with open(self.env_file, 'w') as f: - for key,value in vars.items(): + for key,value in env_vars.items(): f.write(f'{key}={value}\n') def remove_env_vars(self): diff --git a/src/spoc/autostart.py b/src/spoc/autostart.py index 37a8285..2b4be52 100644 --- a/src/spoc/autostart.py +++ b/src/spoc/autostart.py @@ -18,7 +18,7 @@ def set_app(app_name, enabled): try: apps.remove(app_name) except KeyError: - pass + return os.makedirs(config.DATA_DIR, exist_ok=True) with open(config.AUTOSTART_FILE, 'w') as f: for app in apps: diff --git a/src/spoc/config.py b/src/spoc/config.py index ac549d4..4f1d621 100644 --- a/src/spoc/config.py +++ b/src/spoc/config.py @@ -1,23 +1,38 @@ -import configparser -import os +import json +from base64 import b64decode, b64encode -CONFIG_FILE = '/etc/spoc/spoc.conf' +DATA_DIR = '/var/lib/spoc' +AUTOSTART_FILE = '/var/lib/spoc/autostart' LOCK_FILE = '/run/lock/spoc.lock' -DATA_DIR = None -AUTOSTART_FILE = None -REPO_BASE_URL = None +REGISTRY_HOST = None +REGISTRY_AUTH = None +REGISTRY_AUTH_FILE = '/var/lib/spoc/auth.json' REPO_FILE_URL = None -def reload(config_file=CONFIG_FILE): - global DATA_DIR, AUTOSTART_FILE, REPO_BASE_URL, REPO_FILE_URL +def read_auth(): + global REGISTRY_HOST, REGISTRY_AUTH, REPO_FILE_URL # pylint: disable=global-statement - config = configparser.ConfigParser() - config.read(config_file) + try: + with open(REGISTRY_AUTH_FILE) as f: + data = json.load(f) + REGISTRY_HOST = next(iter(data['auths'].keys())) + auth = b64decode(data['auths'][REGISTRY_HOST]['auth'].encode()).decode() + REGISTRY_AUTH = tuple(auth.split(':', 1)) + except FileNotFoundError: + REGISTRY_HOST = 'localhost' + REGISTRY_AUTH = None + REPO_FILE_URL = f'https://{REGISTRY_HOST}/repository.json' - DATA_DIR = config.get('spoc', 'data-dir', fallback='/var/lib/spoc') - AUTOSTART_FILE = os.path.join(DATA_DIR, 'autostart') - REPO_BASE_URL = config.get('spoc', 'repo-url', fallback='https://localhost').rstrip('/') - REPO_FILE_URL = f'{REPO_BASE_URL}/repository.json' +def write_auth(host, username, password): + global REGISTRY_HOST, REGISTRY_AUTH, REPO_FILE_URL # pylint: disable=global-statement -reload() + b64auth = b64encode(f'{username}:{password}'.encode()).decode() + data = json.dumps({'auths': {host: {'auth': b64auth}}}) + with open(REGISTRY_AUTH_FILE, 'w') as f: + f.write(data) + REGISTRY_HOST = host + REGISTRY_AUTH = (username, password) + REPO_FILE_URL = f'https://{REGISTRY_HOST}/repository.json' + +read_auth() diff --git a/src/spoc/depsolver.py b/src/spoc/depsolver.py index bc1357d..0780079 100644 --- a/src/spoc/depsolver.py +++ b/src/spoc/depsolver.py @@ -1,6 +1,7 @@ class CircularDependencyError(Exception): # Dependecy solver has found a circular dependency between items def __init__(self, deps): + super().__init__(deps) self.deps = deps def __str__(self): @@ -12,6 +13,7 @@ class CircularDependencyError(Exception): class MissingDependencyError(Exception): # Dependecy solver has found an items depents on a nonexistent item def __init__(self, deps, missing): + super().__init__(deps, missing) self.deps = deps self.missing = missing @@ -30,7 +32,6 @@ class DepSolver: def add(self, item, dependencies): self.unresolved[item] = set(dependencies) -#flat_list = [item for sublist in t for item in sublist] def solve(self): # Returns a list of instances ordered by dependency resolved = [] @@ -44,9 +45,8 @@ class DepSolver: missing_deps = wanted_deps - set(self.unresolved) if missing_deps: raise MissingDependencyError(self.unresolved, missing_deps) - else: - # If all dependencies exist, we have found a circular dependency - raise CircularDependencyError(self.unresolved) + # If all dependencies exist, we have found a circular dependency + raise CircularDependencyError(self.unresolved) # Add resolved items to the result and remove from the unresolved ones for item in batch: resolved.append(item) diff --git a/src/spoc/flock.py b/src/spoc/flock.py index edbb978..ae40aa2 100644 --- a/src/spoc/flock.py +++ b/src/spoc/flock.py @@ -2,30 +2,39 @@ import errno import fcntl import os import time +import sys from contextlib import contextmanager +from . import config + +def print_lock(pid): + with open(os.path.join('/proc', pid, 'cmdline')) as f: + cmdline = f.read().replace('\0', ' ').strip() + print(f'Waiting for lock currently held by process {pid} - {cmdline}', file=sys.stderr) + @contextmanager -def lock(lock_file, fail_callback=None): - with open(lock_file, 'a'): +def locked(): + with open(config.LOCK_FILE, 'a'): # Open the lock file in append mode first to ensure its existence # but not modify any data if it already exists pass # Open the lock file in read + write mode without truncation - with open(lock_file, 'r+') as f: + with open(config.LOCK_FILE, 'r+') as f: + lock_printed = False while True: try: # Try to obtain exclusive lock in non-blocking mode fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB) break except OSError as e: - # If lock is already locked by another process + # If lock is held by another process if e.errno == errno.EAGAIN: - if fail_callback: - # Call the callback function with contents of the lock file + if not lock_printed: + # Print a message using contents of the lock file # (PID of the process holding the lock) - fail_callback(f.read()) - # Remove the callback function so it's not called in every loop - fail_callback = None + print_lock(f.read()) + # Set flag so the message is not printed in every loop + lock_printed = True # Set the position for future truncation f.seek(0) # Wait for the lock to be freed @@ -38,12 +47,3 @@ def lock(lock_file, fail_callback=None): f.write(str(os.getpid())) f.flush() yield f - -# Function decorator -def locked(lock_file, fail_callback=None): - def decorator(target): - def wrapper(*args, **kwargs): - with lock(lock_file, fail_callback): - return target(*args, **kwargs) - return wrapper - return decorator diff --git a/src/spoc/podman.py b/src/spoc/podman.py index 524858d..1f74c1c 100644 --- a/src/spoc/podman.py +++ b/src/spoc/podman.py @@ -1,11 +1,21 @@ import json +import os import subprocess +from . import config + +ENV = os.environ.copy() +ENV['REGISTRY_AUTH_FILE'] = config.REGISTRY_AUTH_FILE + +def run(cmd, **kwargs): + return subprocess.run(cmd, check=True, env=ENV, **kwargs) + +def out(cmd, **kwargs): + return run(cmd, stdout=subprocess.PIPE, text=True, **kwargs).stdout.rstrip() + def get_apps(): apps = {} - cmd = ['podman', 'pod', 'ps', '--format', 'json'] - pod_ps = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, text=True) - data = json.loads(pod_ps.stdout) + data = json.loads(out(['podman', 'pod', 'ps', '--format', 'json'])) for pod in data: app_name = pod['Labels'].get('spoc.app') app_version = pod['Labels'].get('spoc.version') @@ -14,42 +24,35 @@ def get_apps(): return apps def get_volumes_for_app(app_name): - cmd = ['podman', 'volume', 'ls', '--filter', f'label=spoc.app={app_name}', '--format', 'json'] - volume_ls = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, text=True) - return set(volume['Name'] for volume in json.loads(volume_ls.stdout)) + volume_ls = out(['podman', 'volume', 'ls', '--filter', f'label=spoc.app={app_name}', + '--format', 'json']) + return set(volume['Name'] for volume in json.loads(volume_ls)) def start_pod(app_name): - cmd = ['podman', 'pod', 'start', app_name] - subprocess.run(cmd, check=True) + run(['podman', 'pod', 'start', app_name]) def stop_pod(app_name): - cmd = ['podman', 'pod', 'stop', '--ignore', app_name] - subprocess.run(cmd, check=True) + run(['podman', 'pod', 'stop', '--ignore', app_name]) def get_pod_status(app_name=None): cmd = ['podman', 'pod', 'ps'] if app_name: cmd.extend(['--filter', f'label=spoc.app={app_name}']) - pod_ps = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, text=True) - return pod_ps.stdout.strip() + return out(cmd) def create_volume(app_name, vol_name): - cmd = ['podman', 'volume', 'create', '--label', f'spoc.app={app_name}', vol_name] - subprocess.run(cmd, check=True) + run(['podman', 'volume', 'create', '--label', f'spoc.app={app_name}', vol_name]) def remove_volume(vol_name): - cmd = ['podman', 'volume', 'rm', vol_name] - subprocess.run(cmd, check=True) + run(['podman', 'volume', 'rm', vol_name]) def create_pod(app_name, app_version): - cmd = ['podman', 'pod', 'create', '--name', app_name, - '--label', f'spoc.app={app_name}', '--label', f'spoc.version={app_version}'] - subprocess.run(cmd, check=True) + run(['podman', 'pod', 'create', '--name', app_name, + '--label', f'spoc.app={app_name}', '--label', f'spoc.version={app_version}']) def remove_pod(app_name): stop_pod(app_name) - cmd = ['podman', 'pod', 'rm', '--ignore', app_name] - subprocess.run(cmd, check=True) + run(['podman', 'pod', 'rm', '--ignore', app_name]) def create_container(app_name, cnt_name, image, env_file=None, volumes=None, requires=None, hosts=None): @@ -66,10 +69,8 @@ def create_container(app_name, cnt_name, image, env_file=None, volumes=None, for host in sorted(hosts): cmd.extend(['--add-host', f'{host}:127.0.0.1']) cmd.append(image) - subprocess.run(cmd, check=True) + run(cmd) def prune(): - cmd = ['podman', 'image', 'prune', '--all', '--force'] - subprocess.run(cmd, check=True) - cmd = ['podman', 'volume', 'prune', '--force'] - subprocess.run(cmd, check=True) + run(['podman', 'image', 'prune', '--all', '--force']) + run(['podman', 'volume', 'prune', '--force']) diff --git a/src/spoc/repo.py b/src/spoc/repo.py index 8f6e9a4..3c008f4 100644 --- a/src/spoc/repo.py +++ b/src/spoc/repo.py @@ -5,9 +5,10 @@ from . import config _data = {} def load(force=False): - global _data + global _data # pylint: disable=global-statement if not _data or force: - response = requests.get(config.REPO_FILE_URL, timeout=5) + _data = {} + response = requests.get(config.REPO_FILE_URL, auth=config.REGISTRY_AUTH, timeout=5) response.raise_for_status() _data = response.json() diff --git a/src/spoc_cli.py b/src/spoc_cli.py new file mode 100644 index 0000000..0311ba4 --- /dev/null +++ b/src/spoc_cli.py @@ -0,0 +1,168 @@ +import argparse +import getpass +import requests +import sys + +import spoc + +APP_ERROR_STRINGS = { + 'AppAlreadyInstalledError': 'Application {} is already installed', + 'AppNotInstalledError': 'Application {} is not installed', + 'AppNotInRepoError': 'Application {} does not exist in the repository', + 'AppNotUpdateableError': 'Application {} does not have a newer version to update', +} + +def handle_app_error(exception): + ex_type = type(exception).__name__ + print(APP_ERROR_STRINGS[ex_type].format(exception.app_name), file=sys.stderr) + +def handle_repo_error(exception): + if isinstance(exception, requests.HTTPError): + status = exception.response.status_code + if status == 401: + reason = 'Invalid username/password' + else: + reason = f'{status} {exception.response.reason}' + else: + ex_type = type(exception) + reason = f'{ex_type.__module__}.{ex_type.__name__}' + print(f'Online repository cannot be reached due to: {reason}', file=sys.stderr) + +def listing(list_type): + if list_type == 'installed': + apps = spoc.list_installed() + elif list_type == 'online': + apps = spoc.list_online() + elif list_type == 'updates': + apps = spoc.list_updates() + else: + apps = {} + for app_name, app_version in apps.items(): + print(app_name, app_version) + +def install(app_name): + spoc.install(app_name) + +def update(app_name): + spoc.update(app_name) + +def uninstall(app_name): + spoc.uninstall(app_name) + +def start(app_name): + spoc.start(app_name) + +def stop(app_name): + spoc.stop(app_name) + +def status(app_name): + print(spoc.status(app_name)) + +def set_autostart(app_name, value): + enabled = value.lower() in ('1', 'on', 'enable', 'true') + spoc.set_autostart(app_name, enabled) + +def start_autostarted(): + spoc.start_autostarted() + +def stop_all(): + spoc.stop_all() + +def login(host): + username = input('Username: ') + password = getpass.getpass() + spoc.login(host, username, password) + +def prune(): + spoc.prune() + + +def parse_args(args=None): + parser = argparse.ArgumentParser(description='SPOC application manager') + parser.set_defaults(action=None) + subparsers = parser.add_subparsers() + + parser_list = subparsers.add_parser('list') + parser_list.set_defaults(action=listing) + parser_list.add_argument('type', choices=('installed', 'online', 'updates'), + default='installed', const='installed', nargs='?', + help='Selected repository or application criteria') + + parser_install = subparsers.add_parser('install') + parser_install.set_defaults(action=install) + parser_install.add_argument('app', help='Name of the application to install') + + parser_update = subparsers.add_parser('update') + parser_update.set_defaults(action=update) + parser_update.add_argument('app', help='Name of the application to update') + + parser_uninstall = subparsers.add_parser('uninstall') + parser_uninstall.set_defaults(action=uninstall) + parser_uninstall.add_argument('app', help='Name of the application to uninstall') + + parser_start = subparsers.add_parser('start') + parser_start.set_defaults(action=start) + parser_start.add_argument('app', help='Name of the application to start') + + parser_stop = subparsers.add_parser('stop') + parser_stop.set_defaults(action=stop) + parser_stop.add_argument('app', help='Name of the application to stop') + + parser_status = subparsers.add_parser('status') + parser_status.set_defaults(action=status) + parser_status.add_argument('app', nargs='?', help='Name of the application to check') + + parser_autostart = subparsers.add_parser('autostart') + parser_autostart.set_defaults(action=set_autostart) + parser_autostart.add_argument('app', help='Name of the application to be automatically started') + parser_autostart.add_argument('value', choices=('1', 'on', 'enable', 'true', + '0', 'off', 'disable', 'false'), + help='Set or unset the applications to be automatically ' \ + 'started after the host boots up') + + parser_start_autostarted = subparsers.add_parser('start-autostarted') + parser_start_autostarted.set_defaults(action=start_autostarted) + + parser_stop_all = subparsers.add_parser('stop-all') + parser_stop_all.set_defaults(action=stop_all) + + parser_login = subparsers.add_parser('login') + parser_login.set_defaults(action=login) + parser_login.add_argument('host', help='Hostname of the container registry') + + parser_prune = subparsers.add_parser('prune') + parser_prune.set_defaults(action=prune) + + return parser.parse_args(args) + +def main(): + args = parse_args() + try: + if args.action is listing: + listing(args.type) + elif args.action is install: + install(args.app) + elif args.action is update: + update(args.app) + elif args.action is uninstall: + uninstall(args.app) + elif args.action is start: + start(args.app) + elif args.action is stop: + stop(args.app) + elif args.action is status: + status(args.app) + elif args.action is set_autostart: + set_autostart(args.app, args.value) + elif args.action is start_autostarted: + start_autostarted() + elif args.action is stop_all: + stop_all() + elif args.action is login: + login(args.host) + elif args.action is prune: + prune() + except spoc.AppError as ex: + handle_app_error(ex) + except requests.RequestException as ex: + handle_repo_error(ex) diff --git a/tests/test_app.py b/tests/test_app.py index d5db0c5..5843a35 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -10,11 +10,15 @@ TEST_DATA_DIR = os.path.join(os.path.dirname(__file__), 'test_data') with open(os.path.join(TEST_DATA_DIR, 'repository.json')) as f: MOCK_REPODATA = json.load(f) -with open(os.path.join(TEST_DATA_DIR, 'test.env')) as f: - MOCK_ENV = f.read() +MOCK_ENV = 'RAILS_ENV=test\n' \ + 'POSTGRES_PASSWORD=asdf=1234\n' \ + 'SOMEKEY=someval\n' -with open(os.path.join(TEST_DATA_DIR, 'test.env.json')) as f: - MOCK_ENV_JSON = json.load(f) +MOCK_ENV_DATA = { + 'RAILS_ENV': 'test', + 'POSTGRES_PASSWORD': 'asdf=1234', + 'SOMEKEY': 'someval', +} def test_init(): @@ -49,7 +53,7 @@ def test_install(create_containers, create_pod, write_env_vars, read_env_vars, c @patch('spoc.app.App.get_existing_volumes', return_value=set(('somevol', 'migrate', 'storage'))) @patch('spoc.app.App.remove_volumes') @patch('spoc.app.App.create_volumes') -@patch('spoc.app.App.read_env_vars', return_value=MOCK_ENV_JSON) +@patch('spoc.app.App.read_env_vars', return_value=MOCK_ENV_DATA) @patch('spoc.app.App.write_env_vars') @patch('spoc.app.App.create_pod') @patch('spoc.app.App.create_containers') @@ -64,20 +68,22 @@ def test_update(create_containers, create_pod, write_env_vars, read_env_vars, cr create_volumes.assert_called_once_with(set(('uploads', 'postgres-data'))) read_env_vars.assert_called_once() expected_env_data = MOCK_REPODATA['someapp']['environment'].copy() - expected_env_data.update(MOCK_ENV_JSON) + expected_env_data.update(MOCK_ENV_DATA) del expected_env_data['SOMEKEY'] write_env_vars.assert_called_once_with(expected_env_data) create_pod.assert_called_once_with('0.23.5-210416') create_containers.assert_called_once_with(MOCK_REPODATA['someapp']['containers']) +@patch('spoc.autostart.set_app') @patch('spoc.app.App.remove_pod') @patch('spoc.app.App.remove_env_vars') @patch('spoc.app.App.get_existing_volumes', return_value=set(('somevol', 'anothervol'))) @patch('spoc.app.App.remove_volumes') -def test_uninstall(remove_volumes, get_existing_volumes, remove_env_vars, remove_pod): +def test_uninstall(remove_volumes, get_existing_volumes, remove_env_vars, remove_pod, autostart): instance = app.App('someapp') instance.uninstall() + autostart.assert_called_with('someapp', False) remove_pod.assert_called_once() remove_env_vars.assert_called_once() get_existing_volumes.assert_called_once() @@ -106,7 +112,7 @@ def test_read_env_vars(env_open): env_file = os.path.join(config.DATA_DIR, 'someapp.env') env_open.assert_called_once_with(env_file) - assert env_vars == MOCK_ENV_JSON + assert env_vars == MOCK_ENV_DATA @patch('builtins.open', side_effect=FileNotFoundError('someapp.env')) def test_read_env_vars_filenotfound(env_open): @@ -121,7 +127,7 @@ def test_read_env_vars_filenotfound(env_open): @patch('builtins.open', new_callable=mock_open) def test_write_env_vars(env_open, makedirs): instance = app.App('someapp') - instance.write_env_vars(MOCK_ENV_JSON) + instance.write_env_vars(MOCK_ENV_DATA) makedirs.assert_called_once_with(config.DATA_DIR, exist_ok=True) env_file = os.path.join(config.DATA_DIR, 'someapp.env') diff --git a/tests/test_autostart.py b/tests/test_autostart.py index b6fae03..fc4466c 100644 --- a/tests/test_autostart.py +++ b/tests/test_autostart.py @@ -51,8 +51,6 @@ def test_set_app_nonexistent(file_open, get_apps, makedirs): autostart.set_app('anotherapp', False) get_apps.assert_called_once() - makedirs.assert_called_once_with(config.DATA_DIR, exist_ok=True) - file_open.assert_called_once_with(config.AUTOSTART_FILE, 'w') - file_open().write.assert_has_calls([ - call('someapp\n'), - ]) + makedirs.assert_not_called() + file_open.assert_not_called() + file_open().write.assert_not_called() diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..cad1e7b --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,343 @@ + +import pytest +import requests +from argparse import Namespace +from unittest.mock import patch + +import spoc +import spoc_cli + +class MockResponse: + def __init__(self, status_code, reason): + self.status_code = status_code + self.reason = reason + +@pytest.mark.parametrize('exception,expected',[ + (spoc.AppAlreadyInstalledError('someapp'), + 'Application someapp is already installed\n'), + (spoc.AppNotInstalledError('someapp'), + 'Application someapp is not installed\n'), + (spoc.AppNotInRepoError('someapp'), + 'Application someapp does not exist in the repository\n'), + (spoc.AppNotUpdateableError('someapp'), + 'Application someapp does not have a newer version to update\n'), +]) +def test_handle_repo_error(exception, expected, capsys): + spoc_cli.handle_app_error(exception) + + captured = capsys.readouterr() + assert captured.err == expected + +@pytest.mark.parametrize('exception,expected',[ + (requests.HTTPError(response=MockResponse(401, 'Unauthorized')), + 'Invalid username/password'), + (requests.HTTPError(response=MockResponse(404, 'Not Found')), + '404 Not Found'), + (requests.ConnectTimeout(), + 'requests.exceptions.ConnectTimeout'), +]) +def test_handle_app_error(exception, expected, capsys): + spoc_cli.handle_repo_error(exception) + + captured = capsys.readouterr() + expected = f'Online repository cannot be reached due to: {expected}\n' + assert captured.err == expected + +@patch('spoc.list_installed', return_value={'anotherapp': '0.1', 'someapp': '0.1'}) +@patch('spoc.list_online') +@patch('spoc.list_updates') +def test_listing_installed(list_updates, list_online, list_installed, capsys): + spoc_cli.listing('installed') + + list_installed.assert_called_once() + list_online.assert_not_called() + list_updates.assert_not_called() + + captured = capsys.readouterr() + assert captured.out == 'anotherapp 0.1\nsomeapp 0.1\n' + +@patch('spoc.list_installed') +@patch('spoc.list_online', return_value={'anotherapp': '0.2', 'someapp': '0.2'}) +@patch('spoc.list_updates') +def test_listing_online(list_updates, list_online, list_installed, capsys): + spoc_cli.listing('online') + + list_installed.assert_not_called() + list_online.assert_called_once() + list_updates.assert_not_called() + + captured = capsys.readouterr() + assert captured.out == 'anotherapp 0.2\nsomeapp 0.2\n' + +@patch('spoc.list_installed') +@patch('spoc.list_online') +@patch('spoc.list_updates', return_value={'anotherapp': '0.1 -> 0.2', 'someapp': '0.1 -> 0.2'}) +def test_listing_updates(list_updates, list_online, list_installed, capsys): + spoc_cli.listing('updates') + + list_installed.assert_not_called() + list_online.assert_not_called() + list_updates.assert_called_once() + + captured = capsys.readouterr() + assert captured.out == 'anotherapp 0.1 -> 0.2\nsomeapp 0.1 -> 0.2\n' + +@patch('spoc.list_installed') +@patch('spoc.list_online') +@patch('spoc.list_updates') +def test_listing_invalid(list_updates, list_online, list_installed, capsys): + spoc_cli.listing('invalid') + + list_installed.assert_not_called() + list_online.assert_not_called() + list_updates.assert_not_called() + + captured = capsys.readouterr() + assert not captured.out + +@patch('spoc.install') +def test_install(install): + spoc_cli.install('someapp') + + install.assert_called_once_with('someapp') + +@patch('spoc.update') +def test_update(update): + spoc_cli.update('someapp') + + update.assert_called_once_with('someapp') + +@patch('spoc.uninstall') +def test_uninstall(uninstall): + spoc_cli.uninstall('someapp') + + uninstall.assert_called_once_with('someapp') + +@patch('spoc.start') +def test_start(start): + spoc_cli.start('someapp') + + start.assert_called_once_with('someapp') + +@patch('spoc.stop') +def test_stop(stop): + spoc_cli.stop('someapp') + + stop.assert_called_once_with('someapp') + +@patch('spoc.status', return_value='STATUS') +def test_status(status, capsys): + spoc_cli.status('someapp') + + status.assert_called_once_with('someapp') + captured = capsys.readouterr() + assert captured.out == 'STATUS\n' + +@pytest.mark.parametrize('value,expected',[ + ('1', True), + ('on', True), + ('Enable', True), + ('TRUE', True), + ('0', False), + ('off', False), + ('Disable', False), + ('FALSE', False), + ('whatever', False), +]) +@patch('spoc.set_autostart') +def test_set_autostart(set_autostart, value, expected): + spoc_cli.set_autostart('someapp', value) + + set_autostart.assert_called_once_with('someapp', expected) + +@patch('spoc.start_autostarted') +def test_start_autostarted(start_autostarted): + spoc_cli.start_autostarted() + + start_autostarted.assert_called_once() + +@patch('spoc.stop_all') +def test_stop_all(stop_all): + spoc_cli.stop_all() + + stop_all.assert_called_once() + +@patch('builtins.input', return_value='someuser') +@patch('getpass.getpass', return_value='somepass') +@patch('spoc.login') +def test_login(login, getpass, nput): + spoc_cli.login('somehost') + + nput.assert_called_once_with('Username: ') + getpass.assert_called_once() + login.assert_called_once_with('somehost', 'someuser', 'somepass') + +@patch('spoc.prune') +def test_prune(prune): + spoc_cli.prune() + + prune.assert_called_once() + +@patch('sys.argv', ['foo', 'list']) +@patch('spoc_cli.listing') +def test_main_listing(listing): + spoc_cli.main() + + listing.assert_called_once_with('installed') + +@patch('sys.argv', ['foo', 'list', 'online']) +@patch('spoc_cli.listing') +def test_main_listing_online(listing): + spoc_cli.main() + + listing.assert_called_once_with('online') + +@patch('sys.argv', ['foo', 'install', 'someapp']) +@patch('spoc_cli.install') +def test_main_install(install): + spoc_cli.main() + + install.assert_called_once_with('someapp') + +@patch('sys.argv', ['foo', 'update', 'someapp']) +@patch('spoc_cli.update') +def test_main_update(update): + spoc_cli.main() + + update.assert_called_once_with('someapp') + +@patch('sys.argv', ['foo', 'uninstall', 'someapp']) +@patch('spoc_cli.uninstall') +def test_main_uninstall(uninstall): + spoc_cli.main() + + uninstall.assert_called_once_with('someapp') + +@patch('sys.argv', ['foo', 'start', 'someapp']) +@patch('spoc_cli.start') +def test_main_start(start): + spoc_cli.main() + + start.assert_called_once_with('someapp') + +@patch('sys.argv', ['foo', 'stop', 'someapp']) +@patch('spoc_cli.stop') +def test_main_stop(stop): + spoc_cli.main() + + stop.assert_called_once_with('someapp') + +@patch('sys.argv', ['foo', 'status', 'someapp']) +@patch('spoc_cli.status') +def test_main_status(status): + spoc_cli.main() + + status.assert_called_once_with('someapp') + +@patch('sys.argv', ['foo', 'status']) +@patch('spoc_cli.status') +def test_main_status_all(status): + spoc_cli.main() + + status.assert_called_once_with(None) + +@patch('sys.argv', ['foo', 'autostart', 'someapp', 'on']) +@patch('spoc_cli.set_autostart') +def test_main_autostart(autostart): + spoc_cli.main() + + autostart.assert_called_once_with('someapp', 'on') + +@patch('sys.argv', ['foo', 'start-autostarted']) +@patch('spoc_cli.start_autostarted') +def test_main_start_autostarted(start_autostarted): + spoc_cli.main() + + start_autostarted.assert_called_once() + +@patch('sys.argv', ['foo', 'stop-all']) +@patch('spoc_cli.stop_all') +def test_main_stop_all(stop_all): + spoc_cli.main() + + stop_all.assert_called_once() + + +@patch('sys.argv', ['foo', 'login', 'example.com']) +@patch('spoc_cli.login') +def test_main_login(login): + spoc_cli.main() + + login.assert_called_once_with('example.com') + +@patch('sys.argv', ['foo', 'prune']) +@patch('spoc_cli.prune') +def test_main_prune(prune): + spoc_cli.main() + + prune.assert_called_once() + +@patch('spoc_cli.parse_args', return_value=Namespace(action=None)) +@patch('spoc_cli.listing') +@patch('spoc_cli.install') +@patch('spoc_cli.update') +@patch('spoc_cli.uninstall') +@patch('spoc_cli.start') +@patch('spoc_cli.stop') +@patch('spoc_cli.status') +@patch('spoc_cli.start_autostarted') +@patch('spoc_cli.stop_all') +@patch('spoc_cli.login') +@patch('spoc_cli.prune') +def test_main_invalid(prune, login, stop_all, start_autostarted, status, stop, start, + uninstall, update, install, listing, parse_args): + spoc_cli.main() + + parse_args.assert_called_once() + listing.assert_not_called() + install.assert_not_called() + update.assert_not_called() + uninstall.assert_not_called() + start.assert_not_called() + stop.assert_not_called() + status.assert_not_called() + start_autostarted.assert_not_called() + stop_all.assert_not_called() + login.assert_not_called() + prune.assert_not_called() + +@pytest.mark.parametrize('argv', [ + ['list', 'invalid'], + ['install'], + ['update'], + ['uninstall'], + ['start'], + ['stop'], + ['autostart'], + ['autostart', 'someapp'], + ['login'], + ['invalid'], +]) +def test_main_systemexit(argv): + argv.insert(0, 'foo') + with patch('sys.argv', argv): + with pytest.raises(SystemExit): + spoc_cli.main() + +@patch('sys.argv', ['foo', 'start', 'someapp']) +@patch('spoc_cli.start', side_effect=spoc.AppNotInstalledError('someapp')) +@patch('spoc_cli.handle_app_error') +def test_main_apperror(handle_app_error, start): + spoc_cli.main() + + start.assert_called_once() + handle_app_error.assert_called_once() + +@patch('sys.argv', ['foo', 'login', 'somehost']) +@patch('spoc_cli.login', side_effect=requests.HTTPError(response=MockResponse(401, 'Unauthorized'))) +@patch('spoc_cli.handle_repo_error') +def test_main_repoerror(handle_repo_error, login): + spoc_cli.main() + + login.assert_called_once() + handle_repo_error.assert_called_once() diff --git a/tests/test_config.py b/tests/test_config.py index 56bd74f..0fe08cc 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,25 +1,34 @@ -import os +from unittest.mock import mock_open, patch from spoc import config -TEST_DATA_DIR = os.path.join(os.path.dirname(__file__), 'test_data') -with open(os.path.join(TEST_DATA_DIR, 'spoc.conf')) as f: - MOCK_CONFIG = f.read() +@patch('builtins.open', new_callable=mock_open, + read_data='{"auths": {"example.com": {"auth": "c29tZXVzZXI6c29tZXBhc3N3b3Jk"}}}') +def test_read_auth(auth_open): + config.read_auth() -def test_config(): - config_file = os.path.join(os.path.dirname(__file__), 'test_data/spoc.conf') - config.reload(config_file) + auth_open.assert_called_once_with(config.REGISTRY_AUTH_FILE) + assert config.REGISTRY_HOST == 'example.com' + assert config.REGISTRY_AUTH == ('someuser', 'somepassword') + assert config.REPO_FILE_URL == 'https://example.com/repository.json' - assert config.DATA_DIR == '/some/data/dir' - assert config.AUTOSTART_FILE == '/some/data/dir/autostart' - assert config.REPO_BASE_URL == 'https://user:pass@example.com/spoc' - assert config.REPO_FILE_URL == 'https://user:pass@example.com/spoc/repository.json' +@patch('builtins.open', side_effect=FileNotFoundError('auth.json')) +def test_read_auth_fallback(auth_open): + config.read_auth() -def test_default_config(): - config_file = os.path.join(os.path.dirname(__file__), 'test_data/nonexistent') - config.reload(config_file) - - assert config.DATA_DIR == '/var/lib/spoc' - assert config.AUTOSTART_FILE == '/var/lib/spoc/autostart' - assert config.REPO_BASE_URL == 'https://localhost' + auth_open.assert_called_once_with(config.REGISTRY_AUTH_FILE) + assert config.REGISTRY_HOST == 'localhost' + assert config.REGISTRY_AUTH is None assert config.REPO_FILE_URL == 'https://localhost/repository.json' + +@patch('builtins.open', new_callable=mock_open) +def test_write_auth(auth_open): + config.write_auth('example.org', 'user', 'anotherpwd') + + auth_open.assert_called_once_with(config.REGISTRY_AUTH_FILE, 'w') + expected_data = '{"auths": {"example.org": {"auth": "dXNlcjphbm90aGVycHdk"}}}' + auth_open().write.assert_called_once_with(expected_data) + + assert config.REGISTRY_HOST == 'example.org' + assert config.REGISTRY_AUTH == ('user', 'anotherpwd') + assert config.REPO_FILE_URL == 'https://example.org/repository.json' diff --git a/tests/test_data/repository.json b/tests/test_data/repository.json index 1c572a6..aeb9338 100644 --- a/tests/test_data/repository.json +++ b/tests/test_data/repository.json @@ -13,19 +13,8 @@ "POSTGRES_USER": "someapp", "POSTGRES_PASSWORD": "someapp", "POSTGRES_DB": "someapp", - "POSTGRES_HOST": "someapp-postgres", - "DATABASE_URL": "postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}/${POSTGRES_DB}", - "APP_ADMIN_USER": "admin@example.com", - "APP_ADMIN_PASSWORD": "someapp123456", - "SECRET_KEY_BASE": "some_key", - "SMTP_USERNAME": "admin@example.com", - "SMTP_PASSWORD": "", - "SMTP_ADDRESS": "someapp-smtp", - "SMTP_DOMAIN": "example.com", - "MAPS_API_KEY": "", - "TWILIO_ACCOUNT_SID": "", - "TWILIO_AUTH_TOKEN": "", - "TWILIO_SENDER_NUMBER": "" + "POSTGRES_HOST": "postgres", + "DATABASE_URL": "postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}/${POSTGRES_DB}" }, "containers": { "someapp": { @@ -63,7 +52,6 @@ ], "volumes": { "conf": "/srv/web2py/applications/app/models", - "data-Spotter": "/srv/web2py/applications/app/modules/templates/Spotter", "data-databases": "/srv/web2py/applications/app/databases", "data-errors": "/srv/web2py/applications/app/errors", "data-sessions": "/srv/web2py/applications/app/sessions", diff --git a/tests/test_data/spoc.conf b/tests/test_data/spoc.conf deleted file mode 100644 index 96bd063..0000000 --- a/tests/test_data/spoc.conf +++ /dev/null @@ -1,3 +0,0 @@ -[spoc] -data-dir = /some/data/dir -repo-url = https://user:pass@example.com/spoc/ diff --git a/tests/test_data/test.env b/tests/test_data/test.env deleted file mode 100644 index be1eb9c..0000000 --- a/tests/test_data/test.env +++ /dev/null @@ -1,3 +0,0 @@ -RAILS_ENV=test -POSTGRES_PASSWORD=asdf=1234 -SOMEKEY=someval diff --git a/tests/test_data/test.env.json b/tests/test_data/test.env.json deleted file mode 100644 index 0bac85e..0000000 --- a/tests/test_data/test.env.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "RAILS_ENV": "test", - "POSTGRES_PASSWORD": "asdf=1234", - "SOMEKEY": "someval" -} diff --git a/tests/test_depsolver.py b/tests/test_depsolver.py index 0b33b6d..6bcbe91 100644 --- a/tests/test_depsolver.py +++ b/tests/test_depsolver.py @@ -65,7 +65,8 @@ def test_depsolver_complex(): # Order within the same batch (i.e. items not depending on each other) can be random assert list(sorted(resolved[:2])) == ['dep3', 'dep9'] - assert list(sorted(resolved[2:9])) == ['dep10', 'dep12', 'dep13', 'dep14', 'dep2', 'dep4', 'dep7'] + assert list(sorted(resolved[2:9])) == ['dep10', 'dep12', 'dep13', 'dep14', + 'dep2', 'dep4', 'dep7'] assert list(sorted(resolved[9:12])) == ['dep11', 'dep6', 'dep8'] assert list(sorted(resolved[12:])) == ['dep1', 'dep5'] diff --git a/tests/test_flock.py b/tests/test_flock.py index 2bf6819..ce5bb6d 100644 --- a/tests/test_flock.py +++ b/tests/test_flock.py @@ -1,29 +1,36 @@ import errno import fcntl import pytest -from unittest.mock import patch, call, mock_open +from unittest.mock import call, patch, mock_open +from spoc import config from spoc import flock -def fail_callback(pid): - print(f'Lock held by {pid}') - -@flock.locked('test.lock', fail_callback=fail_callback) +@flock.locked() def mock_func(): pass +@patch('builtins.open', new_callable=mock_open, read_data='foo\0arg1\0arg2\n') +def test_print_lock(cmdline_open, capsys): + flock.print_lock('123') + + cmdline_open.assert_called_once_with('/proc/123/cmdline') + captured = capsys.readouterr() + assert captured.err == 'Waiting for lock currently held by process 123 - foo arg1 arg2\n' + +@patch('spoc.flock.print_lock') @patch('fcntl.flock') @patch('time.sleep') @patch('os.getpid', return_value=1234) @patch('builtins.open', new_callable=mock_open) -def test_lock_success(lock_open, getpid, sleep, fcntl_flock): +def test_locked_success(lock_open, getpid, sleep, fcntl_flock, print_lock): mock_func() lock_open.assert_has_calls([ - call('test.lock', 'a'), + call(config.LOCK_FILE, 'a'), call().__enter__(), call().__exit__(None, None, None), - call('test.lock', 'r+'), + call(config.LOCK_FILE, 'r+'), call().__enter__(), call().truncate(), call().write('1234'), @@ -34,12 +41,14 @@ def test_lock_success(lock_open, getpid, sleep, fcntl_flock): fcntl_flock.assert_called_once_with(lock_open(), fcntl.LOCK_EX | fcntl.LOCK_NB) sleep.assert_not_called() getpid.assert_called_once() + print_lock.assert_not_called() +@patch('spoc.flock.print_lock') @patch('fcntl.flock') @patch('time.sleep') @patch('os.getpid', return_value=5678) @patch('builtins.open', new_callable=mock_open, read_data='1234') -def test_lock_fail(lock_open, getpid, sleep, fcntl_flock, capsys): +def test_locked_fail(lock_open, getpid, sleep, fcntl_flock, print_lock): fcntl_flock.side_effect = [ OSError(errno.EAGAIN, 'in use'), OSError(errno.EAGAIN, 'in use'), @@ -49,10 +58,10 @@ def test_lock_fail(lock_open, getpid, sleep, fcntl_flock, capsys): mock_func() lock_open.assert_has_calls([ - call('test.lock', 'a'), + call(config.LOCK_FILE, 'a'), call().__enter__(), call().__exit__(None, None, None), - call('test.lock', 'r+'), + call(config.LOCK_FILE, 'r+'), call().__enter__(), call().read(), call().seek(0), @@ -67,30 +76,29 @@ def test_lock_fail(lock_open, getpid, sleep, fcntl_flock, capsys): expected_sleep_call = call(0.1) assert sleep.call_args_list.count(expected_sleep_call) == 2 getpid.assert_called_once() + print_lock.assert_called_once_with('1234') - captured = capsys.readouterr() - assert captured.out == 'Lock held by 1234\n' - +@patch('spoc.flock.print_lock') @patch('fcntl.flock', side_effect=OSError(errno.EBADF, 'nope')) @patch('time.sleep') @patch('os.getpid', return_value=5678) @patch('builtins.open', new_callable=mock_open, read_data='1234') -def test_lock_error(lock_open, getpid, sleep, fcntl_flock): +def test_locked_error(lock_open, getpid, sleep, fcntl_flock, print_lock): with pytest.raises(OSError): mock_func() # Last call is # call().__exit__(, OSError(9, 'nope'), ) - # The exception can be captured above and checked as follows + # The exception can be passed by the context manager above and checked as follows # call().__exit__(ex.type, ex.value, ex.tb.tb_next.tb_next.tb_next) # but it may by CPython specific, and frankly, that tb_next chain looks horrible. # hence checking just the method and comparing the args with themselves last_exit_call_args = lock_open().__exit__.call_args_list[-1][0] lock_open.assert_has_calls([ - call('test.lock', 'a'), + call(config.LOCK_FILE, 'a'), call().__enter__(), call().__exit__(None, None, None), - call('test.lock', 'r+'), + call(config.LOCK_FILE, 'r+'), call().__enter__(), call().__exit__(*last_exit_call_args), ]) @@ -98,3 +106,4 @@ def test_lock_error(lock_open, getpid, sleep, fcntl_flock): fcntl_flock.assert_called_once_with(lock_open(), fcntl.LOCK_EX | fcntl.LOCK_NB) sleep.assert_not_called() getpid.assert_not_called() + print_lock.assert_not_called() diff --git a/tests/test_podman.py b/tests/test_podman.py index a2501a4..5506a33 100644 --- a/tests/test_podman.py +++ b/tests/test_podman.py @@ -7,92 +7,108 @@ from spoc import podman TEST_DATA_DIR = os.path.join(os.path.dirname(__file__), 'test_data') @patch('subprocess.run') -def test_get_apps(run): +def test_run(run): + process = podman.run(['foo', 'bar']) + + run.assert_called_once_with(['foo', 'bar'], check=True, env=podman.ENV) + assert process == run.return_value + +@patch('subprocess.run') +def test_out(run): + run.return_value.stdout = 'RESULT\n' + output = podman.out(['foo', 'bar'], arg1=123, arg2=True) + + run.assert_called_once_with(['foo', 'bar'], stdout=subprocess.PIPE, text=True, check=True, + env=podman.ENV, arg1=123, arg2=True) + assert output == 'RESULT' + +@patch('spoc.podman.out') +def test_get_apps(out): with open(os.path.join(TEST_DATA_DIR, 'podman_pod_ps.json')) as f: - run.return_value.stdout = f.read() + out.return_value = f.read() pods = podman.get_apps() expected_cmd = ['podman', 'pod', 'ps', '--format', 'json'] - run.assert_called_once_with(expected_cmd, check=True, stdout=subprocess.PIPE, text=True) + out.assert_called_once_with(expected_cmd) assert pods == {'someapp': '0.1', 'anotherapp': '0.2', 'yetanotherapp': '0.3'} -@patch('subprocess.run') -def test_get_volumes_for_app(run): +@patch('spoc.podman.out') +def test_get_volumes_for_app(out): with open(os.path.join(TEST_DATA_DIR, 'podman_volume_ls.json')) as f: - run.return_value.stdout = f.read() + out.return_value = f.read() volumes = podman.get_volumes_for_app('someapp') expected_cmd = ['podman', 'volume', 'ls', '--filter', 'label=spoc.app=someapp', '--format', 'json'] - run.asert_called_once_with(expected_cmd, check=True, stdout=subprocess.PIPE, text=True) + out.asert_called_once_with(expected_cmd) assert volumes == {'someapp-conf', 'someapp-data'} -@patch('subprocess.run') +@patch('spoc.podman.run') def test_start_pod(run): podman.start_pod('someapp') expected_cmd = ['podman', 'pod', 'start', 'someapp'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) -@patch('subprocess.run') +@patch('spoc.podman.run') def test_stop_pod(run): podman.stop_pod('someapp') expected_cmd = ['podman', 'pod', 'stop', '--ignore', 'someapp'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) -@patch('subprocess.run') -def test_get_pod_status(run): - run.return_value.stdout = 'RESULT\n' +@patch('spoc.podman.out') +def test_get_pod_status(out): + out.return_value = 'RESULT' status = podman.get_pod_status('someapp') expected_cmd = ['podman', 'pod', 'ps', '--filter', 'label=spoc.app=someapp'] - run.assert_called_once_with(expected_cmd, check=True, stdout=subprocess.PIPE, text=True) + out.assert_called_once_with(expected_cmd) assert status == 'RESULT' -@patch('subprocess.run') -def test_get_pod_status_all(run): - run.return_value.stdout = 'RESULT\n' +@patch('spoc.podman.out') +def test_get_pod_status_all(out): + out.return_value = 'RESULT' status = podman.get_pod_status() expected_cmd = ['podman', 'pod', 'ps'] - run.assert_called_once_with(expected_cmd, check=True, stdout=subprocess.PIPE, text=True) + out.assert_called_once_with(expected_cmd) assert status == 'RESULT' -@patch('subprocess.run') +@patch('spoc.podman.run') def test_create_volume(run): podman.create_volume('someapp', 'someapp-vol') expected_cmd = ['podman', 'volume', 'create', '--label', 'spoc.app=someapp', 'someapp-vol'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) -@patch('subprocess.run') +@patch('spoc.podman.run') def test_remove_volume(run): podman.remove_volume('someapp-vol') expected_cmd = ['podman', 'volume', 'rm', 'someapp-vol'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) -@patch('subprocess.run') +@patch('spoc.podman.run') def test_create_pod(run): podman.create_pod('someapp', '0.1') expected_cmd = ['podman', 'pod', 'create', '--name', 'someapp', '--label', 'spoc.app=someapp', '--label', 'spoc.version=0.1'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) @patch('spoc.podman.stop_pod') -@patch('subprocess.run') +@patch('spoc.podman.run') def test_remove_pod(run, stop_pod): podman.remove_pod('someapp') stop_pod.assert_called_once_with('someapp') expected_cmd = ['podman', 'pod', 'rm', '--ignore', 'someapp'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) -@patch('subprocess.run') +@patch('spoc.podman.run') def test_create_container(run): podman.create_container('someapp', 'someapp-cnt', 'example.com/someapp:0.23.6-210515', env_file='/var/lib/spoc/someapp.env', @@ -106,21 +122,21 @@ def test_create_container(run): '--volume', 'someapp-srv:/srv', '--add-host', 'cnt:127.0.0.1', '--add-host', 'cnt2:127.0.0.1', '--add-host', 'cnt3:127.0.0.1', 'example.com/someapp:0.23.6-210515'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) -@patch('subprocess.run') +@patch('spoc.podman.run') def test_create_container_minimal(run): podman.create_container('someapp', 'someapp-cnt', 'example.com/someapp:0.23.6-210515') expected_cmd = ['podman', 'container', 'create', '--name', 'someapp-cnt', '--pod', 'someapp', '--restart', 'unless-stopped', 'example.com/someapp:0.23.6-210515'] - run.assert_called_once_with(expected_cmd, check=True) + run.assert_called_once_with(expected_cmd) -@patch('subprocess.run') +@patch('spoc.podman.run') def test_prune(run): podman.prune() run.assert_has_calls([ - call(['podman', 'image', 'prune', '--all', '--force'], check=True), - call(['podman', 'volume', 'prune', '--force'], check=True), + call(['podman', 'image', 'prune', '--all', '--force']), + call(['podman', 'volume', 'prune', '--force']), ]) diff --git a/tests/test_repo.py b/tests/test_repo.py index 4353aaf..7ce9048 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -1,3 +1,4 @@ +import pytest from unittest.mock import patch, call from spoc import config @@ -8,7 +9,7 @@ from spoc import repo def test_load(req_get): repo.load() - req_get.assert_called_once_with(config.REPO_FILE_URL, timeout=5) + req_get.assert_called_once_with(config.REPO_FILE_URL, auth=config.REGISTRY_AUTH, timeout=5) req_get.return_value.raise_for_status.asert_called_once() req_get.return_value.json.asert_called_once() @@ -18,7 +19,7 @@ def test_load_twice_no_force(req_get): repo.load() repo.load() - req_get.assert_called_once_with(config.REPO_FILE_URL, timeout=5) + req_get.assert_called_once_with(config.REPO_FILE_URL, auth=config.REGISTRY_AUTH, timeout=5) req_get.return_value.raise_for_status.asert_called_once() req_get.return_value.json.asert_called_once() @@ -28,13 +29,24 @@ def test_load_twice_force(req_get): repo.load() repo.load(force=True) - expected_call = call(config.REPO_FILE_URL, timeout=5) + expected_call = call(config.REPO_FILE_URL, auth=config.REGISTRY_AUTH, timeout=5) assert req_get.call_args_list.count(expected_call) == 2 assert req_get.return_value.raise_for_status.call_count == 2 assert req_get.return_value.json.call_count == 2 +@patch('spoc.repo._data', {'someapp': {'version': '0.1'}}) +@patch('requests.get', side_effect=IOError()) +def test_load_empty_on_fail(req_get): + with pytest.raises(IOError): + repo.load(force=True) + + req_get.assert_called_once_with(config.REPO_FILE_URL, auth=config.REGISTRY_AUTH, timeout=5) + assert repo._data == {} # pylint: disable=protected-access + +@patch('spoc.repo._data', {'someapp': {'version': '0.1'}}) @patch('spoc.repo.load') def test_get_apps(repo_load): - repo.get_apps() + apps = repo.get_apps() repo_load.assert_called_once() + assert apps == {'someapp': {'version': '0.1'}} diff --git a/tests/test_spoc.py b/tests/test_spoc.py index a54ef9f..2aa6d0a 100644 --- a/tests/test_spoc.py +++ b/tests/test_spoc.py @@ -1,111 +1,220 @@ import pytest -from argparse import Namespace -from unittest.mock import call, mock_open, patch +from unittest.mock import call, patch import spoc -@patch('builtins.open', new_callable=mock_open, read_data='foo\0arg1\0arg2\n') -def test_print_lock(cmdline_open, capsys): - spoc.print_lock('123') +def test_apperror(): + exception = spoc.AppError('someapp') - cmdline_open.assert_called_once_with('/proc/123/cmdline') - captured = capsys.readouterr() - assert captured.out == 'Waiting for lock currently held by process 123 - foo arg1 arg2\n' + assert exception.app_name == 'someapp' -@patch('spoc.podman.get_apps', return_value={'anotherapp': '0.1', 'someapp': '0.1'}) -def test_listing_installed(get_apps, capsys): - spoc.listing('installed') +def test_appalreadyinstallederror(): + exception = spoc.AppAlreadyInstalledError('someapp') + + assert exception.app_name == 'someapp' + assert isinstance(exception, spoc.AppError) + +def test_appnotinstallederror(): + exception = spoc.AppNotInstalledError('someapp') + + assert exception.app_name == 'someapp' + assert isinstance(exception, spoc.AppError) + +def test_appnotinrepoerror(): + exception = spoc.AppNotInRepoError('someapp') + + assert exception.app_name == 'someapp' + assert isinstance(exception, spoc.AppError) + +def test_appnotupdateableerror(): + exception = spoc.AppNotUpdateableError('someapp') + + assert exception.app_name == 'someapp' + assert isinstance(exception, spoc.AppError) + +@patch('spoc.podman.get_apps', return_value={'someapp': '0.2', 'anotherapp': '0.1'}) +def test_list_installed(get_apps): + apps = spoc.list_installed() get_apps.assert_called_once() - # Order is important here - captured = capsys.readouterr() - assert captured.out == 'anotherapp 0.1\nsomeapp 0.1\n' + assert apps == {'anotherapp': '0.1', 'someapp': '0.2'} -@patch('spoc.repo.get_apps') -def test_listing_online(get_apps): - spoc.listing('online') +@patch('spoc.repo.get_apps', + return_value={'someapp': {'version': '0.2'}, 'anotherapp': {'version': '0.1'}}) +def test_list_online(get_apps): + apps = spoc.list_online() get_apps.assert_called_once() + assert apps == {'anotherapp': '0.1', 'someapp': '0.2'} + @patch('spoc.repo.get_apps', return_value={'someapp': {'version': '0.2'}, 'anotherapp': {'version': '0.1'}}) @patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) -def test_listing_updates(repo_get_apps, podman_get_apps, capsys): - spoc.listing('updates') +def test_list_updates(podman_get_apps, repo_get_apps): + apps = spoc.list_updates() repo_get_apps.assert_called_once() podman_get_apps.assert_called_once() - captured = capsys.readouterr() - assert captured.out == 'someapp 0.1 -> 0.2\n' - -@patch('spoc.repo.get_apps') -@patch('spoc.podman.get_apps') -def test_listing_invalid(repo_get_apps, podman_get_apps, capsys): - spoc.listing('invalid') - - repo_get_apps.assert_not_called() - podman_get_apps.assert_not_called() - - captured = capsys.readouterr() - assert captured.out == '' + assert apps == {'someapp': '0.1 -> 0.2'} +@patch('spoc.repo.get_apps', return_value={'someapp': {'version': '0.1'}}) +@patch('spoc.podman.get_apps', return_value={}) @patch('spoc.app.install') -def test_install(app_install): - spoc.install('someapp') +def test_install(app_install, podman_get_apps, repo_get_apps): + spoc.install.__wrapped__('someapp') + #spoc.install('someapp') + podman_get_apps.assert_called_once() + repo_get_apps.assert_called_once() app_install.assert_called_once_with('someapp') -@patch('spoc.app.update') -def test_update(app_update): - spoc.update('someapp') +@patch('spoc.repo.get_apps', return_value={'someapp': {'version': '0.1'}}) +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) +@patch('spoc.app.install') +def test_install_already_installed(app_install, podman_get_apps, repo_get_apps): + with pytest.raises(spoc.AppAlreadyInstalledError): + spoc.install.__wrapped__('someapp') + podman_get_apps.assert_called_once() + repo_get_apps.assert_not_called() + app_install.assert_not_called() + +@patch('spoc.repo.get_apps', return_value={}) +@patch('spoc.podman.get_apps', return_value={}) +@patch('spoc.app.install') +def test_install_not_in_repo(app_install, podman_get_apps, repo_get_apps): + with pytest.raises(spoc.AppNotInRepoError): + spoc.install.__wrapped__('someapp') + + podman_get_apps.assert_called_once() + repo_get_apps.assert_called_once() + app_install.assert_not_called() + +@patch('spoc.list_updates', return_value={'someapp': '0.1 -> 0.2'}) +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) +@patch('spoc.app.update') +def test_update(app_update, podman_get_apps, list_updates): + spoc.update.__wrapped__('someapp') + + podman_get_apps.assert_called_once() + list_updates.assert_called_once() app_update.assert_called_once_with('someapp') -@patch('spoc.app.uninstall') -def test_uninstall(app_uninstall): - spoc.uninstall('someapp') +@patch('spoc.list_updates', return_value={}) +@patch('spoc.podman.get_apps', return_value={}) +@patch('spoc.app.update') +def test_update_not_installed(app_update, podman_get_apps, list_updates): + with pytest.raises(spoc.AppNotInstalledError): + spoc.update.__wrapped__('someapp') + podman_get_apps.assert_called_once() + list_updates.assert_not_called() + app_update.assert_not_called() + +@patch('spoc.list_updates', return_value={}) +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) +@patch('spoc.app.update') +def test_update_not_updateable(app_update, podman_get_apps, list_updates): + with pytest.raises(spoc.AppNotUpdateableError): + spoc.update.__wrapped__('someapp') + + podman_get_apps.assert_called_once() + list_updates.assert_called_once() + app_update.assert_not_called() + +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) +@patch('spoc.app.uninstall') +def test_uninstall(app_uninstall, podman_get_apps): + spoc.uninstall.__wrapped__('someapp') + + podman_get_apps.assert_called_once() app_uninstall.assert_called_once_with('someapp') -@patch('spoc.podman.start_pod') -def test_start(start_pod): - spoc.start('someapp') +@patch('spoc.podman.get_apps', return_value={}) +@patch('spoc.app.uninstall') +def test_uninstall_not_installed(app_uninstall, podman_get_apps): + with pytest.raises(spoc.AppNotInstalledError): + spoc.uninstall.__wrapped__('someapp') + podman_get_apps.assert_called_once() + app_uninstall.assert_not_called() + +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) +@patch('spoc.podman.start_pod') +def test_start(start_pod, podman_get_apps): + spoc.start.__wrapped__('someapp') + + podman_get_apps.assert_called_once() start_pod.assert_called_once_with('someapp') -@patch('spoc.podman.stop_pod') -def test_stop(stop_pod): - spoc.stop('someapp') +@patch('spoc.podman.get_apps', return_value={}) +@patch('spoc.podman.start_pod') +def test_start_not_installed(start_pod, podman_get_apps): + with pytest.raises(spoc.AppNotInstalledError): + spoc.start.__wrapped__('someapp') + podman_get_apps.assert_called_once() + start_pod.assert_not_called() + +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) +@patch('spoc.podman.stop_pod') +def test_stop(stop_pod, podman_get_apps): + spoc.stop.__wrapped__('someapp') + + podman_get_apps.assert_called_once() stop_pod.assert_called_once_with('someapp') +@patch('spoc.podman.get_apps', return_value={}) +@patch('spoc.podman.stop_pod') +def test_stop_not_installed(stop_pod, podman_get_apps): + with pytest.raises(spoc.AppNotInstalledError): + spoc.stop.__wrapped__('someapp') + + podman_get_apps.assert_called_once() + stop_pod.assert_not_called() + +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) @patch('spoc.podman.get_pod_status', return_value='RESULT') -def test_status(get_pod_status, capsys): - spoc.status('someapp') +def test_status(get_pod_status, podman_get_apps): + status = spoc.status('someapp') + podman_get_apps.assert_called_once() get_pod_status.assert_called_once_with('someapp') - captured = capsys.readouterr() - assert captured.out == 'RESULT\n' + assert status == 'RESULT' -@pytest.mark.parametrize('value,expected',[ - ('1', True), - ('on', True), - ('Enable', True), - ('TRUE', True), - ('whatever', False), -]) +@patch('spoc.podman.get_apps', return_value={}) +@patch('spoc.podman.get_pod_status') +def test_status_not_installed(get_pod_status, podman_get_apps): + with pytest.raises(spoc.AppNotInstalledError): + spoc.status('someapp') + + podman_get_apps.assert_called_once() + get_pod_status.assert_not_called() + +@patch('spoc.podman.get_apps', return_value={'someapp': '0.1'}) @patch('spoc.autostart.set_app') -def test_set_autostart(set_app, value, expected): - spoc.set_autostart('someapp', value) +def test_set_autostart(set_app, podman_get_apps): + spoc.set_autostart.__wrapped__('someapp', True) - set_app.assert_called_once_with('someapp', expected) + podman_get_apps.assert_called_once() + set_app.assert_called_once_with('someapp', True) + +@patch('spoc.podman.get_apps', return_value={}) +@patch('spoc.autostart.set_app') +def test_set_autostart_not_installed(set_app, podman_get_apps): + with pytest.raises(spoc.AppNotInstalledError): + spoc.set_autostart.__wrapped__('someapp', True) + + podman_get_apps.assert_called_once() + set_app.assert_not_called() @patch('spoc.autostart.get_apps', return_value={'someapp', 'anotherapp'}) @patch('spoc.podman.start_pod') def test_start_autostarted(start_pod, get_apps): - spoc.start_autostarted() + spoc.start_autostarted.__wrapped__() get_apps.assert_called_once() start_pod.assert_has_calls([ @@ -116,7 +225,7 @@ def test_start_autostarted(start_pod, get_apps): @patch('spoc.podman.get_apps', return_value={'someapp': '0.1', 'anotherapp': '0.1'}) @patch('spoc.podman.stop_pod') def test_stop_all(stop_pod, get_apps): - spoc.stop_all() + spoc.stop_all.__wrapped__() get_apps.assert_called_once() stop_pod.assert_has_calls([ @@ -124,143 +233,16 @@ def test_stop_all(stop_pod, get_apps): call('anotherapp'), ], any_order=True) +@patch('spoc.config.write_auth') +@patch('spoc.repo.load') +def test_login(repo_load, write_auth): + spoc.login.__wrapped__('somehost', 'someuser', 'somepass') + + write_auth.assert_called_once_with('somehost', 'someuser', 'somepass') + repo_load.assert_called_once_with(force=True) + @patch('spoc.podman.prune') def test_prune(prune): - spoc.prune() + spoc.prune.__wrapped__() prune.assert_called_once() - -@patch('sys.argv', ['foo', 'list']) -@patch('spoc.listing') -def test_main_listing(listing): - spoc.main() - - listing.assert_called_once_with('installed') - -@patch('sys.argv', ['foo', 'list', 'online']) -@patch('spoc.listing') -def test_main_listing_online(listing): - spoc.main() - - listing.assert_called_once_with('online') - -@patch('sys.argv', ['foo', 'install', 'someapp']) -@patch('spoc.install') -def test_main_install(install): - spoc.main() - - install.assert_called_once_with('someapp') - -@patch('sys.argv', ['foo', 'update', 'someapp']) -@patch('spoc.update') -def test_main_update(update): - spoc.main() - - update.assert_called_once_with('someapp') - -@patch('sys.argv', ['foo', 'uninstall', 'someapp']) -@patch('spoc.uninstall') -def test_main_uninstall(uninstall): - spoc.main() - - uninstall.assert_called_once_with('someapp') - -@patch('sys.argv', ['foo', 'start', 'someapp']) -@patch('spoc.start') -def test_main_start(start): - spoc.main() - - start.assert_called_once_with('someapp') - -@patch('sys.argv', ['foo', 'stop', 'someapp']) -@patch('spoc.stop') -def test_main_stop(stop): - spoc.main() - - stop.assert_called_once_with('someapp') - -@patch('sys.argv', ['foo', 'status', 'someapp']) -@patch('spoc.status') -def test_main_status(status): - spoc.main() - - status.assert_called_once_with('someapp') - -@patch('sys.argv', ['foo', 'status']) -@patch('spoc.status') -def test_main_status_all(status): - spoc.main() - - status.assert_called_once_with(None) - -@patch('sys.argv', ['foo', 'autostart', 'someapp', 'on']) -@patch('spoc.set_autostart') -def test_main_autostart(autostart): - spoc.main() - - autostart.assert_called_once_with('someapp', 'on') - -@patch('sys.argv', ['foo', 'start-autostarted']) -@patch('spoc.start_autostarted') -def test_main_start_autostarted(start_autostarted): - spoc.main() - - start_autostarted.assert_called_once() - -@patch('sys.argv', ['foo', 'stop-all']) -@patch('spoc.stop_all') -def test_main_stop_all(stop_all): - spoc.main() - - stop_all.assert_called_once() - -@patch('sys.argv', ['foo', 'prune']) -@patch('spoc.prune') -def test_main_prune(prune): - spoc.main() - - prune.assert_called_once() - -@patch('spoc.parse_args', return_value=Namespace(action=None)) -@patch('spoc.listing') -@patch('spoc.install') -@patch('spoc.update') -@patch('spoc.uninstall') -@patch('spoc.start') -@patch('spoc.stop') -@patch('spoc.status') -@patch('spoc.start_autostarted') -@patch('spoc.stop_all') -@patch('spoc.prune') -def test_main_invalid(prune, stop_all, start_autostarted, status, stop, start, - uninstall, update, install, listing, parse_args): - spoc.main() - - parse_args.assert_called_once() - listing.assert_not_called() - install.assert_not_called() - update.assert_not_called() - uninstall.assert_not_called() - start.assert_not_called() - stop.assert_not_called() - status.assert_not_called() - start_autostarted.assert_not_called() - stop_all.assert_not_called() - prune.assert_not_called() - -@pytest.mark.parametrize('argv', [ - ['list', 'invalid'], - ['install'], - ['update'], - ['uninstall'], - ['start'], - ['stop'], - ['autostart'], - ['autostart', 'someapp'], - ['invalid'], -]) -def test_main_systemexit(argv): - argv.insert(0, 'foo') - with patch('sys.argv', argv): - with pytest.raises(SystemExit): - spoc.main()