8000 Extend the find_items method and test it by SchoolGuy · Pull Request #2663 · cobbler/cobbler · GitHub
[go: up one dir, main page]
More Web Proxy on the site http://driver.im/
Skip to content

Extend the find_items method and test it #2663

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 9, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 29 additions & 10 deletions cobbler/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -844,28 +844,47 @@ def add_menu(self, ref, check_for_duplicate_names=False, save=True):

# ==========================================================================

# FIXME: find_items should take all the arguments the other find methods do.

def find_items(self, what, criteria=None):
def find_items(self, what: str, criteria: dict = None, name: Optional[str] = None, return_list: bool = True,
no_errors: bool = False):
"""
This is the abstract base method for finding object int the api. It should not be used by external resources.
Please reefer to the specific implementations of this method called ``find_<object type>``.

:param what: The object type of the item to search for.
:param criteria: The dictionary with the key-value pairs to find objects with.
:param name: The name of the object.
:param return_list: If only the first result or all results should be returned.
:param no_errors: Silence some errors which would raise if this turned to False.
:return: The list of items witch match the search criteria.
"""
self.log("find_items", [what])
# defaults

if criteria is None:
criteria = {}

if what == "" and ("name" in criteria or name is not None):
return self.__find_by_name(criteria.get("name", name))

if what not in ["distro", "profile", "system", "repo", "image", "mgmtclass", "package", "file", "menu"]:
raise ValueError("what needs to be a valid collection!")

items = self._collection_mgr.get_items(what)
# empty criteria returns everything
if criteria == {}:
res = items
else:
res = items.find(return_list=True, no_errors=False, **criteria)
return res
return items.find(name=name, return_list=return_list, no_errors=no_errors, **criteria)

def __find_by_name(self, name: str):
"""
This is a magic method which just searches all collections for the specified name directly,
:param name: The name of the item(s).
:return: The found item or None.
"""
if not isinstance(name, str):
raise TypeError("name of an object must be of type str!")
collections = ["distro", "profile", "system", "repo", "image", "mgmtclass", "package", "file", "menu"]
for collection_name in collections:
match = self.find_items(collection_name, name=name, return_list=False)
if match is not None:
return match
return None

def find_distro(self, name=None, return_list=False, no_errors=False, **kargs):
"""
Expand Down
2 changes: 1 addition & 1 deletion cobbler/cobbler_collections/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -235,5 +235,5 @@ def get_items(self, collection_type: str) -> Union[Distros, Profiles, Systems, R
elif collection_type == "settings":
result = self._settings
else:
raise CX("internal error, collection name %s not supported" % collection_type)
raise CX("internal error, collection name \"%s\" not supported" % collection_type)
return result
17 changes: 12 additions & 5 deletions cobbler/remote.py
Original file line number Diff line number Diff line change
Expand Up @@ -572,7 +572,7 @@ def _log(self, msg, user=None, token=None, name=None, object_id=None, attribute=
else:
self.logger.info(msg)

def __sort(self, data, sort_field=None):
def __sort(self, data, sort_field: Optional[str] = None):
"""
Helper function used by the various find/search functions to return object representations in order.

Expand Down Expand Up @@ -921,7 +921,7 @@ def get_menus(self, page=None, results_per_page=None, token=None, **rest):
"""
return self.get_items("menu")

def find_items(self, what, criteria=None, sort_field=None, expand: bool = True) -> list:
def find_items(self, what, criteria: dict = None, sort_field=None, expand: bool = True) -> list:
"""Works like get_items but also accepts criteria as a dict to search on.

