import pytest from influxdb_client import PermissionResource, Permission, Authorization, AuthorizationsApi from influxdb_client.rest import ApiException from tests.base_test import BaseTest class AuthorizationsClientTest(BaseTest): def setUp(self) -> None: super(AuthorizationsClientTest, self).setUp() self.user = self.users_api.me() self.organization = self.find_my_org() def test_createAuthorization(self): user_resource = PermissionResource(org_id=self.organization.id, type="users") read_users = Permission(action="read", resource=user_resource) org_resource = PermissionResource(org_id=self.organization.id, type="orgs") write_organizations = Permission(action="write", resource=org_resource) permissions = [read_users, write_organizations] authorization = self.authorizations_api.create_authorization(self.organization.id, permissions) self.log(authorization) self.assertIsNotNone(authorization) self.assertTrue(len(authorization.token) > 0) self.assertEqual(authorization.user_id, self.user.id) self.assertEqual(authorization.user, self.user.name) self.assertEqual(authorization.org_id, self.organization.id) self.assertEqual(authorization.org, self.organization.name) self.assertEqual(authorization.status, "active") self.assertTrue(len(authorization.permissions) == 2) self.assertEqual(authorization.permissions[0].resource.type, "users") self.assertEqual(authorization.permissions[0].resource.org_id, self.organization.id) self.assertEqual(authorization.permissions[0].action, "read") self.assertEqual(authorization.permissions[1].resource.type, "orgs") self.assertEqual(authorization.permissions[1].resource.org_id, self.organization.id) self.assertEqual(authorization.permissions[1].action, "write") self.assertTrue(authorization.links is not None) self.assertEqual(authorization.links["self"], "/api/v2/authorizations/" + authorization.id) self.assertEqual(authorization.links["user"], "/api/v2/users/" + self.user.id) def test_authorizationDescription(self): organization = self.my_organization resource = PermissionResource(org_id=organization.id, type="sources") create_source = Permission(action="write", resource=resource) permissions = [create_source] authorization = Authorization(org_id=organization.id, permissions=permissions) authorization.status = "active" authorization.description = "My description!" created = self.authorizations_api.create_authorization(authorization=authorization) self.assertIsNotNone(created) self.assertEqual(created.description, "My description!") def test_createAuthorizationTask(self): resource = PermissionResource(org_id=self.organization.id, type="tasks") create_task = Permission(action="read", resource=resource) delete_task = Permission(action="write", resource=resource) permissions = [create_task, delete_task] authorization = self.authorizations_api.create_authorization(self.organization.id, permissions) self.assertTrue(len(authorization.permissions) == 2) self.assertEqual(authorization.permissions[0].resource.type, "tasks") self.assertEqual(authorization.permissions[0].resource.org_id, self.organization.id) self.assertEqual(authorization.permissions[0].action, "read") self.assertEqual(authorization.permissions[1].resource.type, "tasks") self.assertEqual(authorization.permissions[1].resource.org_id, self.organization.id) self.assertEqual(authorization.permissions[1].action, "write") def test_createAuthorizationBucket(self): organization = self.client.organizations_api().create_organization(self.generate_name("Auth Organization")) bucket = self.client.buckets_api().create_bucket(bucket_name=self.generate_name("Auth Bucket"), retention_rules=BaseTest.retention_rule(), org=self.organization) resource = PermissionResource(org_id=organization.id, type="buckets", id=bucket.id) create_bucket = Permission(action="read", resource=resource) delete_bucket = Permission(action="write", resource=resource) permissions = [create_bucket, delete_bucket] authorization = self.authorizations_api.create_authorization(organization.id, permissions) self.log(authorization) self.assertTrue(len(authorization.permissions) == 2) self.assertEqual(authorization.permissions[0].resource.id, bucket.id) self.assertEqual(authorization.permissions[0].resource.type, "buckets") self.assertEqual(authorization.permissions[0].action, "read") self.assertEqual(authorization.permissions[1].resource.id, bucket.id) self.assertEqual(authorization.permissions[1].resource.type, "buckets") self.assertEqual(authorization.permissions[1].action, "write") self.delete_test_bucket(bucket) def test_findAuthorizationsByID(self): with pytest.raises(ApiException) as e: self.authorization_api().find_authorization_by_id("020f755c3c082000") self.assertEqual(e.value.status, 404) self.assertIn("authorization not found", e.value.body) def test_findAuthorizations(self): authorizations = self.authorization_api().find_authorizations() size = len(authorizations) self.authorization_api().create_authorization(org_id=self.organization.id, permissions=self.new_permissions()) def test_findAuthorizationsByUser(self): size = len(self.authorization_api().find_authorizations_by_user(self.user)) self.authorization_api().create_authorization(org_id=self.organization.id, permissions=self.new_permissions()) self.assertEqual(len(self.authorization_api().find_authorizations_by_user(user=self.user)), size + 1) def test_findAuthorizationsByUserName(self): size = len(self.authorization_api().find_authorizations_by_user_name(self.user.name)) self.authorization_api().create_authorization(org_id=self.organization.id, permissions=self.new_permissions()) self.assertEqual( len(self.authorization_api().find_authorizations_by_user_name(user_name=self.user.name)), size + 1) def test_findAuthorizationsByOrg(self): size = len(self.authorization_api().find_authorizations_by_org(self.organization)) self.authorization_api().create_authorization(org_id=self.organization.id, permissions=self.new_permissions()) self.assertEqual(len(self.authorization_api().find_authorizations_by_org(self.organization)), size + 1) def test_findAuthorizationsByOrgNotFound(self): authorizations_by_org_id = self.authorization_api().find_authorizations_by_org_id("ffffffffffffffff") self.assertEqual(len(authorizations_by_org_id), 0) def test_updateAuthorizationStatus(self): resource = PermissionResource(org_id=self.organization.id, type="users") read_user = Permission(action="read", resource=resource) permissions = [read_user] authorization = self.authorization_api().create_authorization(org_id=self.my_organization.id, permissions=permissions) self.assertEqual(authorization.status, "active") authorization.status = "inactive" authorization = self.authorization_api().update_authorization(authorization) self.assertEqual(authorization.status, "inactive") authorization.status = "active" authorization = self.authorization_api().update_authorization(authorization) self.assertEqual(authorization.status, "active") def test_deleteAuthorization(self): create_authorization = self.authorization_api().create_authorization(self.organization.id, self.new_permissions()) self.assertIsNotNone(create_authorization) found_authorization = self.authorization_api().find_authorization_by_id(create_authorization.id) self.assertIsNotNone(found_authorization) self.authorization_api().delete_authorization(create_authorization) with pytest.raises(ApiException) as e: self.authorization_api().find_authorization_by_id(create_authorization.id) self.assertIn("authorization not found", e.value.body) def test_clone_authorization(self): source = self.authorization_api().create_authorization(self.organization.id, self.new_permissions()) cloned = self.authorization_api().clone_authorization(source.id) self.assertIsNotNone(cloned.token) self.assertNotEqual(source.token, cloned.token) self.assertNotEqual(source.id, cloned.id) self.assertEqual(source.user_id, cloned.user_id) self.assertEqual(source.user, cloned.user) self.assertEqual(source.org_id, cloned.org_id) self.assertEqual(source.org, cloned.org) self.assertEqual(source.status, cloned.status) self.assertEqual(source.description, cloned.description) self.assertTrue(len(cloned.permissions), 1) self.assertEqual(cloned.permissions[0].action, "read") self.assertEqual(cloned.permissions[0].resource.type, "users") self.assertEqual(cloned.permissions[0].resource.org_id, self.organization.id) def test_cloneAuthorizationNotFound(self): with pytest.raises(ApiException) as e: self.authorization_api().find_authorization_by_id("020f755c3c082000") self.assertIn("authorization not found", e.value.body) def authorization_api(self) -> AuthorizationsApi: return self.client.authorizations_api() def new_permissions(self): resource = PermissionResource(org_id=self.organization.id, type="users") permission = Permission(action="read", resource=resource) return [permission]