Skip to content
Snippets Groups Projects
Verified Commit f173e59c authored by Karel Koci's avatar Karel Koci :metal:
Browse files

nsfarm/setup/utils: add directory and file deployment setups

These setups are handy in general. They created files and diretories
that are automatically removed on revert.
parent cd2d9f85
Branches
Tags
No related merge requests found
"""Generic utilities that work on all Unix-like platforms as they are based on pure shell operations.
"""
import pathlib
import random
import string
import typing
......@@ -8,6 +9,74 @@ from .. import cli
from ._setup import Setup as _Setup
class Dir(_Setup):
"""Make sure that directory exists."""
def __init__(self, shell: cli.Shell, path: typing.Union[str, pathlib.PurePosixPath], expect_empty: bool = True):
"""shell: shell used for setup
path: path to directory to be created
expect_empty: if we expect to be empty on removal (rmdir) or if we should remove content (rm -rf)
"""
self._sh = shell
self._path = pathlib.PurePosixPath(path)
self._existed = None
self.expect_empty = expect_empty
def prepare(self):
self._sh.run(f'( path=\'{self._path}\'; while [ ! -d "$path" ]; do path="${{path%/*}}"; done; echo "$path" )')
self._existed = pathlib.PurePosixPath(self._sh.output.strip())
if self._existed != self._path: # No need to create existing directory
self._sh.run(f"mkdir -p '{self._path}'")
def revert(self):
if self._existed == self._path:
return
if self.expect_empty:
self._sh.run(f"( cd '{self._existed}' && rmdir -p '{self._path.relative_to(self._existed)}' )")
else:
to_remove = pathlib.PurePosixPath(*self._path.parts[: len(self._existed.parts) + 1])
self._sh.run(f"rm -rf '{to_remove}'")
class DeployFile(_Setup):
"""Simple way to deploy file trough shell instance."""
def __init__(
self,
shell: cli.Shell,
path: typing.Union[str, pathlib.PurePosixPath],
content: typing.Union[str, bytes],
mkdir: bool = False,
):
"""shell: shell used for setup
path: path where file should be deployed
content: content of file to be deployed
mkdir: if upper directory should be created or not
"""
self._sh = shell
self._path = pathlib.PurePosixPath(path)
self._content = content
self._previous = None
self._dir = Dir(shell, self._path.parent) if mkdir else None
def prepare(self):
self._previous = self._sh.bin_read(self._path, expect_exist=False)
if self._previous is None and self._dir is not None:
self._dir.prepare()
if isinstance(self._content, str):
self._sh.txt_write(self._path, self._content)
else:
self._sh.bin_write(self._path, self._content)
def revert(self):
if self._previous is None:
self._sh.run(f"rm -f '{self._path}'")
if self._dir is not None:
self._dir.revert()
else:
self._sh.bin_write(self._path, self._previous)
class RootPassword(_Setup):
"""Sets given or random password as the one for root."""
......
......@@ -22,6 +22,46 @@ class Common:
def fixture_shell(self, container):
return Shell(container.pexpect())
def test_dir(self, shell):
"""Check that Dir setup correctly manages directories."""
path = "/tmp/foo/fee/faa"
shell.run(f"[ ! -d '{path}' ]")
with utils.Dir(shell, path) as _:
shell.run(f"[ -d '{path}' ]")
shell.run(f"[ ! -d '{path}' ]")
def test_dir_existed(self, shell):
"""Check that Dir setup does nothing if directory already exists."""
path = "/tmp"
shell.run(f"[ -d '{path}' ]")
with utils.Dir(shell, path) as _:
shell.run(f"[ -d '{path}' ]")
shell.run(f"[ -d '{path}' ]")
def test_deployfile(self, shell):
"""Check that DeployFile works if there is no previous file."""
path = "/tmp/deployfile/test-file"
content = "Some content"
shell.run(f"[ ! -f '{path}' ]")
with utils.DeployFile(shell, path, content, mkdir=True) as _:
shell.run(f"cat '{path}'")
assert shell.output.strip() == content
shell.run(f"[ ! -f '{path}' ]")
def test_deployfile_exist(self, shell):
"""Check that DeployFile works if there is previous file. We do nested usage here to simulate that."""
path = "/tmp/deployfile/test-file"
content1 = "Some content"
content2 = "Different content"
shell.run(f"[ ! -f '{path}' ]")
with utils.DeployFile(shell, path, content1, mkdir=True) as _:
shell.run(f"cat '{path}'")
assert shell.output.strip() == content1
with utils.DeployFile(shell, path, content2, mkdir=True) as _:
shell.run(f"cat '{path}'")
assert shell.output.strip() == content2
shell.run(f"[ ! -f '{path}' ]")
def test_password(self, shell):
"""Check RootPassword setup."""
password = "ILikeWhenPlansPanOut"
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment