CCT-1051: tests for register with activation keys

CCT-1051: reverted a change in test.sh to run post-activation-keys.sh again
CCT-1051: added tests for wrong cases

integration-tests/test_register_with_activation_keys.py

@@ -0,0 +1,203 @@
+import pytest
+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
+from funcy import first, partial, take
+
+import json
+import re
+import logging
+
+logger = logging.getLogger(__name__)
+locale = "en_US.UTF-8"
+
+"""
+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", [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)
+        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 which 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)
+        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.skip
+def test_register_with_activation_keys_with_empty_list(external_candlepin, subman, test_config):
+    """
+    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_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)
+        response = json.loads(
+            private_proxy.RegisterWithActivationKeys(candlepin_config("org"), act_keys, {}, {}, locale)
+        )
+        assert "No activation key specified" in response["message"]
+        assert response["activationKeys"] == []
+
+    assert subman.is_registered
+
+
+def test_reqister_with_activation_keys_with_wrong_key_among_good_ones(any_candlepin, subman, test_config):
+    """
+    https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6
+
+    Given a DBus method RegisterWithActivationKeys is used to register a system
+    When an invalid activation key appears in a list of valid activation keys
+    Then an application registers a system using all valid activation keys
+    and the application returns a list of activation keys that was used for registration.
+    """
+    valid_act_keys = test_config.get("candlepin", "activation_keys") or []
+    act_keys = valid_act_keys + ["wrong-act-key"]
+    org_to_use = test_config.get("candlepin", "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)
+        response = json.loads(private_proxy.RegisterWithActivationKeys(org_to_use, act_keys, {}, {}, locale))
+        assert (
+            "activationKeys" in response
+        ), "DBus method returns which activation keys were used to register a system"
+        assert sorted([ii["activationKeyName"] for ii in response["activationKeys"]]) == sorted(
+            valid_act_keys
+        ), "Just valid activation keys should appear in response of the call"
+
+    assert subman.is_registered
+
+
+def test_register_with_activation_keys_wrong_act_key(external_candlepin, subman, test_config):
+    """
+    https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6
+    """
+    act_keys = ["wrong-act-key"]
+    org_to_use = test_config.get("candlepin", "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)
+        with pytest.raises(DBusError) as excinfo:
+            private_proxy.RegisterWithActivationKeys(org_to_use, act_keys, {}, {}, locale)
+        assert "None of the activation keys specified exist for this org" in str(excinfo.value)
+
+    assert not subman.is_registered
+
+
+def test_register_with_activation_keys_wrong_org(external_candlepin, subman, test_config):
+    """
+    https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6
+    """
+    candlepin_config = partial(test_config.get, "candlepin")
+    act_keys = candlepin_config("activation_keys")
+    org_to_use = "wrong-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)
+        with pytest.raises(DBusError) as excinfo:
+            json.loads(private_proxy.RegisterWithActivationKeys(org_to_use, act_keys, {}, {}, locale))
+        assert f"Organization {org_to_use} does not exist." in str(excinfo.value)
+
+    assert not subman.is_registered
+
+
+def test_register_with_activation_keys_wrong_org_and_wrong_key(external_candlepin, subman, test_config):
+    """
+    https://www.candlepinproject.org/docs/subscription-manager/dbus_objects.html#methods-6
+    """
+    act_keys = ["wrong-act-key"]
+    org_to_use = "wrong-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)
+        with pytest.raises(DBusError) as excinfo:
+            json.loads(private_proxy.RegisterWithActivationKeys(org_to_use, act_keys, {}, {}, locale))
+        assert f"Organization {org_to_use} does not exist." in str(excinfo.value)
+
+    assert not subman.is_registered

systemtest/tests/integration/test.sh

@@ -39,10 +39,10 @@ EOF
 
 
 # run local candlepin for testing purpose
+./integration-tests/scripts/post-activation-keys.sh
 ./integration-tests/scripts/run-local-candlepin.sh
 
 # create testing data in local candlepin
-./integration-tests/scripts/post-activation-keys.sh
 ./integration-tests/scripts/post-environments.sh
 
 # There is a problem with SELinux in current version of selinux-roles (for rhsm.service)