-
Notifications
You must be signed in to change notification settings - Fork 120
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
utils created, parametrized activation keys registration
- Loading branch information
Showing
5 changed files
with
163 additions
and
122 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
142 changes: 142 additions & 0 deletions
142
integration-tests/test_register_with_activation_keys.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| import os | ||
| import pytest | ||
| import contextlib | ||
| from pytest_client_tools.util import Version | ||
| from conftest import RHSMPrivateBus | ||
| from constants import RHSM, RHSM_REGISTER_SERVER, RHSM_REGISTER | ||
| from dasbus.error import DBusError | ||
| from dasbus.typing import get_variant, Str, get_native, List | ||
|
|
||
| import sh | ||
| import subprocess | ||
| import json | ||
| import logging | ||
|
|
||
| import re | ||
| from dataclasses import dataclass | ||
| from functools import reduce | ||
| import json | ||
| from funcy import first, partial, take | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| """ | ||
| It is important to run tests as root. Since RegisterServer is a system dbus service. | ||
| And it provides a unix socket connection. | ||
| """ | ||
|
|
||
| @pytest.mark.parametrize("num_of_act_keys_to_use",[0,1,2]) | ||
| def test_register_with_activation_keys(external_candlepin, subman, test_config, num_of_act_keys_to_use): | ||
| """ | ||
| https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6 | ||
| """ | ||
| assert not subman.is_registered | ||
|
|
||
| candlepin_config = partial(test_config.get,"candlepin") | ||
| act_keynames = take(num_of_act_keys_to_use, candlepin_config("activation_keys")) | ||
|
|
||
| proxy = RHSM.get_proxy(RHSM_REGISTER_SERVER) | ||
| with RHSMPrivateBus(proxy) as private_bus: | ||
| private_proxy = private_bus.get_proxy(RHSM.service_name, | ||
| RHSM_REGISTER.object_path) | ||
| locale = os.environ.get("LANG", "") | ||
| response = json.loads(private_proxy.RegisterWithActivationKeys(candlepin_config("org"), | ||
| act_keynames, | ||
| {}, | ||
| {}, | ||
| locale)) | ||
| if num_of_act_keys_to_use == 0: | ||
| assert "No activation key specified" in response | ||
| else: | ||
| assert 'activationKeys' in response,\ | ||
| "DBus method returns what activation keys were used to register a system" | ||
|
|
||
| logger.debug(response['activationKeys']) | ||
| assert sorted([ii['activationKeyName'] for ii in response['activationKeys']]) == sorted(act_keynames) | ||
|
|
||
| assert subman.is_registered | ||
|
|
||
| def test_register_with_activation_keys_and_environments(external_candlepin, subman, test_config): | ||
| """ | ||
| https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6 | ||
| """ | ||
| candlepin_config = partial(test_config.get,"candlepin") | ||
|
|
||
| assert not subman.is_registered | ||
| proxy = RHSM.get_proxy(RHSM_REGISTER_SERVER) | ||
| act_keynames = candlepin_config("activation_keys") | ||
| with RHSMPrivateBus(proxy) as private_bus: | ||
| private_proxy = private_bus.get_proxy(RHSM.service_name, | ||
| RHSM_REGISTER.object_path) | ||
| locale = os.environ.get("LANG", "") | ||
| response = json.loads(private_proxy.RegisterWithActivationKeys(candlepin_config("org"), | ||
| act_keynames, | ||
| {"environments":get_variant(Str, first(candlepin_config("environment","ids")))}, | ||
| {}, | ||
| locale)) | ||
|
|
||
| assert 'activationKeys' in response,\ | ||
| "DBus method returns what activation keys were used to register a system" | ||
|
|
||
| logger.debug(response['activationKeys']) | ||
| assert sorted([ii['activationKeyName'] for ii in response['activationKeys']]) == sorted(act_keynames) | ||
|
|
||
| subman_response = subman.run("identity") | ||
| """ | ||
| (env) [root@kvm-08-guest21 integration-tests]# subscription-manager identity | ||
| system identity: 5c00d2c6-5bea-4b6d-8662-8680e38f0dab | ||
| name: kvm-08-guest21.lab.eng.rdu2.dc.redhat.com | ||
| org name: Donald Duck | ||
| org ID: donaldduck | ||
| environment name: env-name-01 | ||
| """ | ||
| def read_pair(line): | ||
| result = re.search(r"^([^:]+):(.*)",line.strip()) | ||
| if result: | ||
| pair = [g.strip() for g in result.groups()] | ||
| return pair | ||
| return [] | ||
|
|
||
| pairs = dict([read_pair(line) for line in subman_response.stdout.splitlines()]) | ||
| logger.debug(pairs) | ||
|
|
||
| assert "environment name" in pairs | ||
| assert pairs["environment name"] == first(candlepin_config("environment","names")) | ||
|
|
||
| assert subman.is_registered | ||
|
|
||
|
|
||
| @pytest.mark.parametrize("wrong_case",["wrong-act-key inside","just one wrong-act-key", "wrong-org","wrong-act-key and wrong-org"]) | ||
| def test_register_with_activation_keys_wrong_case(external_candlepin, subman, test_config, wrong_case): | ||
| """ | ||
| https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6 | ||
| """ | ||
| candlepin_config = partial(test_config.get,"candlepin") | ||
| act_keynames = ("just one" not in wrong_case) and candlepin_config("activation_keys") or [] | ||
| if "wrong-act-key" in wrong_case: | ||
| act_keynames.append("wrong-act-key") | ||
|
|
||
| org_to_use = ("org" in wrong_case) and "wrong-org" or candlepin_config("org") | ||
|
|
||
| proxy = RHSM.get_proxy(RHSM_REGISTER_SERVER) | ||
| with RHSMPrivateBus(proxy) as private_bus: | ||
| private_proxy = private_bus.get_proxy(RHSM.service_name, | ||
| RHSM_REGISTER.object_path) | ||
| locale = os.environ.get("LANG", "") | ||
| with pytest.raises(DBusError) as excinfo: | ||
| response = json.loads(private_proxy.RegisterWithActivationKeys(org_to_use, | ||
| act_keynames, | ||
| {}, | ||
| {}, | ||
| locale)) | ||
| if "wrong-org" in wrong_case: #two wrong cases together | ||
| assert "Organization wrong-org does not exist." in str(excinfo.value) | ||
|
|
||
| if "wrong-act-key inside" == wrong_case: | ||
| assert "some activation key specified does not exist for this org" in str(excinfo.value) | ||
|
|
||
| if "just one wrong-act-key" == wrong_case: | ||
| assert "None of the activation keys specified exist for this org" in str(excinfo.value) | ||
|
|
||
| assert not subman.is_registered | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| import time | ||
|
|
||
| def loop_until(predicate, poll_sec=5, timeout_sec=120): | ||
| """ | ||
| An helper function to handle a time period waiting for an external service | ||
| to update its state. | ||
| an example: | ||
| assert loop_until(lambda: insights_client.is_registered) | ||
| The loop function will retry to run predicate every 5secs | ||
| until the total time exceeds timeout_sec. | ||
| """ | ||
| start = time.time() | ||
| ok = False | ||
| while (not ok) and (time.time() - start < timeout_sec): | ||
| time.sleep(poll_sec) | ||
| ok = predicate() | ||
| return ok | ||
|
|