Example: ``{ "name" : "*.example.org" }``
Expand All @@ -935,7 +935,11 @@ def find_items(self, what, criteria=None, sort_field=None, expand: bool = True)
:returns: A list of dicts.
"""
self._log("find_items(%s); criteria(%s); sort(%s)" % (what, criteria, sort_field))
items = self.api.find_items(what, criteria=criteria)
if "name" in criteria:
name = criteria.pop("name")
items = self.api.find_items(what, criteria=criteria, name=name)
else:
items = self.api.find_items(what, criteria=criteria)
items = self.__sort(items, sort_field)
if not expand:
items = [x.name for x in items]
Expand Down Expand Up @@ -1065,7 +1069,10 @@ def find_items_paged(self, what, criteria=None, sort_field=None, page=None, item
:return: The found items.
"""
self._log("find_items_paged(%s); criteria(%s); sort(%s)" % (what, criteria, sort_field), token=token)
items = self.api.find_items(what, criteria=criteria)
if criteria is None:
items = self.api.get_items(what)
else:
items = self.api.find_items(what, criteria=criteria)
items = self.__sort(items, sort_field)
(items, pageinfo) = self.__paginate(items, page, items_per_page)
items = [x.to_dict() for x in items]
Expand Down Expand Up @@ -1575,7 +1582,7 @@ def new_item(self, what, token, is_subobject: bool = False):
elif what == "menu":
d = menu.Menu(self.api, is_subobject=is_subobject)
else:
raise CX("internal error, collection name is %s" % what)
raise CX("internal error, collection name is \"%s\"" % what)
key = "___NEW___%s::%s" % (what, self.__get_random(25))
self.object_cache[key] = (time.time(), d)
return key
Expand Down
Empty file added tests/api/__init__.py
Empty file.
207 changes: 207 additions & 0 deletions tests/api/find_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,207 @@
import pytest

from cobbler.api import CobblerAPI
from tests.conftest import does_not_raise

# TODO: Extend the fillup and add more testcases


@pytest.fixture
def find_fillup():
test_api = CobblerAPI()

return test_api


@pytest.mark.parametrize("what,criteria,name,return_list,no_errors,expected_exception,expected_result", [
("", None, "", False, False, does_not_raise(), None),
("distro", {}, "", False, False, does_not_raise(), None)
])
def test_find_items(find_fillup, what, criteria, name, return_list, no_errors, expected_exception, expected_result):
# Arrange
test_api = find_fillup

# Act
with expected_exception:
result = test_api.find_items(what, criteria, name, return_list, no_errors)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_distro(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = find_fillup

# Act
with expected_exception:
result = test_api.find_distro(name, return_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_profile(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = find_fillup

# Act
with expected_exception:
result = test_api.find_profile(name, retur 6D47 n_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_system(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = find_fillup

# Act
with expected_exception:
result = test_api.find_system(name, return_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_repo(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = CobblerAPI()

# Act
with expected_exception:
result = test_api.find_repo(name, return_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
9E81 def test_find_image(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = CobblerAPI()

# Act
with expected_exception:
result = test_api.find_image(name, return_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_mgmtclass(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = CobblerAPI()

# Act
with expected_exception:
result = test_api.find_mgmtclass(name, return_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_package(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = CobblerAPI()

# Act
with expected_exception:
result = test_api.find_package(name, return_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
(None, False, False, {}, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_file(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = CobblerAPI()

# Act
with expected_exception:
result = test_api.find_file(name, return_list, no_errors, **criteria)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result


@pytest.mark.parametrize("name,return_list,no_errors,criteria,expected_exception,expected_result", [
("", False, False, {}, does_not_raise(), None),
(None, False, False, None, pytest.raises(ValueError), None),
("testdistro", False, False, {}, does_not_raise(), None)
])
def test_find_menu(find_fillup, name, return_list, no_errors, criteria, expected_exception, expected_result):
# Arrange
test_api = CobblerAPI()

# Act
with expected_exception:
if criteria is not None:
result = test_api.find_menu(name, return_list, no_errors, **criteria)
else:
result = test_api.find_menu(name, return_list, no_errors)

# Assert
if expected_result is None:
assert result is None
else:
assert result == expected_result
File renamed without changes.
0