spoc/tests/test_podman.py

import subprocess
import os
from unittest.mock import patch, call

from spoc import podman

TEST_DATA_DIR = os.path.join(os.path.dirname(__file__), 'test_data')

@patch('subprocess.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', return_value='     0     1000        1\n     1   100000    65536')
def test_get_subuidgid(out):
    subuidgid = podman.get_subuidgid()

    assert subuidgid == (100000, 100000)
    out.assert_has_calls([
        call(['su', '-', 'spoc', '-c', 'podman unshare cat /proc/self/uid_map']),
        call(['su', '-', 'spoc', '-c', 'podman unshare cat /proc/self/gid_map']),
    ])

@patch('spoc.podman.out', return_value='')
def test_get_subuidgid_no_id(out):
    subuidgid = podman.get_subuidgid()

    assert subuidgid == (0, 0)
    assert out.call_count == 2

@patch('spoc.podman.out')
def test_get_apps(out):
    with open(os.path.join(TEST_DATA_DIR, 'podman_pod_ps.json'), encoding='utf-8') as f:
        out.return_value = f.read()

    pods = podman.get_apps()

    expected_cmd = ['podman', 'pod', 'ps', '--format', 'json']
    out.assert_called_once_with(expected_cmd)
    assert pods == {'someapp': '0.1', 'anotherapp': '0.2', 'yetanotherapp': '0.3'}

@patch('spoc.podman.out')
def test_get_volumes_for_app(out):
    with open(os.path.join(TEST_DATA_DIR, 'podman_volume_ls.json'), encoding='utf-8') as f:
        out.return_value = f.read()

    volumes = podman.get_volumes_for_app('someapp')

    expected_cmd = ['podman', 'volume', 'ls', '--filter', 'label=spoc.app=someapp',
                    '--format', 'json']
    out.assert_called_once_with(expected_cmd)
    assert volumes == {'someapp-conf', 'someapp-data'}

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

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

@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']
    out.assert_called_once_with(expected_cmd)
    assert status == 'RESULT'

@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', '--filter', 'label=spoc.app']
    out.assert_called_once_with(expected_cmd)
    assert status == 'RESULT'

@patch('spoc.podman.run')
@patch('spoc.podman.get_subuidgid', return_value=(100000, 100000))
def test_create_volume(get_subuidgid, run):
    podman.create_volume('someapp', 'someapp-vol')

    expected_cmd = ['podman', 'volume', 'create',
                    '--opt', 'o=uid=100000,gid=100000',
                    '--label', 'spoc.app=someapp', 'someapp-vol']
    get_subuidgid.assert_called_once()
    run.assert_called_once_with(expected_cmd)

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

@patch('spoc.podman.run')
def test_create_pod(run):
    podman.create_pod('someapp', '0.1')

    expected_cmd = ['podman', 'pod', 'create', '--name', 'someapp',
                    '--subuidname', 'spoc', '--subgidname', 'spoc',
                    '--label', 'spoc.app=someapp', '--label', 'spoc.version=0.1']
    run.assert_called_once_with(expected_cmd)

@patch('spoc.podman.stop_pod')
@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)

@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',
                            volumes={'someapp-srv': '/srv', 'someapp-mnt': '/mnt'},
                            requires={'someapp-cnt3', 'someapp-cnt2'},
                            hosts={'cnt2', 'cnt3', 'cnt'})

    expected_cmd = ['podman', 'container', 'create', '--name', 'someapp-cnt', '--pod', 'someapp',
                    '--subuidname', 'spoc', '--subgidname', 'spoc',
                    '--restart', 'unless-stopped', '--env-file', '/var/lib/spoc/someapp.env',
                    '--requires', 'someapp-cnt2,someapp-cnt3', '--volume', 'someapp-mnt:/mnt',
                    '--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)

@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',
                    '--subuidname', 'spoc', '--subgidname', 'spoc',
                    '--restart', 'unless-stopped', 'example.com/someapp:0.23.6-210515']
    run.assert_called_once_with(expected_cmd)

@patch('spoc.podman.run')
def test_prune(run):
    podman.prune()

    run.assert_has_calls([
        call(['podman', 'image', 'prune', '--all', '--force', '--volumes']),
    ])