<reponame>cyberark/conjur-openapi-spec
from __future__ import absolute_import

import unittest
from unittest.mock import patch

import conjur

from . import api_config

AUTHENTICATOR_FIELDS = [
    "enabled",
    "installed",
    "configured",
]

WHOAMI_FIELDS = [
    "client_ip",
    "user_agent",
    "account",
    "username",
    "token_issued_at",
]

class TestStatusApi(api_config.ConfiguredTest):
    """StatusApi unit test stubs"""
    @classmethod
    def setup_webservice(cls):
        """loads the webservice policy into conjur and gets it setup"""
        policies_api = conjur.api.policies_api.PoliciesApi(cls.client)
        secrets_api = conjur.api.secrets_api.SecretsApi(cls.client)
        policies_api.update_policy(cls.account, 'root', api_config.get_webservice_policy())
        secrets_api.create_secret(
            cls.account,
            "variable",
            'conjur/authn-oidc/okta/provider-uri',
            body='invalid'
        )
        secrets_api.create_secret(
            cls.account,
            "variable",
            'conjur/authn-oidc/okta/id-token-user-property',
            body='admin'
        )

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.setup_webservice()

    def setUp(self):
        self.api = conjur.api.status_api.StatusApi(self.client)
        self.bad_auth_api = conjur.api.status_api.StatusApi(self.bad_auth_client)

    def test_who_am_i_200(self):
        """Test case for who_am_i 200 response"""
        result = self.api.who_am_i()

        self.assertIsInstance(result, conjur.models.who_am_i.WhoAmI)

        for i in WHOAMI_FIELDS:
            value = getattr(result, i)
            # Just make sure that all the attributes have a value assigned to them
            self.assertNotEqual(value, '')

    def test_who_am_i_401(self):
        """Test case for who_am_i 401 response"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.bad_auth_api.who_am_i()

        self.assertEqual(context.exception.status, 401)

    @unittest.skipIf(api_config.ENTERPRISE_TESTS,
                     'Dont support testing external authenticators on Enterprise currently')
    def test_get_service_authenticator_status_200(self):
        """Test case for get_service_authenticator_status 200 return code"""
        api_config.setup_oidc_webservice()
        resp, status, _ = self.api.get_service_authenticator_status_with_http_info(
            'authn-oidc',
            'test',
            self.account
        )
        self.assertEqual(resp.status, 'ok')
        self.assertEqual(status, 200)

        with patch.object(conjur.api_client.ApiClient, 'call_api', return_value=None) \
                as mock:
            self.api.get_service_authenticator_status('authn-azure', 'test', self.account)

        mock.assert_called_once_with(
            '/{authenticator}/{service_id}/{account}/status',
            'GET',
            {'account': self.account, 'authenticator': 'authn-azure', 'service_id': 'test'},
            [],
            {'Accept': 'application/json'},
            body=None,
            files={},
            post_params=[],
            response_type='AuthenticatorStatus',
            auth_settings=['conjurAuth'],
            async_req=None,
            _return_http_data_only=True,
            _preload_content=True,
            _request_timeout=None,
            collection_formats={}
        )

    def test_get_service_authenticator_status_403(self):
        """Test case for get_service_authenticator_status 403 return code"""
        alice_client = api_config.get_api_client(username='alice')
        alice_api = conjur.api.status_api.StatusApi(alice_client)

        with self.assertRaises(conjur.exceptions.ApiException) as context:
            alice_api.get_service_authenticator_status('authn-oidc', 'okta', self.account)

        self.assertEqual(context.exception.status, 403)

    def test_get_service_authenticator_status_404(self):
        """Test case for get_service_authenticator_status 404 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_service_authenticator_status('nonexist', 'test', self.account)

        self.assertEqual(context.exception.status, 404)

    def test_get_service_authenticator_status_500(self):
        """Test case for get_service_authenticator_status 500 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_service_authenticator_status('authn-oidc', 'okta', self.account)

        self.assertEqual(context.exception.status, 500)

    def test_get_service_authenticator_status_501(self):
        """Test case for get_service_authenticator_status 500 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_service_authenticator_status('authn-ldap', 'test', self.account)

        self.assertEqual(context.exception.status, 501)

    def test_get_gcp_authenticator_status_200(self):
        """Test case for get_gcp_authenticator_status 200 return code"""
        with patch.object(conjur.api_client.ApiClient, 'call_api', return_value=None) \
                as mock:
            self.api.get_gcp_authenticator_status(self.account)

        mock.assert_called_once_with(
            '/authn-gcp/{account}/status',
            'GET',
            { 'account': self.account },
            [],
            { 'Accept': 'application/json' },
            body=None,
            files={},
            post_params=[],
            response_type='AuthenticatorStatus',
            auth_settings=['conjurAuth'],
            async_req=None,
            _return_http_data_only=True,
            _preload_content=True,
            _request_timeout=None,
            collection_formats={}
        )

    def test_get_gcp_authenticator_status_403(self):
        """Test case for get_gcp_authenticator_status 403 return code"""
        alice_client = api_config.get_api_client(username='alice')
        alice_api = conjur.api.status_api.StatusApi(alice_client)

        with self.assertRaises(conjur.exceptions.ApiException) as context:
            alice_api.get_gcp_authenticator_status(self.account)

        self.assertEqual(context.exception.status, 403)

    def test_get_gcp_authenticator_status_500(self):
        """Test case for get_gcp_authenticator_status 500 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_gcp_authenticator_status(self.account)

        self.assertEqual(context.exception.status, 500)

    def test_get_authenticators_200(self):
        """Test case for authenticators 200 response"""
        authenticators, status, _ = self.api.get_authenticators_with_http_info()

        self.assertEqual(status, 200)
        self.assertIsInstance(
            authenticators,
            conjur.models.authenticators_response.AuthenticatorsResponse
        )

        for i in AUTHENTICATOR_FIELDS:
            lst  = getattr(authenticators, i)
            self.assertIsInstance(lst, list)
            # authn is the default authenticator so it will always be installed and enabled
            self.assertIn('authn', lst)

            for value in lst:
                self.assertIsInstance(value, str)
                self.assertNotEqual(value, '')

    def test_set_request_id(self):
        """Test case for setting the request ID on an endpoint"""
        request_id = 'testing'
        _, status, headers = self.api.get_authenticators_with_http_info(x_request_id=request_id)

        self.assertEqual(status, 200)
        self.assertEqual(headers['X-Request-Id'], request_id)

    @unittest.skipUnless(api_config.ENTERPRISE_TESTS, "Endpoint not available in Conjur")
    def test_health_200(self):
        """Test case for Enterprise health 200 response"""
        resp, status, _ = self.api.health_with_http_info()
        status_keys = ['services', 'audit', 'database']

        self.assertEqual(status, 200)
        self.assertTrue(resp['ok'])
        for i in status_keys:
            self.assertTrue(resp[i]['ok'])

    @unittest.skipUnless(api_config.ENTERPRISE_TESTS, "Endpoint not available in Conjur")
    def test_remote_health_200(self):
        """Test case for Enterprise remote health 200 response"""
        resp, status, _ = self.api.remote_health_with_http_info('conjur-master.mycompany.local')
        status_keys = ['services', 'audit', 'database']

        self.assertEqual(status, 200)
        self.assertTrue(resp['ok'])
        for i in status_keys:
            self.assertTrue(resp[i]['ok'])

    @unittest.skipUnless(api_config.ENTERPRISE_TESTS, "Endpoint not available in Conjur")
    def test_info_200(self):
        """Test case for Enterprise info 200 response"""
        resp, status, _ = self.api.info_with_http_info()
        keys = [
            'release',
            'version',
            'services',
            'container',
            'role',
            'configuration',
            'authenticators'
        ]

        self.assertEqual(status, 200)
        for i in keys:
            self.assertIsNotNone(getattr(resp, i))

if __name__ == '__main__':
    unittest.main()

from __future__ import absolute_import

import unittest
from unittest.mock import patch

import conjur

from . import api_config

AUTHENTICATOR_FIELDS = [
    "enabled",
    "installed",
    "configured",
]

WHOAMI_FIELDS = [
    "client_ip",
    "user_agent",
    "account",
    "username",
    "token_issued_at",
]

class TestStatusApi(api_config.ConfiguredTest):
    """StatusApi unit test stubs"""
    @classmethod
    def setup_webservice(cls):
        """loads the webservice policy into conjur and gets it setup"""
        policies_api = conjur.api.policies_api.PoliciesApi(cls.client)
        secrets_api = conjur.api.secrets_api.SecretsApi(cls.client)
        policies_api.update_policy(cls.account, 'root', api_config.get_webservice_policy())
        secrets_api.create_secret(
            cls.account,
            "variable",
            'conjur/authn-oidc/okta/provider-uri',
            body='invalid'
        )
        secrets_api.create_secret(
            cls.account,
            "variable",
            'conjur/authn-oidc/okta/id-token-user-property',
            body='admin'
        )

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.setup_webservice()

    def setUp(self):
        self.api = conjur.api.status_api.StatusApi(self.client)
        self.bad_auth_api = conjur.api.status_api.StatusApi(self.bad_auth_client)

    def test_who_am_i_200(self):
        """Test case for who_am_i 200 response"""
        result = self.api.who_am_i()

        self.assertIsInstance(result, conjur.models.who_am_i.WhoAmI)

        for i in WHOAMI_FIELDS:
            value = getattr(result, i)
            # Just make sure that all the attributes have a value assigned to them
            self.assertNotEqual(value, '')

    def test_who_am_i_401(self):
        """Test case for who_am_i 401 response"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.bad_auth_api.who_am_i()

        self.assertEqual(context.exception.status, 401)

    @unittest.skipIf(api_config.ENTERPRISE_TESTS,
                     'Dont support testing external authenticators on Enterprise currently')
    def test_get_service_authenticator_status_200(self):
        """Test case for get_service_authenticator_status 200 return code"""
        api_config.setup_oidc_webservice()
        resp, status, _ = self.api.get_service_authenticator_status_with_http_info(
            'authn-oidc',
            'test',
            self.account
        )
        self.assertEqual(resp.status, 'ok')
        self.assertEqual(status, 200)

        with patch.object(conjur.api_client.ApiClient, 'call_api', return_value=None) \
                as mock:
            self.api.get_service_authenticator_status('authn-azure', 'test', self.account)

        mock.assert_called_once_with(
            '/{authenticator}/{service_id}/{account}/status',
            'GET',
            {'account': self.account, 'authenticator': 'authn-azure', 'service_id': 'test'},
            [],
            {'Accept': 'application/json'},
            body=None,
            files={},
            post_params=[],
            response_type='AuthenticatorStatus',
            auth_settings=['conjurAuth'],
            async_req=None,
            _return_http_data_only=True,
            _preload_content=True,
            _request_timeout=None,
            collection_formats={}
        )

    def test_get_service_authenticator_status_403(self):
        """Test case for get_service_authenticator_status 403 return code"""
        alice_client = api_config.get_api_client(username='alice')
        alice_api = conjur.api.status_api.StatusApi(alice_client)

        with self.assertRaises(conjur.exceptions.ApiException) as context:
            alice_api.get_service_authenticator_status('authn-oidc', 'okta', self.account)

        self.assertEqual(context.exception.status, 403)

    def test_get_service_authenticator_status_404(self):
        """Test case for get_service_authenticator_status 404 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_service_authenticator_status('nonexist', 'test', self.account)

        self.assertEqual(context.exception.status, 404)

    def test_get_service_authenticator_status_500(self):
        """Test case for get_service_authenticator_status 500 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_service_authenticator_status('authn-oidc', 'okta', self.account)

        self.assertEqual(context.exception.status, 500)

    def test_get_service_authenticator_status_501(self):
        """Test case for get_service_authenticator_status 500 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_service_authenticator_status('authn-ldap', 'test', self.account)

        self.assertEqual(context.exception.status, 501)

    def test_get_gcp_authenticator_status_200(self):
        """Test case for get_gcp_authenticator_status 200 return code"""
        with patch.object(conjur.api_client.ApiClient, 'call_api', return_value=None) \
                as mock:
            self.api.get_gcp_authenticator_status(self.account)

        mock.assert_called_once_with(
            '/authn-gcp/{account}/status',
            'GET',
            { 'account': self.account },
            [],
            { 'Accept': 'application/json' },
            body=None,
            files={},
            post_params=[],
            response_type='AuthenticatorStatus',
            auth_settings=['conjurAuth'],
            async_req=None,
            _return_http_data_only=True,
            _preload_content=True,
            _request_timeout=None,
            collection_formats={}
        )

    def test_get_gcp_authenticator_status_403(self):
        """Test case for get_gcp_authenticator_status 403 return code"""
        alice_client = api_config.get_api_client(username='alice')
        alice_api = conjur.api.status_api.StatusApi(alice_client)

        with self.assertRaises(conjur.exceptions.ApiException) as context:
            alice_api.get_gcp_authenticator_status(self.account)

        self.assertEqual(context.exception.status, 403)

    def test_get_gcp_authenticator_status_500(self):
        """Test case for get_gcp_authenticator_status 500 return code"""
        with self.assertRaises(conjur.exceptions.ApiException) as context:
            self.api.get_gcp_authenticator_status(self.account)

        self.assertEqual(context.exception.status, 500)

    def test_get_authenticators_200(self):
        """Test case for authenticators 200 response"""
        authenticators, status, _ = self.api.get_authenticators_with_http_info()

        self.assertEqual(status, 200)
        self.assertIsInstance(
            authenticators,
            conjur.models.authenticators_response.AuthenticatorsResponse
        )

        for i in AUTHENTICATOR_FIELDS:
            lst  = getattr(authenticators, i)
            self.assertIsInstance(lst, list)
            # authn is the default authenticator so it will always be installed and enabled
            self.assertIn('authn', lst)

            for value in lst:
                self.assertIsInstance(value, str)
                self.assertNotEqual(value, '')

    def test_set_request_id(self):
        """Test case for setting the request ID on an endpoint"""
        request_id = 'testing'
        _, status, headers = self.api.get_authenticators_with_http_info(x_request_id=request_id)

        self.assertEqual(status, 200)
        self.assertEqual(headers['X-Request-Id'], request_id)

    @unittest.skipUnless(api_config.ENTERPRISE_TESTS, "Endpoint not available in Conjur")
    def test_health_200(self):
        """Test case for Enterprise health 200 response"""
        resp, status, _ = self.api.health_with_http_info()
        status_keys = ['services', 'audit', 'database']

        self.assertEqual(status, 200)
        self.assertTrue(resp['ok'])
        for i in status_keys:
            self.assertTrue(resp[i]['ok'])

    @unittest.skipUnless(api_config.ENTERPRISE_TESTS, "Endpoint not available in Conjur")
    def test_remote_health_200(self):
        """Test case for Enterprise remote health 200 response"""
        resp, status, _ = self.api.remote_health_with_http_info('conjur-master.mycompany.local')
        status_keys = ['services', 'audit', 'database']

        self.assertEqual(status, 200)
        self.assertTrue(resp['ok'])
        for i in status_keys:
            self.assertTrue(resp[i]['ok'])

    @unittest.skipUnless(api_config.ENTERPRISE_TESTS, "Endpoint not available in Conjur")
    def test_info_200(self):
        """Test case for Enterprise info 200 response"""
        resp, status, _ = self.api.info_with_http_info()
        keys = [
            'release',
            'version',
            'services',
            'container',
            'role',
            'configuration',
            'authenticators'
        ]

        self.assertEqual(status, 200)
        for i in keys:
            self.assertIsNotNone(getattr(resp, i))

if __name__ == '__main__':
    unittest.main()
StatusApi unit test stubs loads the webservice policy into conjur and gets it setup Test case for who_am_i 200 response # Just make sure that all the attributes have a value assigned to them Test case for who_am_i 401 response Test case for get_service_authenticator_status 200 return code Test case for get_service_authenticator_status 403 return code Test case for get_service_authenticator_status 404 return code Test case for get_service_authenticator_status 500 return code Test case for get_service_authenticator_status 500 return code Test case for get_gcp_authenticator_status 200 return code Test case for get_gcp_authenticator_status 403 return code Test case for get_gcp_authenticator_status 500 return code Test case for authenticators 200 response # authn is the default authenticator so it will always be installed and enabled Test case for setting the request ID on an endpoint Test case for Enterprise health 200 response Test case for Enterprise remote health 200 response Test case for Enterprise info 200 response