Spotter-Cluster/spoc
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Further work - add auth, split CLI, simplify locking

Browse Source
This commit is contained in:
Disassembler 2021-07-11 00:58:28 +02:00
parent 7004b0767e
commit 0b585dee0d
No known key found for this signature in database
GPG Key ID: 524BD33A0EE29499
24 changed files with 989 additions and 499 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
service/spoc.conf
View File

@ -1,3 +0,0 @@
[spoc]
data-dir = /var/lib/spoc/
repo-url = https://repo.spotter.cz/spoc/

3
setup.cfg
View File

@ -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

176
src/spoc/__init__.py
View File

@ -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()

12
src/spoc/app.py
View File

@ -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):

2
src/spoc/autostart.py
View File

@ -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:

45
src/spoc/config.py
View File

@ -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()

8
src/spoc/depsolver.py
View File

@ -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)

36
src/spoc/flock.py
View File

@ -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

53
src/spoc/podman.py
View File

@ -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'])

5
src/spoc/repo.py
View File

@ -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()

168
src/spoc_cli.py Normal file
View File

@ -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)

24
tests/test_app.py
View File

@ -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')

8
tests/test_autostart.py
View File

@ -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()

343
tests/test_cli.py Normal file
View File

@ -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()

45
tests/test_config.py
View File

@ -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'

16
tests/test_data/repository.json
View File

@ -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",

3
tests/test_data/spoc.conf
View File

@ -1,3 +0,0 @@
[spoc]
data-dir = /some/data/dir
repo-url = https://user:pass@example.com/spoc/

3
tests/test_data/test.env
View File

@ -1,3 +0,0 @@
RAILS_ENV=test
POSTGRES_PASSWORD=asdf=1234
SOMEKEY=someval

5
tests/test_data/test.env.json
View File

@ -1,5 +0,0 @@
{
"RAILS_ENV": "test",
"POSTGRES_PASSWORD": "asdf=1234",
"SOMEKEY": "someval"
}

3
tests/test_depsolver.py
View File

@ -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']

45
tests/test_flock.py
View File

@ -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__(<class 'OSError'>, OSError(9, 'nope'), <traceback object at 0xaddress>)
# 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()

84
tests/test_podman.py
View File

@ -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']),
])

20
tests/test_repo.py
View File

@ -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'}}

378
tests/test_spoc.py
View File

@ -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()