import pytest
from render import Render
from formatter import get_hashed_name_for_volume
@pytest.fixture
def mock_values():
return {
"images": {
"test_image": {
"repository": "nginx",
"tag": "latest",
}
},
}
def test_add_volume_invalid_type(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
with pytest.raises(Exception):
c1.add_storage("/some/path", {"type": "invalid_type"})
def test_add_volume_empty_mount_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
with pytest.raises(Exception):
c1.add_storage("", {"type": "tmpfs"})
def test_add_volume_duplicate_mount_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.add_storage("/some/path", {"type": "tmpfs"})
with pytest.raises(Exception):
c1.add_storage("/some/path", {"type": "tmpfs"})
def test_add_volume_host_path_invalid_propagation(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {
"type": "host_path",
"host_path_config": {"path": "/mnt/test", "propagation": "invalid_propagation"},
}
with pytest.raises(Exception):
c1.add_storage("/some/path", host_path_config)
def test_add_host_path_volume_no_host_path_config(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path"}
with pytest.raises(Exception):
c1.add_storage("/some/path", host_path_config)
def test_add_host_path_volume_no_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "host_path_config": {"path": ""}}
with pytest.raises(Exception):
c1.add_storage("/some/path", host_path_config)
def test_add_host_path_with_acl_no_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "host_path_config": {"acl_enable": True, "acl": {"path": ""}}}
with pytest.raises(Exception):
c1.add_storage("/some/path", host_path_config)
def test_add_host_path_volume_mount(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "host_path_config": {"path": "/mnt/test"}}
c1.add_storage("/some/path", host_path_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test",
"target": "/some/path",
"read_only": False,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_host_path_volume_mount_with_acl(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {
"type": "host_path",
"host_path_config": {"path": "/mnt/test", "acl_enable": True, "acl": {"path": "/mnt/test/acl"}},
}
c1.add_storage("/some/path", host_path_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test/acl",
"target": "/some/path",
"read_only": False,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_host_path_volume_mount_with_propagation(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "host_path_config": {"path": "/mnt/test", "propagation": "slave"}}
c1.add_storage("/some/path", host_path_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test",
"target": "/some/path",
"read_only": False,
"bind": {"create_host_path": False, "propagation": "slave"},
}
]
def test_add_host_path_volume_mount_with_create_host_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "host_path_config": {"path": "/mnt/test", "create_host_path": True}}
c1.add_storage("/some/path", host_path_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test",
"target": "/some/path",
"read_only": False,
"bind": {"create_host_path": True, "propagation": "rprivate"},
}
]
def test_add_host_path_volume_mount_with_read_only(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "read_only": True, "host_path_config": {"path": "/mnt/test"}}
c1.add_storage("/some/path", host_path_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test",
"target": "/some/path",
"read_only": True,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_ix_volume_invalid_dataset_name(mock_values):
mock_values["ix_volumes"] = {"test_dataset": "/mnt/test"}
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
ix_volume_config = {"type": "ix_volume", "ix_volume_config": {"dataset_name": "invalid_dataset"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", ix_volume_config)
def test_add_ix_volume_no_ix_volume_config(mock_values):
mock_values["ix_volumes"] = {"test_dataset": "/mnt/test"}
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
ix_volume_config = {"type": "ix_volume"}
with pytest.raises(Exception):
c1.add_storage("/some/path", ix_volume_config)
def test_add_ix_volume_volume_mount(mock_values):
mock_values["ix_volumes"] = {"test_dataset": "/mnt/test"}
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
ix_volume_config = {"type": "ix_volume", "ix_volume_config": {"dataset_name": "test_dataset"}}
c1.add_storage("/some/path", ix_volume_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test",
"target": "/some/path",
"read_only": False,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_ix_volume_volume_mount_with_options(mock_values):
mock_values["ix_volumes"] = {"test_dataset": "/mnt/test"}
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
ix_volume_config = {
"type": "ix_volume",
"ix_volume_config": {"dataset_name": "test_dataset", "propagation": "rslave", "create_host_path": True},
}
c1.add_storage("/some/path", ix_volume_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test",
"target": "/some/path",
"read_only": False,
"bind": {"create_host_path": True, "propagation": "rslave"},
}
]
def test_cifs_volume_missing_server(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {"type": "cifs", "cifs_config": {"path": "/path", "username": "user", "password": "password"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_missing_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {"type": "cifs", "cifs_config": {"server": "server", "username": "user", "password": "password"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_missing_username(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {"type": "cifs", "cifs_config": {"server": "server", "path": "/path", "password": "password"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_missing_password(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {"type": "cifs", "cifs_config": {"server": "server", "path": "/path", "username": "user"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_without_cifs_config(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {"type": "cifs"}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_duplicate_option(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {
"type": "cifs",
"cifs_config": {
"server": "server",
"path": "/path",
"username": "user",
"password": "pas$word",
"options": ["verbose=true", "verbose=true"],
},
}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_disallowed_option(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {
"type": "cifs",
"cifs_config": {
"server": "server",
"path": "/path",
"username": "user",
"password": "pas$word",
"options": ["user=username"],
},
}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_invalid_options(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {
"type": "cifs",
"cifs_config": {
"server": "server",
"path": "/path",
"username": "user",
"password": "pas$word",
"options": {"verbose": True},
},
}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_cifs_volume_invalid_options2(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_config = {
"type": "cifs",
"cifs_config": {
"server": "server",
"path": "/path",
"username": "user",
"password": "pas$word",
"options": [{"verbose": True}],
},
}
with pytest.raises(Exception):
c1.add_storage("/some/path", cifs_config)
def test_add_cifs_volume(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_inner_config = {"server": "server", "path": "/path", "username": "user", "password": "pas$word"}
cifs_config = {"type": "cifs", "cifs_config": cifs_inner_config}
c1.add_storage("/some/path", cifs_config)
output = render.render()
vol_name = get_hashed_name_for_volume("cifs", cifs_inner_config)
assert output["volumes"] == {
vol_name: {
"driver_opts": {"type": "cifs", "device": "//server/path", "o": "noperm,password=pas$$word,user=user"}
}
}
assert output["services"]["test_container"]["volumes"] == [
{"type": "volume", "source": vol_name, "target": "/some/path", "read_only": False, "volume": {"nocopy": False}}
]
def test_cifs_volume_with_options(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
cifs_inner_config = {
"server": "server",
"path": "/path",
"username": "user",
"password": "pas$word",
"options": ["vers=3.0", "verbose=true"],
}
cifs_config = {"type": "cifs", "cifs_config": cifs_inner_config}
c1.add_storage("/some/path", cifs_config)
output = render.render()
vol_name = get_hashed_name_for_volume("cifs", cifs_inner_config)
assert output["volumes"] == {
vol_name: {
"driver_opts": {
"type": "cifs",
"device": "//server/path",
"o": "noperm,password=pas$$word,user=user,verbose=true,vers=3.0",
}
}
}
assert output["services"]["test_container"]["volumes"] == [
{"type": "volume", "source": vol_name, "target": "/some/path", "read_only": False, "volume": {"nocopy": False}}
]
def test_nfs_volume_missing_server(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_config = {"type": "nfs", "nfs_config": {"path": "/path"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", nfs_config)
def test_nfs_volume_missing_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_config = {"type": "nfs", "nfs_config": {"server": "server"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", nfs_config)
def test_nfs_volume_without_nfs_config(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_config = {"type": "nfs"}
with pytest.raises(Exception):
c1.add_storage("/some/path", nfs_config)
def test_nfs_volume_duplicate_option(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_config = {
"type": "nfs",
"nfs_config": {"server": "server", "path": "/path", "options": ["verbose=true", "verbose=true"]},
}
with pytest.raises(Exception):
c1.add_storage("/some/path", nfs_config)
def test_nfs_volume_disallowed_option(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_config = {"type": "nfs", "nfs_config": {"server": "server", "path": "/path", "options": ["addr=server"]}}
with pytest.raises(Exception):
c1.add_storage("/some/path", nfs_config)
def test_nfs_volume_invalid_options(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_config = {"type": "nfs", "nfs_config": {"server": "server", "path": "/path", "options": {"verbose": True}}}
with pytest.raises(Exception):
c1.add_storage("/some/path", nfs_config)
def test_nfs_volume_invalid_options2(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_config = {"type": "nfs", "nfs_config": {"server": "server", "path": "/path", "options": [{"verbose": True}]}}
with pytest.raises(Exception):
c1.add_storage("/some/path", nfs_config)
def test_add_nfs_volume(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_inner_config = {"server": "server", "path": "/path"}
nfs_config = {"type": "nfs", "nfs_config": nfs_inner_config}
c1.add_storage("/some/path", nfs_config)
output = render.render()
vol_name = get_hashed_name_for_volume("nfs", nfs_inner_config)
assert output["volumes"] == {vol_name: {"driver_opts": {"type": "nfs", "device": ":/path", "o": "addr=server"}}}
assert output["services"]["test_container"]["volumes"] == [
{"type": "volume", "source": vol_name, "target": "/some/path", "read_only": False, "volume": {"nocopy": False}}
]
def test_nfs_volume_with_options(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
nfs_inner_config = {"server": "server", "path": "/path", "options": ["vers=3.0", "verbose=true"]}
nfs_config = {"type": "nfs", "nfs_config": nfs_inner_config}
c1.add_storage("/some/path", nfs_config)
output = render.render()
vol_name = get_hashed_name_for_volume("nfs", nfs_inner_config)
assert output["volumes"] == {
vol_name: {
"driver_opts": {
"type": "nfs",
"device": ":/path",
"o": "addr=server,verbose=true,vers=3.0",
}
}
}
assert output["services"]["test_container"]["volumes"] == [
{"type": "volume", "source": vol_name, "target": "/some/path", "read_only": False, "volume": {"nocopy": False}}
]
def test_tmpfs_invalid_size(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "tmpfs", "tmpfs_config": {"size": "2"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", vol_config)
def test_tmpfs_zero_size(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "tmpfs", "tmpfs_config": {"size": 0}}
with pytest.raises(Exception):
c1.add_storage("/some/path", vol_config)
def test_tmpfs_invalid_mode(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "tmpfs", "tmpfs_config": {"mode": "invalid"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", vol_config)
def test_tmpfs_volume(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.add_storage("/some/path", {"type": "tmpfs"})
c1.add_storage("/some/other/path", {"type": "tmpfs", "tmpfs_config": {"size": 100}})
c1.add_storage(
"/some/other/path2", {"type": "tmpfs", "tmpfs_config": {"size": 100, "mode": "0777", "uid": 1000, "gid": 1000}}
)
output = render.render()
assert output["services"]["test_container"]["tmpfs"] == [
"/some/other/path2:gid=1000,mode=0777,size=104857600,uid=1000",
"/some/other/path:size=104857600",
"/some/path",
]
def test_add_tmpfs_with_existing_volume(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.add_storage("/some/path", {"type": "volume", "volume_config": {"volume_name": "test_volume"}})
with pytest.raises(Exception):
c1.add_storage("/some/path", {"type": "tmpfs", "tmpfs_config": {"size": 100}})
def test_add_volume_with_existing_tmpfs(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.add_storage("/some/path", {"type": "tmpfs", "tmpfs_config": {"size": 100}})
with pytest.raises(Exception):
c1.add_storage("/some/path", {"type": "volume", "volume_config": {"volume_name": "test_volume"}})
def test_temporary_volume(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "temporary", "volume_config": {"volume_name": "test_temp_volume"}}
c1.add_storage("/some/path", vol_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"source": "test_temp_volume",
"type": "volume",
"target": "/some/path",
"read_only": False,
"volume": {"nocopy": False},
}
]
def test_docker_volume_missing_config(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "volume", "volume_config": {}}
with pytest.raises(Exception):
c1.add_storage("/some/path", vol_config)
def test_docker_volume_missing_volume_name(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "volume", "volume_config": {"volume_name": ""}}
with pytest.raises(Exception):
c1.add_storage("/some/path", vol_config)
def test_docker_volume(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "volume", "volume_config": {"volume_name": "test_volume"}}
c1.add_storage("/some/path", vol_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "volume",
"source": "test_volume",
"target": "/some/path",
"read_only": False,
"volume": {"nocopy": False},
}
]
assert output["volumes"] == {"test_volume": {}}
def test_anonymous_volume(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
vol_config = {"type": "anonymous", "volume_config": {"nocopy": True}}
c1.add_storage("/some/path", vol_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{"type": "volume", "target": "/some/path", "read_only": False, "volume": {"nocopy": True}}
]
assert "volumes" not in output
def test_add_udev(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.add_udev()
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/run/udev",
"target": "/run/udev",
"read_only": True,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_udev_not_read_only(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.add_udev(read_only=False)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/run/udev",
"target": "/run/udev",
"read_only": False,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_docker_socket(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.storage._add_docker_socket(mount_path="/var/run/docker.sock")
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/var/run/docker.sock",
"target": "/var/run/docker.sock",
"read_only": True,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_docker_socket_not_read_only(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.storage._add_docker_socket(read_only=False, mount_path="/var/run/docker.sock")
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/var/run/docker.sock",
"target": "/var/run/docker.sock",
"read_only": False,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_add_docker_socket_mount_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
c1.storage._add_docker_socket(mount_path="/some/path")
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/var/run/docker.sock",
"target": "/some/path",
"read_only": True,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]
def test_host_path_with_disallowed_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "host_path_config": {"path": "/mnt"}}
with pytest.raises(Exception):
c1.add_storage("/some/path", host_path_config)
def test_host_path_without_disallowed_path(mock_values):
render = Render(mock_values)
c1 = render.add_container("test_container", "test_image")
c1.healthcheck.disable()
host_path_config = {"type": "host_path", "host_path_config": {"path": "/mnt/test"}}
c1.add_storage("/mnt", host_path_config)
output = render.render()
assert output["services"]["test_container"]["volumes"] == [
{
"type": "bind",
"source": "/mnt/test",
"target": "/mnt",
"read_only": False,
"bind": {"create_host_path": False, "propagation": "rprivate"},
}
]