Skip to content
Permalink
Browse files

Add some more packaging tests

Change-Id: I260d8e8fcfc56b91fc9691df8f081e45fb2cf335
  • Loading branch information
LarsMichelsen committed Nov 29, 2019
1 parent 99e129b commit b05c8ecd681863404d2756b440edb5d99123193d
Showing with 171 additions and 23 deletions.
  1. +3 −3 cmk/utils/packaging.py
  2. +3 −1 cmk_base/packaging.py
  3. +165 −19 tests/unit/cmk/utils/test_packaging.py
@@ -136,8 +136,8 @@ def release_package(pacname):
_remove_package_info(pacname)


def create_mkp_file(package, file_name=None, file_object=None):
# type: (PackageInfo, Optional[Text], BinaryIO) -> None
def create_mkp_file(package, file_object=None):
# type: (PackageInfo, BinaryIO) -> None
package["version.packaged"] = cmk.__version__

def create_tar_info(filename, size):
@@ -151,7 +151,7 @@ def create_tar_info(filename, size):
info.name = filename
return info

tar = tarfile.open(name=file_name, fileobj=file_object, mode="w:gz")
tar = tarfile.open(fileobj=file_object, mode="w:gz")

# add the regular info file (Python format)
info_file = StringIO(pprint.pformat(package))
@@ -262,8 +262,10 @@ def package_pack(args):
if not package:
raise PackageException("Package %s not existing or corrupt." % pacname)
tarfilename = "%s-%s%s" % (pacname, package["version"], _pac_ext)

logger.log(VERBOSE, "Packing %s into %s...", pacname, tarfilename)
create_mkp_file(package, file_name=tarfilename)
with Path(tarfilename).open("wb") as f:
create_mkp_file(package, f)
logger.log(VERBOSE, "Successfully created %s", tarfilename)


@@ -1,7 +1,11 @@
#!/usr/bin/env python

import pytest # type: ignore
import shutil
import tarfile
import ast
import json
from io import BytesIO
import pytest # type: ignore
from pathlib2 import Path

import cmk.utils.paths
@@ -15,6 +19,17 @@ def package_dir():
shutil.rmtree(str(packaging.package_dir()))


@pytest.fixture(autouse=True)
def clean_dirs():
for part in packaging.get_package_parts():
Path(part.path).mkdir(parents=True, exist_ok=True)

yield

for part in packaging.get_package_parts():
shutil.rmtree(part.path)


