Skip to content

Commit

Permalink
cleanup: remove check methods and relevant unit tests from authz_checker
Browse files Browse the repository at this point in the history
  • Loading branch information
@pb82
pb82 committed Jul 4, 2024
1 parent 0253f30 commit b4aaad7
Show file tree
Hide file tree
Showing 2 changed files with 2 additions and 191 deletions.
85 changes: 2 additions & 83 deletions ansible_ai_connect/users/authz_checker.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,10 +63,6 @@ def self_test(self):
"""
pass

@abstractmethod
def check(self, _user_id: str, username: str, organization_id: int) -> bool:
pass


def fatal_exception(exc) -> bool:
"""Determine if an exception is fatal or not"""
Expand Down Expand Up @@ -153,33 +149,6 @@ def self_test(self):
)
r.raise_for_status()

def check(self, user_id, username, organization_id) -> bool:
self._session.headers.update({"Authorization": f"Bearer {self._token.get()}"})
try:
r = self._session.post(
self._api_server + "/v1alpha/check",
json={
"subject": str(user_id),
"operation": "access",
"resourcetype": "license",
"resourceid": f"{organization_id}/smarts",
},
# Note: A ping from France against the preprod env, is slightly below 300ms
timeout=0.8,
)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
logger.error("Cannot reach the CIAM backend in time")
return False
if r.status_code != HTTPStatus.OK:
logger.error("Unexpected error code (%s) returned by CIAM backend" % r.status_code)
return False
data = r.json()
try:
return data["result"]
except (KeyError, TypeError):
logger.error("Unexpected Answer from CIAM")
return False


class AMSCheck(BaseCheck):
# An AMS Organization ID that should never match anything
Expand Down Expand Up @@ -276,48 +245,6 @@ def self_test(self):
)
r.raise_for_status()

def check(self, user_id: str, username: str, organization_id: int) -> bool:
try:
ams_org_id = self.get_ams_org(organization_id)
except AMSCheck.AMSError:
# See https://issues.redhat.com/browse/AAP-22758
# If the AMS Organisation lookup fails assume the check failed.
# The 'check()' function is obsolete and not called. This code
# only exists as a matter of 'completeness'.
return False

if ams_org_id == AMSCheck.ERROR_AMS_ORG_UNDEFINED:
# Organization has not yet been created in AMS, either the organization
# is too recent (sync is done every 1h), or the organization has no AMS related
# services (e.g Ansible, OpenShift, cloud.r.c) and so is not synchronized.
logger.warning(f"Organization not found in AMS, organization_id={organization_id}")
return False

params = {
"search": "plan.id = 'AnsibleWisdom' AND status = 'Active' AND "
f"creator.username = '{username}' AND organization_id='{ams_org_id}'"
}
self.update_bearer_token()

try:
r = self._session.get(
self._api_server + "/api/accounts_mgmt/v1/subscriptions",
params=params,
timeout=0.8,
)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
logger.error(self.ERROR_AMS_CONNECTION_TIMEOUT)
return False
if r.status_code != HTTPStatus.OK:
logger.error("Unexpected error code (%s) returned by AMS backend (sub)" % r.status_code)
return False
data = r.json()
try:
return len(data["items"]) > 0
except (KeyError, ValueError):
logger.error("Unexpected subscription answer from AMS")
return False

def rh_user_is_org_admin(self, username: str, organization_id: int):
try:
ams_org_id = self.get_ams_org(organization_id)
Expand All @@ -330,7 +257,7 @@ def rh_user_is_org_admin(self, username: str, organization_id: int):
# Organization has not yet been created in AMS, either the organization
# is too recent (sync is done every 1h), or the organization has no AMS related
# services (e.g Ansible, OpenShift, cloud.r.c) and so is not synchronized.
logger.warning(f"Organization not found in AMS, organization_id={organization_id}")
logger.warning(f"Organization unavailable in AMS, organization_id={organization_id}")
return False

params = {"search": f"account.username = '{username}' AND organization.id='{ams_org_id}'"}
Expand Down Expand Up @@ -375,7 +302,7 @@ def rh_org_has_subscription(self, organization_id: int) -> bool:
# Organization has not yet been created in AMS, either the organization
# is too recent (sync is done every 1h), or the organization has no AMS related
# services (e.g Ansible, OpenShift, cloud.r.c) and so is not synchronized.
logger.warning(f"Organization not found in AMS, organization_id={organization_id}")
logger.warning(f"Organization unavailable in AMS, organization_id={organization_id}")
return False

# Check cache
Expand Down Expand Up @@ -442,14 +369,6 @@ def self_test(self):
# Always passes. No exception raised.
pass

def check(self, _user_id: str, username: str, organization_id: int) -> bool:
if not self.rh_org_has_subscription(organization_id):
return False
if settings.AUTHZ_DUMMY_USERS_WITH_SEAT == "*":
return True
seated_user = settings.AUTHZ_DUMMY_USERS_WITH_SEAT.split(",")
return username in seated_user

def rh_org_has_subscription(self, organization_id: int) -> bool:
if settings.AUTHZ_DUMMY_ORGS_WITH_SUBSCRIPTION == "*":
return True
Expand Down
108 changes: 0 additions & 108 deletions ansible_ai_connect/users/tests/test_authz_checker.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -141,40 +141,6 @@ def test_fatal_exception(self):
b = fatal_exception(exc)
self.assertTrue(b)

def test_ciam_check(self):
m_r = Mock()
m_r.json.return_value = {"result": True}
m_r.status_code = 200

checker = CIAMCheck("foo", "bar", "https://sso.redhat.com", "https://some-api.server.host")
checker._token = Mock()
checker._session = Mock()
checker._session.post.return_value = m_r
self.assertTrue(checker.check("my_id", "my_name", 123))
checker._session.post.assert_called_with(
"https://some-api.server.host/v1alpha/check",
json={
"subject": "my_id",
"operation": "access",
"resourcetype": "license",
"resourceid": "123/smarts",
},
timeout=0.8,
)

def test_ciam_check_with_500_status_code(self):
m_r = Mock()
m_r.status_code = 500

checker = CIAMCheck("foo", "bar", "https://sso.redhat.com", "https://some-api.server.host")
checker._token = Mock()
checker._session = Mock()
checker._session.post.return_value = m_r

with self.assertLogs(logger="root", level="ERROR") as log:
self.assertFalse(checker.check("my_id", "my_name", 123))
self.assertInLog("Unexpected error code (500) returned by CIAM backend", log)

def test_ciam_self_test_success(self):
m_r = Mock()
m_r.status_code = 200
Expand Down Expand Up @@ -271,62 +237,6 @@ def test_ams_get_ams_org_with_empty_data(self):
log,
)

def test_ams_check(self):
m_r = Mock()
m_r.json.side_effect = [{"items": [{"id": "qwe"}]}, {"items": [{"id": "asd"}]}]
m_r.status_code = 200

checker = self.get_default_ams_checker()
checker._token = Mock()
checker._session = Mock()
checker._session.get.return_value = m_r
self.assertTrue(checker.check("my_id", "my_name", 123))
checker._session.get.assert_called_with(
"https://some-api.server.host/api/accounts_mgmt/v1/subscriptions",
params={
"search": "plan.id = 'AnsibleWisdom' AND status = 'Active' "
"AND creator.username = 'my_name' AND organization_id='qwe'"
},
timeout=0.8,
)

def test_ams_check_multiple_seats(self):
m_r = Mock()
m_r.json.side_effect = [
{"items": [{"id": "qwe"}, {"id": "rty"}]},
{"items": [{"id": "asd"}, {"id": "fgh"}]},
]
m_r.status_code = 200

checker = self.get_default_ams_checker()
checker._token = Mock()
checker._session = Mock()
checker._session.get.return_value = m_r
self.assertTrue(checker.check("my_id", "my_name", 123))
checker._session.get.assert_called_with(
"https://some-api.server.host/api/accounts_mgmt/v1/subscriptions",
params={
"search": "plan.id = 'AnsibleWisdom' AND status = 'Active' "
"AND creator.username = 'my_name' AND organization_id='qwe'"
},
timeout=0.8,
)

def test_ams_check_with_500_status_code(self):
m_r = Mock()
m_r.status_code = 500

checker = self.get_default_ams_checker()
checker._token = Mock()
checker._session = Mock()
checker._session.get.return_value = m_r

with self.assertLogs(logger="root", level="ERROR") as log:
self.assertFalse(checker.check("my_id", "my_name", 123))
self.assertInLog(
"Unexpected error code (500) returned by AMS backend (organizations)", log
)

def test_ams_self_test_success(self):
m_r = Mock()
m_r.status_code = 200
Expand Down Expand Up @@ -652,24 +562,6 @@ def setUp(self):
def test_self_test(self):
self.assertIsNone(self.checker.self_test())

@override_settings(AUTHZ_DUMMY_USERS_WITH_SEAT="yves")
@override_settings(AUTHZ_DUMMY_ORGS_WITH_SUBSCRIPTION="123")
def test_check_with_seat(self):
self.assertTrue(self.checker.check(None, "yves", 123))

@override_settings(AUTHZ_DUMMY_ORGS_WITH_SUBSCRIPTION="123")
def test_check_with_no_seat(self):
self.assertFalse(self.checker.check(None, "noseat", 123))

@override_settings(AUTHZ_DUMMY_USERS_WITH_SEAT="*")
@override_settings(AUTHZ_DUMMY_ORGS_WITH_SUBSCRIPTION="123")
def test_check_with_wildcard(self):
self.assertTrue(self.checker.check(None, "rose", 123))

@override_settings(AUTHZ_DUMMY_USERS_WITH_SEAT="yves")
def test_check_with_no_sub(self):
self.assertFalse(self.checker.check(None, "noseat", 123))

@override_settings(AUTHZ_DUMMY_ORGS_WITH_SUBSCRIPTION="123")
def test_rh_org_has_subscription_with_sub(self):
self.assertTrue(self.checker.rh_org_has_subscription(123))
Expand Down

0 comments on commit b4aaad7

Please sign in to comment.