diff --git a/src/rhsmlib/services/attach.py b/src/rhsmlib/services/attach.py deleted file mode 100644 index c141da1400..0000000000 --- a/src/rhsmlib/services/attach.py +++ /dev/null @@ -1,83 +0,0 @@ -# Copyright (c) 2017 Red Hat, Inc. -# -# This software is licensed to you under the GNU General Public License, -# version 2 (GPLv2). There is NO WARRANTY for this software, express or -# implied, including the implied warranties of MERCHANTABILITY or FITNESS -# FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 -# along with this software; if not, see -# http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. -# -# Red Hat trademarks are not licensed under GPLv2. No permission is -# granted to use or replicate Red Hat trademarks that are incorporated -# in this software or its documentation. -import logging - -from subscription_manager import injection as inj -from rhsm.connection import UEPConnection - -log = logging.getLogger(__name__) - - -class AttachService: - """ - Service using for attach pools by ID or auto-attaching - """ - - def __init__(self, cp: UEPConnection) -> None: - self.plugin_manager = inj.require(inj.PLUGIN_MANAGER) - self.identity = inj.require(inj.IDENTITY) - self.cp = cp - - def attach_auto(self, service_level: str = None) -> dict: - """ - Try to perform auto-attach. Candlepin server tries to attach pool according - installed product certificates and other attributes (service_level, syspurpose, etc.) - :param service_level: String with requested service level - :return: Dictionary with result of REST-API call - """ - - # FIXME: First check if current service_level is the same as provided in - # argument and if not then try to set something new. Otherwise it is useless - # and it only cause unnecessary REST API call - if service_level is not None: - self.cp.updateConsumer(self.identity.uuid, service_level=service_level) - log.debug("Service level set to: %s" % service_level) - - self.plugin_manager.run( - "pre_auto_attach", - consumer_uuid=self.identity.uuid, - ) - - resp = self.cp.bind(self.identity.uuid) - - self.plugin_manager.run( - "post_auto_attach", - consumer_uuid=self.identity.uuid, - entitlement_data=resp, - ) - - return resp - - def attach_pool(self, pool: str, quantity: int) -> dict: - """ - Try to attach pool - :param pool: String with pool ID - :param quantity: Quantity of subscriptions user wants to consume - :return: Dictionary with result of REST-API call - """ - - # If quantity is None, server will assume 1. pre_subscribe will - # report the same. - self.plugin_manager.run( - "pre_subscribe", consumer_uuid=self.identity.uuid, pool_id=pool, quantity=quantity - ) - - resp = self.cp.bindByEntitlementPool(self.identity.uuid, pool, quantity) - - self.plugin_manager.run( - "post_subscribe", - consumer_uuid=self.identity.uuid, - entitlement_data=resp, - ) - - return resp diff --git a/test/rhsmlib/services/test_attach.py b/test/rhsmlib/services/test_attach.py deleted file mode 100644 index 80d0a16512..0000000000 --- a/test/rhsmlib/services/test_attach.py +++ /dev/null @@ -1,153 +0,0 @@ -# Copyright (c) 2017 Red Hat, Inc. -# -# This software is licensed to you under the GNU General Public License, -# version 2 (GPLv2). There is NO WARRANTY for this software, express or -# implied, including the implied warranties of MERCHANTABILITY or FITNESS -# FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 -# along with this software; if not, see -# http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. -# -# Red Hat trademarks are not licensed under GPLv2. No permission is -# granted to use or replicate Red Hat trademarks that are incorporated -# in this software or its documentation. -from unittest import mock - -from test.rhsmlib.base import InjectionMockingTest - -from subscription_manager import injection as inj -from subscription_manager.identity import Identity -from subscription_manager.plugins import PluginManager - -from rhsm import connection - -from rhsmlib.services import attach - - -CONTENT_JSON = [ - { - "id": "19ec0d4f93ae47e18233b2590b3e71f3", - "consumer": { - "id": "8a8d01865d2cb201015d331b0078006a", - "uuid": "47680d96-cfa4-4326-b545-5a6e02a4e95a", - "name": "orgBConsumer-tZTbHviW", - "href": "/consumers/47680d96-cfa4-4326-b545-5a6e02a4e95a", - }, - "pool": { - "id": "8a8d01865d2cb201015d331b01b6006f", - "type": "NORMAL", - "owner": { - "id": "8a8d01865d2cb201015d331afec50059", - "key": "orgB-txDmAJWq", - "displayName": "orgB-txDmAJWq", - "href": "/owners/orgB-txDmAJWq", - }, - "activeSubscription": True, - "quantity": 1, - "startDate": "2017-07-11T19:23:14+0000", - "endDate": "2018-07-11T19:23:14+0000", - "attributes": [], - "consumed": 1, - "exported": 0, - "shared": 0, - "branding": [], - "calculatedAttributes": {"compliance_type": "Standard"}, - "productId": "prod-25G4r19T", - "productAttributes": [{"name": "type", "value": "SVC"}], - "derivedProductAttributes": [], - "productName": "prod-Fz0IBfN6", - "stacked": False, - "developmentPool": False, - "href": "/pools/8a8d01865d2cb201015d331b01b6006f", - "created": "2017-07-11T19:23:14+0000", - "updated": "2017-07-11T19:23:14+0000", - "providedProducts": [], - "derivedProvidedProducts": [], - "subscriptionId": "source_sub_-LO4l9YKv", - "subscriptionSubKey": "master", - }, - "certificates": [ - { - "key": "FAKE KEY", - "cert": "FAKE_CERT", - "serial": { - "id": 7020569423934353740, - "revoked": False, - "collected": False, - "expiration": "2018-07-11T19:23:14+0000", - "serial": 7020569423934353740, - "created": "2017-07-11T19:23:14+0000", - "updated": "2017-07-11T19:23:14+0000", - }, - "id": "8a8d01865d2cb201015d331b02870072", - "created": "2017-07-11T19:23:14+0000", - "updated": "2017-07-11T19:23:14+0000", - } - ], - "quantity": 1, - "startDate": "2017-07-11T19:23:14+0000", - "endDate": "2018-07-11T19:23:14+0000", - "href": "/entitlements/19ec0d4f93ae47e18233b2590b3e71f3", - "created": "2017-07-11T19:23:14+0000", - "updated": "2017-07-11T19:23:14+0000", - } -] - - -class TestAttachService(InjectionMockingTest): - def setUp(self): - super(TestAttachService, self).setUp() - self.mock_identity = mock.Mock(spec=Identity, name="Identity").return_value - self.mock_cp = mock.Mock(spec=connection.UEPConnection, name="UEPConnection").return_value - self.mock_pm = mock.Mock(spec=PluginManager, name="PluginManager").return_value - - def injection_definitions(self, *args, **kwargs): - if args[0] == inj.IDENTITY: - return self.mock_identity - elif args[0] == inj.PLUGIN_MANAGER: - return self.mock_pm - else: - return None - - def test_pool_attach(self): - self.mock_identity.is_valid.return_value = True - self.mock_identity.uuid = "id" - - self.mock_cp.bindByEntitlementPool.return_value = CONTENT_JSON - - result = attach.AttachService(self.mock_cp).attach_pool("x", 1) - - self.assertEqual(CONTENT_JSON, result) - - expected_bind_calls = [ - mock.call("id", "x", 1), - ] - self.assertEqual(expected_bind_calls, self.mock_cp.bindByEntitlementPool.call_args_list) - - expected_plugin_calls = [ - mock.call("pre_subscribe", consumer_uuid="id", pool_id="x", quantity=1), - mock.call("post_subscribe", consumer_uuid="id", entitlement_data=CONTENT_JSON), - ] - self.assertEqual(expected_plugin_calls, self.mock_pm.run.call_args_list) - - def test_auto_attach(self): - self.mock_identity.is_valid.return_value = True - self.mock_identity.uuid = "id" - - self.mock_cp.bind.return_value = CONTENT_JSON - - result = attach.AttachService(self.mock_cp).attach_auto("service_level") - self.assertEqual(CONTENT_JSON, result) - - expected_update_calls = [mock.call("id", service_level="service_level")] - self.assertEqual(expected_update_calls, self.mock_cp.updateConsumer.call_args_list) - - expected_bind_calls = [ - mock.call("id"), - ] - self.assertEqual(expected_bind_calls, self.mock_cp.bind.call_args_list) - - expected_plugin_calls = [ - mock.call("pre_auto_attach", consumer_uuid="id"), - mock.call("post_auto_attach", consumer_uuid="id", entitlement_data=CONTENT_JSON), - ] - self.assertEqual(expected_plugin_calls, self.mock_pm.run.call_args_list)