def test_package_parts():
assert packaging.get_package_parts() == [
packaging.PackagePart('checks', 'Checks', 'local/share/check_mk/checks'),
@@ -83,12 +98,21 @@ def test_get_package_parts():


def _create_simple_test_package(pacname):
cmk.utils.paths.local_checks_dir.mkdir(parents=True, exist_ok=True)
with cmk.utils.paths.local_checks_dir.joinpath(pacname).open("w", encoding="utf-8") as f:
f.write(u"lala\n")

_create_test_file(pacname)
package_info = packaging.get_initial_package_info(pacname)

package_info["files"] = {
"checks": [pacname],
}

packaging.create_package(package_info)
return packaging.read_package_info("aaa")


def _create_test_file(name):
check_path = cmk.utils.paths.local_checks_dir.joinpath(name)
with check_path.open("w", encoding="utf-8") as f:
f.write(u"lala\n")


def test_create_package():
@@ -125,8 +149,8 @@ def test_edit_package():
new_package_info = packaging.get_initial_package_info("aaa")
new_package_info["version"] = "2.0"

_create_simple_test_package("aaa")
assert packaging.read_package_info("aaa")["version"] == "1.0"
package_info = _create_simple_test_package("aaa")
assert package_info["version"] == "1.0"

packaging.edit_package("aaa", new_package_info)

@@ -153,20 +177,142 @@ def test_edit_package_rename_conflict():
packaging.edit_package("aaa", new_package_info)


# TODO
#def test_install_package_by_path()
#def test_install_package()
#def test_release_package():
#def test_create_mkp_file():
#def test_remove_package()
def test_install_package():
# Create
_create_simple_test_package("aaa")
package_info = packaging.read_package_info("aaa")

# Build MKP in memory
mkp = BytesIO()
packaging.create_mkp_file(package_info, mkp)
mkp.seek(0)

# Remove files from local hierarchy
packaging.remove_package(package_info)
assert packaging._package_exists("aaa") is False

# And now install the package from memory
packaging.install_package(mkp)

# Check result
assert packaging._package_exists("aaa") is True
package_info = packaging.read_package_info("aaa")
assert package_info["version"] == "1.0"
assert package_info["files"]["checks"] == ["aaa"]
assert cmk.utils.paths.local_checks_dir.joinpath("aaa").exists()


def test_install_package_by_path(tmp_path):
# Create
_create_simple_test_package("aaa")
package_info = packaging.read_package_info("aaa")

# Write MKP file
mkp_path = tmp_path.joinpath("aaa.mkp")
with mkp_path.open("wb") as mkp:
packaging.create_mkp_file(package_info, mkp)

# Remove files from local hierarchy
packaging.remove_package(package_info)
assert packaging._package_exists("aaa") is False

# And now install the package from memory
packaging.install_package_by_path(mkp_path)

# Check result
assert packaging._package_exists("aaa") is True
package_info = packaging.read_package_info("aaa")
assert package_info["version"] == "1.0"
assert package_info["files"]["checks"] == ["aaa"]
assert cmk.utils.paths.local_checks_dir.joinpath("aaa").exists()


def test_release_package_not_existing():
with pytest.raises(packaging.PackageException):
packaging.release_package("abc")


def test_release_package():
_create_simple_test_package("aaa")
assert packaging._package_exists("aaa") is True
assert cmk.utils.paths.local_checks_dir.joinpath("aaa").exists()

packaging.release_package("aaa")

assert packaging._package_exists("aaa") is False
assert cmk.utils.paths.local_checks_dir.joinpath("aaa").exists()


def test_create_mkp_file():
package_info = _create_simple_test_package("aaa")

mkp = BytesIO()
packaging.create_mkp_file(package_info, mkp)
mkp.seek(0)

tar = tarfile.open(fileobj=mkp, mode="r:gz")
assert sorted(tar.getnames()) == sorted(["info", "info.json", "checks.tar"])

info = ast.literal_eval(tar.extractfile("info").read())
assert info["name"] == "aaa"

info2 = json.loads(tar.extractfile("info.json").read())
assert info2["name"] == "aaa"


def test_remove_package():
package_info = _create_simple_test_package("aaa")
packaging.remove_package(package_info)
assert packaging._package_exists("aaa") is False


def test_unpackaged_files_none():
assert packaging.unpackaged_files() == {
'agents': [],
'alert_handlers': [],
'bin': [],
'checkman': [],
'checks': [],
'doc': [],
'ec_rule_packs': [],
'inventory': [],
'lib': [],
'locales': [],
'mibs': [],
'notifications': [],
'pnp-templates': [],
'web': [],
}


def test_unpackaged_files():
_create_test_file("abc")

p = cmk.utils.paths.local_doc_dir.joinpath("docxx")
with p.open("w", encoding="utf-8") as f:
f.write(u"lala\n")

assert packaging.unpackaged_files() == {
'agents': [],
'alert_handlers': [],
'bin': [],
'checkman': [],
'checks': ['abc'],
'doc': ["docxx"],
'ec_rule_packs': [],
'inventory': [],
'lib': [],
'locales': [],
'mibs': [],
'notifications': [],
'pnp-templates': [],
'web': [],
}


# TODO:
#def test_get_all_package_infos()
#def test_unpackaged_files()
#def test_package_part_info()
#def test_unpackaged_files_in_dir()
#def test_packaged_files_in_dir()
#def test_all_package_names()
#def test_write_package_info()


def test_parse_package_info_pre_160():
assert packaging.parse_package_info(repr({"name": "aaa"}))["version.usable_until"] is None

0 comments on commit b05c8ec

Please sign in to comment.
You can’t perform that action at this time.