diff --git a/integration-tests/conftest.py b/integration-tests/conftest.py index 96cb92fe1c..b353f56b7a 100644 --- a/integration-tests/conftest.py +++ b/integration-tests/conftest.py @@ -51,25 +51,6 @@ def register_subman( yield subman_session -def loop_until(predicate, poll_sec=5, timeout_sec=120): - """ - An helper function to handle a time periond 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 - class RHSMPrivateBus(MessageBus): """Representation of RHSM private bus connection that can be used as a context manager.""" diff --git a/integration-tests/dev-requirements.txt b/integration-tests/dev-requirements.txt deleted file mode 100644 index 7d3f66d606..0000000000 --- a/integration-tests/dev-requirements.txt +++ /dev/null @@ -1,2 +0,0 @@ --r requirements.txt -pytest-hot-reloading diff --git a/integration-tests/test_register.py b/integration-tests/test_register.py index d1fd50b1c3..e847cb0949 100644 --- a/integration-tests/test_register.py +++ b/integration-tests/test_register.py @@ -201,104 +201,3 @@ def as_set(env): assert orig_env.issubset(env_from_response), \ "environment from sub-man command is a subset of fields returned by dbus GetEnvironments" - -def test_register_with_activation_key(external_candlepin, subman, test_config): - """ - https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6 - """ - assert not subman.is_registered - proxy = RHSM.get_proxy(RHSM_REGISTER_SERVER) - act_keyname_in_use = first(test_config.get("candlepin","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(test_config.get("candlepin","org"), - [ act_keyname_in_use ], - {}, - {}, - locale)) - - assert 'activationKeys' in response,\ - "DBus method returns what activation keys were used to register a system" - - assert [ii['activationKeyName'] for ii in response['activationKeys']] == [act_keyname_in_use] - - assert subman.is_registered - - -def test_register_with_activation_keys(external_candlepin, subman, test_config): - """ - https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6 - """ - assert not subman.is_registered - proxy = RHSM.get_proxy(RHSM_REGISTER_SERVER) - act_keynames = test_config.get("candlepin","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(test_config.get("candlepin","org"), - act_keynames, - {}, - {}, - 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) - - 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 - diff --git a/integration-tests/test_register_with_activation_keys.py b/integration-tests/test_register_with_activation_keys.py new file mode 100644 index 0000000000..65627b8fd0 --- /dev/null +++ b/integration-tests/test_register_with_activation_keys.py @@ -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 + diff --git a/integration-tests/utils.py b/integration-tests/utils.py new file mode 100644 index 0000000000..957e215204 --- /dev/null +++ b/integration-tests/utils.py @@ -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 +