diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index 1eaf926..0de9ff0 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -57,7 +57,6 @@ jobs: env: ACCOUNT_ID: ${{ secrets.ACCOUNT_ID }} FINDINGS_ENV: ${{ secrets.FINDINGS_ENV }} - NOTIFICATIONS_ENV: ${{ secrets.NOTIFICATIONS_ENV }} CONFIGURATION_GOVERNANCE_ENV: ${{ secrets.CONFIGURATION_GOVERNANCE_ENV }} RESOURCE_GROUP_ID: ${{ secrets.RESOURCE_GROUP_ID }} run: build/testScript.sh diff --git a/README.md b/README.md index 7346988..96baf9f 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,6 @@ IBM Cloud services: Service Name | Module Name | Imported Class Name --- | --- | --- [Findings](https://cloud.ibm.com/apidocs/security-compliance/findings) | findings_v1 | FindingsV1 -[Notifications](https://cloud.ibm.com/apidocs/security-compliance/si-notifications) | notifications_v1 | NotificationsV1 [Configuration Governance](https://cloud.ibm.com/apidocs/security-compliance/config) | configuration_governance_v1 | ConfigurationGovernanceV1 ## Prerequisites diff --git a/build/testScript.sh b/build/testScript.sh index 8a94e7b..1cc7864 100755 --- a/build/testScript.sh +++ b/build/testScript.sh @@ -4,6 +4,5 @@ set -euo pipefail curl https://us-south.functions.appdomain.cloud/api/v1/web/e6b54af6-ab44-4149-a8e4-e906dcc58136/default/secadvstg-location-shift.json echo "${FINDINGS_ENV}" | base64 -d >> findings_v1.env -echo "${NOTIFICATIONS_ENV}" | base64 -d >> notifications_v1.env echo "${CONFIGURATION_GOVERNANCE_ENV}" | base64 -d >> configuration_governance_v1.env python -m pytest test/integration diff --git a/ibm_scc/notifications_v1.py b/ibm_scc/notifications_v1.py deleted file mode 100644 index 266dbde..0000000 --- a/ibm_scc/notifications_v1.py +++ /dev/null @@ -1,1373 +0,0 @@ -# coding: utf-8 - -# (C) Copyright IBM Corp. 2021. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# IBM OpenAPI SDK Code Generator Version: 3.32.0-4c6a3129-20210514-210323 - -""" -API specification for the Notifications service. -""" - -from enum import Enum -from typing import Dict, List -import json - -from ibm_cloud_sdk_core import BaseService, DetailedResponse -from ibm_cloud_sdk_core.authenticators.authenticator import Authenticator -from ibm_cloud_sdk_core.get_authenticator import get_authenticator_from_environment -from ibm_cloud_sdk_core.utils import convert_model - -from .common import get_sdk_headers - -############################################################################## -# Service -############################################################################## - -class NotificationsV1(BaseService): - """The Notifications V1 service.""" - - DEFAULT_SERVICE_URL = 'https://us-south.secadvisor.cloud.ibm.com/notifications' - DEFAULT_SERVICE_NAME = 'notifications' - - @classmethod - def new_instance(cls, - service_name: str = DEFAULT_SERVICE_NAME, - ) -> 'NotificationsV1': - """ - Return a new client for the Notifications service using the specified - parameters and external configuration. - """ - authenticator = get_authenticator_from_environment(service_name) - service = cls( - authenticator - ) - service.configure_service(service_name) - return service - - def __init__(self, - authenticator: Authenticator = None, - ) -> None: - """ - Construct a new client for the Notifications service. - - :param Authenticator authenticator: The authenticator specifies the authentication mechanism. - Get up to date information from https://github.com/IBM/python-sdk-core/blob/master/README.md - about initializing the authenticator of your choice. - """ - BaseService.__init__(self, - service_url=self.DEFAULT_SERVICE_URL, - authenticator=authenticator) - - - ######################### - # notificationChannel - ######################### - - - def list_all_channels(self, - account_id: str, - *, - transaction_id: str = None, - limit: int = None, - skip: int = None, - **kwargs - ) -> DetailedResponse: - """ - list all channels. - - list all channels under this account. - - :param str account_id: Account ID. - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param int limit: (optional) Limit the number of the returned documents to - the specified number. - :param int skip: (optional) The offset is the index of the item from which - you want to start returning data from. Default is 0. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `ChannelsList` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='list_all_channels') - headers.update(sdk_headers) - - params = { - 'limit': limit, - 'skip': skip - } - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id'] - path_param_values = self.encode_path_vars(account_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/channels'.format(**path_param_dict) - request = self.prepare_request(method='GET', - url=url, - headers=headers, - params=params) - - response = self.send(request) - return response - - - def create_notification_channel(self, - account_id: str, - name: str, - type: str, - endpoint: str, - *, - description: str = None, - severity: List[str] = None, - enabled: bool = None, - alert_source: List['NotificationChannelAlertSourceItem'] = None, - transaction_id: str = None, - **kwargs - ) -> DetailedResponse: - """ - create notification channel. - - create notification channel. - - :param str account_id: Account ID. - :param str name: The name of the notification channel in the form - "v1/{account_id}/notifications/channelName". - :param str type: Type of callback URL. - :param str endpoint: The callback URL which receives the notification. - :param str description: (optional) A one sentence description of this - `Channel`. - :param List[str] severity: (optional) Severity of the notification to be - received. - :param bool enabled: (optional) Channel is enabled or not. Default is - disabled. - :param List[NotificationChannelAlertSourceItem] alert_source: (optional) - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `ChannelInfo` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - if name is None: - raise ValueError('name must be provided') - if type is None: - raise ValueError('type must be provided') - if endpoint is None: - raise ValueError('endpoint must be provided') - if alert_source is not None: - alert_source = [convert_model(x) for x in alert_source] - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='create_notification_channel') - headers.update(sdk_headers) - - data = { - 'name': name, - 'type': type, - 'endpoint': endpoint, - 'description': description, - 'severity': severity, - 'enabled': enabled, - 'alert_source': alert_source - } - data = {k: v for (k, v) in data.items() if v is not None} - data = json.dumps(data) - headers['content-type'] = 'application/json' - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id'] - path_param_values = self.encode_path_vars(account_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/channels'.format(**path_param_dict) - request = self.prepare_request(method='POST', - url=url, - headers=headers, - data=data) - - response = self.send(request) - return response - - - def delete_notification_channels(self, - account_id: str, - request_body: List[str], - *, - transaction_id: str = None, - **kwargs - ) -> DetailedResponse: - """ - bulk delete of channels. - - bulk delete of channels. - - :param str account_id: Account ID. - :param List[str] request_body: Body for bulk delete notification channels. - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `ChannelsDelete` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - if request_body is None: - raise ValueError('request_body must be provided') - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='delete_notification_channels') - headers.update(sdk_headers) - - data = json.dumps(request_body) - headers['content-type'] = 'application/json' - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id'] - path_param_values = self.encode_path_vars(account_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/channels'.format(**path_param_dict) - request = self.prepare_request(method='DELETE', - url=url, - headers=headers, - data=data) - - response = self.send(request) - return response - - - def delete_notification_channel(self, - account_id: str, - channel_id: str, - *, - transaction_id: str = None, - **kwargs - ) -> DetailedResponse: - """ - delete the details of a specific channel. - - delete the details of a specific channel. - - :param str account_id: Account ID. - :param str channel_id: Channel ID. - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `ChannelDelete` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - if channel_id is None: - raise ValueError('channel_id must be provided') - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='delete_notification_channel') - headers.update(sdk_headers) - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id', 'channel_id'] - path_param_values = self.encode_path_vars(account_id, channel_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/channels/{channel_id}'.format(**path_param_dict) - request = self.prepare_request(method='DELETE', - url=url, - headers=headers) - - response = self.send(request) - return response - - - def get_notification_channel(self, - account_id: str, - channel_id: str, - *, - transaction_id: str = None, - **kwargs - ) -> DetailedResponse: - """ - get the details of a specific channel. - - get the details of a specific channel. - - :param str account_id: Account ID. - :param str channel_id: Channel ID. - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `ChannelGet` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - if channel_id is None: - raise ValueError('channel_id must be provided') - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='get_notification_channel') - headers.update(sdk_headers) - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id', 'channel_id'] - path_param_values = self.encode_path_vars(account_id, channel_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/channels/{channel_id}'.format(**path_param_dict) - request = self.prepare_request(method='GET', - url=url, - headers=headers) - - response = self.send(request) - return response - - - def update_notification_channel(self, - account_id: str, - channel_id: str, - name: str, - type: str, - endpoint: str, - *, - description: str = None, - severity: List[str] = None, - enabled: bool = None, - alert_source: List['NotificationChannelAlertSourceItem'] = None, - transaction_id: str = None, - **kwargs - ) -> DetailedResponse: - """ - update notification channel. - - update notification channel. - - :param str account_id: Account ID. - :param str channel_id: Channel ID. - :param str name: The name of the notification channel in the form - "v1/{account_id}/notifications/channelName". - :param str type: Type of callback URL. - :param str endpoint: The callback URL which receives the notification. - :param str description: (optional) A one sentence description of this - `Channel`. - :param List[str] severity: (optional) Severity of the notification to be - received. - :param bool enabled: (optional) Channel is enabled or not. Default is - disabled. - :param List[NotificationChannelAlertSourceItem] alert_source: (optional) - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `ChannelInfo` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - if channel_id is None: - raise ValueError('channel_id must be provided') - if name is None: - raise ValueError('name must be provided') - if type is None: - raise ValueError('type must be provided') - if endpoint is None: - raise ValueError('endpoint must be provided') - if alert_source is not None: - alert_source = [convert_model(x) for x in alert_source] - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='update_notification_channel') - headers.update(sdk_headers) - - data = { - 'name': name, - 'type': type, - 'endpoint': endpoint, - 'description': description, - 'severity': severity, - 'enabled': enabled, - 'alert_source': alert_source - } - data = {k: v for (k, v) in data.items() if v is not None} - data = json.dumps(data) - headers['content-type'] = 'application/json' - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id', 'channel_id'] - path_param_values = self.encode_path_vars(account_id, channel_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/channels/{channel_id}'.format(**path_param_dict) - request = self.prepare_request(method='PUT', - url=url, - headers=headers, - data=data) - - response = self.send(request) - return response - - - def test_notification_channel(self, - account_id: str, - channel_id: str, - *, - transaction_id: str = None, - **kwargs - ) -> DetailedResponse: - """ - test notification channel. - - test a nofication channel under this account. - - :param str account_id: Account ID. - :param str channel_id: Channel ID. - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `TestChannel` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - if channel_id is None: - raise ValueError('channel_id must be provided') - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='test_notification_channel') - headers.update(sdk_headers) - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id', 'channel_id'] - path_param_values = self.encode_path_vars(account_id, channel_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/channels/{channel_id}/test'.format(**path_param_dict) - request = self.prepare_request(method='GET', - url=url, - headers=headers) - - response = self.send(request) - return response - - - def get_public_key(self, - account_id: str, - *, - transaction_id: str = None, - **kwargs - ) -> DetailedResponse: - """ - fetch notifications public key. - - fetch public key to decrypt messages in notification payload. - - :param str account_id: Account ID. - :param str transaction_id: (optional) The transaction id for the request in - uuid v4 format. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `PublicKeyGet` object - """ - - if account_id is None: - raise ValueError('account_id must be provided') - headers = { - 'Transaction-Id': transaction_id - } - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='get_public_key') - headers.update(sdk_headers) - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - path_param_keys = ['account_id'] - path_param_values = self.encode_path_vars(account_id) - path_param_dict = dict(zip(path_param_keys, path_param_values)) - url = '/v1/{account_id}/notifications/public_key'.format(**path_param_dict) - request = self.prepare_request(method='GET', - url=url, - headers=headers) - - response = self.send(request) - return response - - -############################################################################## -# Models -############################################################################## - - -class ChannelAlertSourceItem(): - """ - The providers that act as alert sources and the potential findings that can be flagged - as alerts. - - :attr str provider_name: (optional) The providers that you can receive alerts - for. To view your available providers, you can call the - /v1/{account_id}/providers endpoint of the Findings API. - :attr List[object] finding_types: (optional) The types of findings for each - provider that you want to receive alerts for. Options are dependent upon the - provider that you select. Depending on that selection, some available options - include `image_with_vulnerabilities`, `anonym_server`, `server_suspected_ratio`, - `appid`, `cos`, `expired_cert`, and `expiring_1day_cert`For a full list of - available finding types, see [the docs](/docs/). - """ - - def __init__(self, - *, - provider_name: str = None, - finding_types: List[object] = None) -> None: - """ - Initialize a ChannelAlertSourceItem object. - - :param str provider_name: (optional) The providers that you can receive - alerts for. To view your available providers, you can call the - /v1/{account_id}/providers endpoint of the Findings API. - :param List[object] finding_types: (optional) The types of findings for - each provider that you want to receive alerts for. Options are dependent - upon the provider that you select. Depending on that selection, some - available options include `image_with_vulnerabilities`, `anonym_server`, - `server_suspected_ratio`, `appid`, `cos`, `expired_cert`, and - `expiring_1day_cert`For a full list of available finding types, see [the - docs](/docs/). - """ - self.provider_name = provider_name - self.finding_types = finding_types - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ChannelAlertSourceItem': - """Initialize a ChannelAlertSourceItem object from a json dictionary.""" - args = {} - if 'provider_name' in _dict: - args['provider_name'] = _dict.get('provider_name') - if 'finding_types' in _dict: - args['finding_types'] = _dict.get('finding_types') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ChannelAlertSourceItem object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'provider_name') and self.provider_name is not None: - _dict['provider_name'] = self.provider_name - if hasattr(self, 'finding_types') and self.finding_types is not None: - _dict['finding_types'] = self.finding_types - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ChannelAlertSourceItem object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ChannelAlertSourceItem') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ChannelAlertSourceItem') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - - class ProviderNameEnum(str, Enum): - """ - The providers that you can receive alerts for. To view your available providers, - you can call the /v1/{account_id}/providers endpoint of the Findings API. - """ - VA = 'VA' - NA = 'NA' - ATA = 'ATA' - CERT = 'CERT' - ALL = 'ALL' - - -class ChannelDelete(): - """ - The returned response when a channel is deleted. - - :attr str channel_id: (optional) The ID of the deleted channel. - :attr str message: (optional) response message. - """ - - def __init__(self, - *, - channel_id: str = None, - message: str = None) -> None: - """ - Initialize a ChannelDelete object. - - :param str channel_id: (optional) The ID of the deleted channel. - :param str message: (optional) response message. - """ - self.channel_id = channel_id - self.message = message - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ChannelDelete': - """Initialize a ChannelDelete object from a json dictionary.""" - args = {} - if 'channel_id' in _dict: - args['channel_id'] = _dict.get('channel_id') - if 'message' in _dict: - args['message'] = _dict.get('message') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ChannelDelete object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'channel_id') and self.channel_id is not None: - _dict['channel_id'] = self.channel_id - if hasattr(self, 'message') and self.message is not None: - _dict['message'] = self.message - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ChannelDelete object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ChannelDelete') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ChannelDelete') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class ChannelGet(): - """ - The returned response when get channel is run. - - :attr Channel channel: (optional) get channel. - """ - - def __init__(self, - *, - channel: 'Channel' = None) -> None: - """ - Initialize a ChannelGet object. - - :param Channel channel: (optional) get channel. - """ - self.channel = channel - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ChannelGet': - """Initialize a ChannelGet object from a json dictionary.""" - args = {} - if 'channel' in _dict: - args['channel'] = Channel.from_dict(_dict.get('channel')) - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ChannelGet object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'channel') and self.channel is not None: - _dict['channel'] = self.channel.to_dict() - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ChannelGet object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ChannelGet') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ChannelGet') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class ChannelInfo(): - """ - The returned response when a channel is created or updated. - - :attr str channel_id: (optional) The ID of the created channel. - :attr int status_code: (optional) response code. - """ - - def __init__(self, - *, - channel_id: str = None, - status_code: int = None) -> None: - """ - Initialize a ChannelInfo object. - - :param str channel_id: (optional) The ID of the created channel. - :param int status_code: (optional) response code. - """ - self.channel_id = channel_id - self.status_code = status_code - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ChannelInfo': - """Initialize a ChannelInfo object from a json dictionary.""" - args = {} - if 'channel_id' in _dict: - args['channel_id'] = _dict.get('channel_id') - if 'status_code' in _dict: - args['status_code'] = _dict.get('status_code') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ChannelInfo object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'channel_id') and self.channel_id is not None: - _dict['channel_id'] = self.channel_id - if hasattr(self, 'status_code') and self.status_code is not None: - _dict['status_code'] = self.status_code - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ChannelInfo object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ChannelInfo') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ChannelInfo') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class ChannelSeverity(): - """ - The severity of the notification. - - :attr bool critical: (optional) Critical severity. - :attr bool high: (optional) High severity. - :attr bool medium: (optional) Medium severity. - :attr bool low: (optional) Low severity. - """ - - def __init__(self, - *, - critical: bool = None, - high: bool = None, - medium: bool = None, - low: bool = None) -> None: - """ - Initialize a ChannelSeverity object. - - :param bool critical: (optional) Critical severity. - :param bool high: (optional) High severity. - :param bool medium: (optional) Medium severity. - :param bool low: (optional) Low severity. - """ - self.critical = critical - self.high = high - self.medium = medium - self.low = low - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ChannelSeverity': - """Initialize a ChannelSeverity object from a json dictionary.""" - args = {} - if 'critical' in _dict: - args['critical'] = _dict.get('critical') - if 'high' in _dict: - args['high'] = _dict.get('high') - if 'medium' in _dict: - args['medium'] = _dict.get('medium') - if 'low' in _dict: - args['low'] = _dict.get('low') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ChannelSeverity object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'critical') and self.critical is not None: - _dict['critical'] = self.critical - if hasattr(self, 'high') and self.high is not None: - _dict['high'] = self.high - if hasattr(self, 'medium') and self.medium is not None: - _dict['medium'] = self.medium - if hasattr(self, 'low') and self.low is not None: - _dict['low'] = self.low - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ChannelSeverity object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ChannelSeverity') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ChannelSeverity') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class ChannelsDelete(): - """ - The returned response when more than one channel is deleted. - - :attr str message: (optional) response message. - """ - - def __init__(self, - *, - message: str = None) -> None: - """ - Initialize a ChannelsDelete object. - - :param str message: (optional) response message. - """ - self.message = message - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ChannelsDelete': - """Initialize a ChannelsDelete object from a json dictionary.""" - args = {} - if 'message' in _dict: - args['message'] = _dict.get('message') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ChannelsDelete object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'message') and self.message is not None: - _dict['message'] = self.message - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ChannelsDelete object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ChannelsDelete') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ChannelsDelete') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class ChannelsList(): - """ - Available channels in your account are listed. - - :attr List[Channel] channels: (optional) List of channels. - """ - - def __init__(self, - *, - channels: List['Channel'] = None) -> None: - """ - Initialize a ChannelsList object. - - :param List[Channel] channels: (optional) List of channels. - """ - self.channels = channels - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ChannelsList': - """Initialize a ChannelsList object from a json dictionary.""" - args = {} - if 'channels' in _dict: - args['channels'] = [Channel.from_dict(x) for x in _dict.get('channels')] - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ChannelsList object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'channels') and self.channels is not None: - _dict['channels'] = [x.to_dict() for x in self.channels] - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ChannelsList object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ChannelsList') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ChannelsList') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class NotificationChannelAlertSourceItem(): - """ - The alert sources. They identify the providers and their finding types which makes the - findings available to Security Advisor. - - :attr str provider_name: Below is a list of builtin providers that you can - select in addition to the ones you obtain by calling Findings API - /v1/{account_id}/providers : - | provider_name | The source they represent | - |-----|-----| - | VA | Vulnerable image findings| - | NA | Network Insights findings| - | ATA | Activity Insights findings| - | CERT | Certificate Manager findings| - | ALL | Special provider name to represent all the providers. Its mutually - exclusive with other providers meaning either you choose ALL or you don't|. - :attr List[str] finding_types: (optional) An array of the finding types of the - provider_name or "ALL" to specify all finding types under that provider Below is - a list of supported finding types for each built in providers - | provider_name | Supported finding types | - |-----|-----| - | VA | "image_with_vulnerabilities", "image_with_config_issues"| - | NA | "anonym_server", "malware_server", "bot_server", "miner_server", - "server_suspected_ratio", "server_response", "data_extrusion", - "server_weaponized_total"| - | ATA | "appid", "cos", "iks", "iam", "kms", "cert", "account", "app"| - | CERT | "expired_cert", "expiring_1day_cert", "expiring_10day_cert", - "expiring_30day_cert", "expiring_60day_cert", "expiring_90day_cert"| - | ALL | "ALL"|. - """ - - def __init__(self, - provider_name: str, - *, - finding_types: List[str] = None) -> None: - """ - Initialize a NotificationChannelAlertSourceItem object. - - :param str provider_name: Below is a list of builtin providers that you can - select in addition to the ones you obtain by calling Findings API - /v1/{account_id}/providers : - | provider_name | The source they represent | - |-----|-----| - | VA | Vulnerable image findings| - | NA | Network Insights findings| - | ATA | Activity Insights findings| - | CERT | Certificate Manager findings| - | ALL | Special provider name to represent all the providers. Its mutually - exclusive with other providers meaning either you choose ALL or you don't|. - :param List[str] finding_types: (optional) An array of the finding types of - the provider_name or "ALL" to specify all finding types under that provider - Below is a list of supported finding types for each built in providers - | provider_name | Supported finding types | - |-----|-----| - | VA | "image_with_vulnerabilities", "image_with_config_issues"| - | NA | "anonym_server", "malware_server", "bot_server", "miner_server", - "server_suspected_ratio", "server_response", "data_extrusion", - "server_weaponized_total"| - | ATA | "appid", "cos", "iks", "iam", "kms", "cert", "account", "app"| - | CERT | "expired_cert", "expiring_1day_cert", "expiring_10day_cert", - "expiring_30day_cert", "expiring_60day_cert", "expiring_90day_cert"| - | ALL | "ALL"|. - """ - self.provider_name = provider_name - self.finding_types = finding_types - - @classmethod - def from_dict(cls, _dict: Dict) -> 'NotificationChannelAlertSourceItem': - """Initialize a NotificationChannelAlertSourceItem object from a json dictionary.""" - args = {} - if 'provider_name' in _dict: - args['provider_name'] = _dict.get('provider_name') - else: - raise ValueError('Required property \'provider_name\' not present in NotificationChannelAlertSourceItem JSON') - if 'finding_types' in _dict: - args['finding_types'] = _dict.get('finding_types') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a NotificationChannelAlertSourceItem object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'provider_name') and self.provider_name is not None: - _dict['provider_name'] = self.provider_name - if hasattr(self, 'finding_types') and self.finding_types is not None: - _dict['finding_types'] = self.finding_types - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this NotificationChannelAlertSourceItem object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'NotificationChannelAlertSourceItem') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'NotificationChannelAlertSourceItem') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class PublicKeyGet(): - """ - PublicKeyGet. - - :attr str public_key: - """ - - def __init__(self, - public_key: str) -> None: - """ - Initialize a PublicKeyGet object. - - :param str public_key: - """ - self.public_key = public_key - - @classmethod - def from_dict(cls, _dict: Dict) -> 'PublicKeyGet': - """Initialize a PublicKeyGet object from a json dictionary.""" - args = {} - if 'public_key' in _dict: - args['public_key'] = _dict.get('public_key') - else: - raise ValueError('Required property \'public_key\' not present in PublicKeyGet JSON') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a PublicKeyGet object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'public_key') and self.public_key is not None: - _dict['public_key'] = self.public_key - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this PublicKeyGet object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'PublicKeyGet') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'PublicKeyGet') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class TestChannel(): - """ - The returned response when a webhook is tested for a channel. - - :attr str test: (optional) response status. - """ - - def __init__(self, - *, - test: str = None) -> None: - """ - Initialize a TestChannel object. - - :param str test: (optional) response status. - """ - self.test = test - - @classmethod - def from_dict(cls, _dict: Dict) -> 'TestChannel': - """Initialize a TestChannel object from a json dictionary.""" - args = {} - if 'test' in _dict: - args['test'] = _dict.get('test') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a TestChannel object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'test') and self.test is not None: - _dict['test'] = self.test - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this TestChannel object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'TestChannel') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'TestChannel') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class Channel(): - """ - Response including channels. - - :attr str channel_id: (optional) A unique ID for the channel. - :attr str name: (optional) The name of the notification channel in the form - "v1/{account_id}/notifications/channelName". - :attr str description: (optional) A one sentence description of this `Channel`. - :attr str type: (optional) Type of callback URL. - :attr ChannelSeverity severity: (optional) The severity of the notification. - :attr str endpoint: (optional) The callback URL which receives the notification. - :attr bool enabled: (optional) Whether the channel is enabled. The default is - disabled. - :attr List[ChannelAlertSourceItem] alert_source: (optional) - :attr str frequency: (optional) - """ - - def __init__(self, - *, - channel_id: str = None, - name: str = None, - description: str = None, - type: str = None, - severity: 'ChannelSeverity' = None, - endpoint: str = None, - enabled: bool = None, - alert_source: List['ChannelAlertSourceItem'] = None, - frequency: str = None) -> None: - """ - Initialize a Channel object. - - :param str channel_id: (optional) A unique ID for the channel. - :param str name: (optional) The name of the notification channel in the - form "v1/{account_id}/notifications/channelName". - :param str description: (optional) A one sentence description of this - `Channel`. - :param str type: (optional) Type of callback URL. - :param ChannelSeverity severity: (optional) The severity of the - notification. - :param str endpoint: (optional) The callback URL which receives the - notification. - :param bool enabled: (optional) Whether the channel is enabled. The default - is disabled. - :param List[ChannelAlertSourceItem] alert_source: (optional) - :param str frequency: (optional) - """ - self.channel_id = channel_id - self.name = name - self.description = description - self.type = type - self.severity = severity - self.endpoint = endpoint - self.enabled = enabled - self.alert_source = alert_source - self.frequency = frequency - - @classmethod - def from_dict(cls, _dict: Dict) -> 'Channel': - """Initialize a Channel object from a json dictionary.""" - args = {} - if 'channel_id' in _dict: - args['channel_id'] = _dict.get('channel_id') - if 'name' in _dict: - args['name'] = _dict.get('name') - if 'description' in _dict: - args['description'] = _dict.get('description') - if 'type' in _dict: - args['type'] = _dict.get('type') - if 'severity' in _dict: - args['severity'] = ChannelSeverity.from_dict(_dict.get('severity')) - if 'endpoint' in _dict: - args['endpoint'] = _dict.get('endpoint') - if 'enabled' in _dict: - args['enabled'] = _dict.get('enabled') - if 'alert_source' in _dict: - args['alert_source'] = [ChannelAlertSourceItem.from_dict(x) for x in _dict.get('alert_source')] - if 'frequency' in _dict: - args['frequency'] = _dict.get('frequency') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a Channel object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'channel_id') and self.channel_id is not None: - _dict['channel_id'] = self.channel_id - if hasattr(self, 'name') and self.name is not None: - _dict['name'] = self.name - if hasattr(self, 'description') and self.description is not None: - _dict['description'] = self.description - if hasattr(self, 'type') and self.type is not None: - _dict['type'] = self.type - if hasattr(self, 'severity') and self.severity is not None: - _dict['severity'] = self.severity.to_dict() - if hasattr(self, 'endpoint') and self.endpoint is not None: - _dict['endpoint'] = self.endpoint - if hasattr(self, 'enabled') and self.enabled is not None: - _dict['enabled'] = self.enabled - if hasattr(self, 'alert_source') and self.alert_source is not None: - _dict['alert_source'] = [x.to_dict() for x in self.alert_source] - if hasattr(self, 'frequency') and self.frequency is not None: - _dict['frequency'] = self.frequency - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this Channel object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'Channel') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'Channel') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - - class TypeEnum(str, Enum): - """ - Type of callback URL. - """ - WEBHOOK = 'Webhook' - diff --git a/test/integration/test_notifications_v1.py b/test/integration/test_notifications_v1.py deleted file mode 100644 index 01508e9..0000000 --- a/test/integration/test_notifications_v1.py +++ /dev/null @@ -1,218 +0,0 @@ -# -*- coding: utf-8 -*- -# (C) Copyright IBM Corp. 2021. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -""" -Integration Tests for NotificationsV1 -""" - -import os -import pytest -import time -from ibm_cloud_sdk_core import * -from ibm_scc.notifications_v1 import * - -# Config file name -config_file = 'notifications_v1.env' -account_id = os.getenv("ACCOUNT_ID") -testString = "testString" -channel_id = "" -identifier = "py-{0}".format(str(time.time()).split(".")[0]) - -class TestNotificationsV1(): - """ - Integration Test Class for NotificationsV1 - """ - - @classmethod - def setup_class(cls): - if os.path.exists(config_file): - os.environ['IBM_CREDENTIALS_FILE'] = config_file - - cls.notifications_service = NotificationsV1.new_instance( - ) - assert cls.notifications_service is not None - - cls.config = read_external_sources( - NotificationsV1.DEFAULT_SERVICE_NAME) - assert cls.config is not None - - print('Setup complete.') - - @classmethod - def teardown_class(cls): - if os.path.exists(config_file): - os.environ['IBM_CREDENTIALS_FILE'] = config_file - - cls.notifications_service = NotificationsV1.new_instance( - ) - assert cls.notifications_service is not None - - cls.config = read_external_sources( - NotificationsV1.DEFAULT_SERVICE_NAME) - assert cls.config is not None - - print('Setup complete.') - print(f'cleaning up account: {account_id}\n') - list_all_channels_response = cls.notifications_service.list_all_channels( - account_id=account_id, - ) - for channel in list_all_channels_response.get_result()['channels']: - if channel['channel_id'] == channel_id: - cls.notifications_service.delete_notification_channel( - account_id=account_id, - channel_id=channel['channel_id'], - ) - print("cleanup was successful\n") - - needscredentials = pytest.mark.skipif( - not os.path.exists(config_file), reason="External configuration not available, skipping..." - ) - - @needscredentials - def test_list_all_channels(self): - - list_all_channels_response = self.notifications_service.list_all_channels( - account_id=account_id, - ) - - assert list_all_channels_response.get_status_code() == 200 - channels_list = list_all_channels_response.get_result() - assert channels_list is not None - - @needscredentials - def test_create_notification_channel(self): - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = { - 'provider_name': 'VA', - 'finding_types': ['image_with_vulnerabilities'] - } - - create_notification_channel_response = self.notifications_service.create_notification_channel( - account_id=account_id, - name=f'testString-{identifier}', - type='Webhook', - endpoint='https://us-south.functions.appdomain.cloud/api/v1/web/e6b54af6-ab44-4149-a8e4-e906dcc58136/default/sdk-integration.json', - description=testString, - severity=['low'], - enabled=True, - alert_source=[notification_channel_alert_source_item_model], - ) - - global channel_id - channel_id = create_notification_channel_response.get_result()['channel_id'] - - assert create_notification_channel_response.get_status_code() == 200 - channel_info = create_notification_channel_response.get_result() - assert channel_info is not None - - @needscredentials - def test_get_notification_channel(self): - - get_notification_channel_response = self.notifications_service.get_notification_channel( - account_id=account_id, - channel_id=channel_id, - ) - - assert get_notification_channel_response.get_status_code() == 200 - channel_get = get_notification_channel_response.get_result() - assert channel_get is not None - - @needscredentials - def test_update_notification_channel(self): - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = { - 'provider_name': 'VA', - 'finding_types': ['image_with_vulnerabilities'] - } - - update_notification_channel_response = self.notifications_service.update_notification_channel( - account_id=account_id, - channel_id=channel_id, - name=f'testString-{identifier}', - type='Webhook', - endpoint='https://us-south.functions.appdomain.cloud/api/v1/web/e6b54af6-ab44-4149-a8e4-e906dcc58136/default/sdk-integration.json', - description=testString, - severity=['low'], - enabled=True, - alert_source=[notification_channel_alert_source_item_model], - ) - - assert update_notification_channel_response.get_status_code() == 200 - channel_info = update_notification_channel_response.get_result() - assert channel_info is not None - - @needscredentials - def test_test_notification_channel(self): - test_notification_channel_response = self.notifications_service.test_notification_channel( - account_id=account_id, - channel_id=channel_id, - ) - - assert test_notification_channel_response.get_status_code() == 200 - test_channel = test_notification_channel_response.get_result() - assert test_channel is not None - - @needscredentials - def test_get_public_key(self): - get_public_key_response = self.notifications_service.get_public_key( - account_id=account_id, - ) - - assert get_public_key_response.get_status_code() == 200 - public_key_get = get_public_key_response.get_result() - assert public_key_get is not None - - @needscredentials - def test_delete_notification_channel(self): - delete_notification_channel_response = self.notifications_service.delete_notification_channel( - account_id=account_id, - channel_id=channel_id, - ) - - assert delete_notification_channel_response.get_status_code() == 200 - channel_delete = delete_notification_channel_response.get_result() - assert channel_delete is not None - - @needscredentials - def test_delete_notification_channels(self): - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = { - 'provider_name': 'VA', - 'finding_types': ['image_with_vulnerabilities'] - } - - create_notification_channel_response = self.notifications_service.create_notification_channel( - account_id=account_id, - name=f'testString-{identifier}', - type='Webhook', - endpoint='https://us-south.functions.appdomain.cloud/api/v1/web/e6b54af6-ab44-4149-a8e4-e906dcc58136/default/sdk-integration.json', - description=testString, - severity=['low'], - enabled=True, - alert_source=[notification_channel_alert_source_item_model], - ) - - channel_id = create_notification_channel_response.get_result()['channel_id'] - - delete_notification_channels_response = self.notifications_service.delete_notification_channels( - account_id=account_id, - request_body=[channel_id], - ) - - assert delete_notification_channels_response.get_status_code() == 200 - channels_delete = delete_notification_channels_response.get_result() - assert channels_delete is not None diff --git a/test/unit/test_notifications_v1.py b/test/unit/test_notifications_v1.py deleted file mode 100644 index dd25331..0000000 --- a/test/unit/test_notifications_v1.py +++ /dev/null @@ -1,1437 +0,0 @@ -# -*- coding: utf-8 -*- -# (C) Copyright IBM Corp. 2021. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -""" -Unit Tests for NotificationsV1 -""" - -from ibm_cloud_sdk_core.authenticators.no_auth_authenticator import NoAuthAuthenticator -import inspect -import json -import pytest -import re -import requests -import responses -import urllib -from ibm_scc.notifications_v1 import * -from ibm_scc.notifications_v1 import TestChannel as TestNotificationChannel - - -_service = NotificationsV1( - authenticator=NoAuthAuthenticator() - ) - -_base_url = 'https://us-south.secadvisor.cloud.ibm.com/notifications' -_service.set_service_url(_base_url) - -############################################################################## -# Start of Service: NotificationChannel -############################################################################## -# region - -class TestListAllChannels(): - """ - Test Class for list_all_channels - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_list_all_channels_all_params(self): - """ - list_all_channels() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"channels": [{"channel_id": "channel_id", "name": "name", "description": "description", "type": "Webhook", "severity": {"critical": true, "high": true, "medium": true, "low": false}, "endpoint": "endpoint", "enabled": false, "alert_source": [{"provider_name": "VA", "finding_types": [{"anyKey": "anyValue"}]}], "frequency": "frequency"}]}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - transaction_id = 'testString' - limit = 38 - skip = 38 - - # Invoke method - response = _service.list_all_channels( - account_id, - transaction_id=transaction_id, - limit=limit, - skip=skip, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # Validate query params - query_string = responses.calls[0].request.url.split('?',1)[1] - query_string = urllib.parse.unquote_plus(query_string) - assert 'limit={}'.format(limit) in query_string - assert 'skip={}'.format(skip) in query_string - - - @responses.activate - def test_list_all_channels_required_params(self): - """ - test_list_all_channels_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"channels": [{"channel_id": "channel_id", "name": "name", "description": "description", "type": "Webhook", "severity": {"critical": true, "high": true, "medium": true, "low": false}, "endpoint": "endpoint", "enabled": false, "alert_source": [{"provider_name": "VA", "finding_types": [{"anyKey": "anyValue"}]}], "frequency": "frequency"}]}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - - # Invoke method - response = _service.list_all_channels( - account_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_list_all_channels_value_error(self): - """ - test_list_all_channels_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"channels": [{"channel_id": "channel_id", "name": "name", "description": "description", "type": "Webhook", "severity": {"critical": true, "high": true, "medium": true, "low": false}, "endpoint": "endpoint", "enabled": false, "alert_source": [{"provider_name": "VA", "finding_types": [{"anyKey": "anyValue"}]}], "frequency": "frequency"}]}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.list_all_channels(**req_copy) - - - -class TestCreateNotificationChannel(): - """ - Test Class for create_notification_channel - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_create_notification_channel_all_params(self): - """ - create_notification_channel() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"channel_id": "channel_id", "status_code": 11}' - responses.add(responses.POST, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = {} - notification_channel_alert_source_item_model['provider_name'] = 'testString' - notification_channel_alert_source_item_model['finding_types'] = ['testString'] - - # Set up parameter values - account_id = 'testString' - name = 'testString' - type = 'Webhook' - endpoint = 'testString' - description = 'testString' - severity = ['low'] - enabled = True - alert_source = [notification_channel_alert_source_item_model] - transaction_id = 'testString' - - # Invoke method - response = _service.create_notification_channel( - account_id, - name, - type, - endpoint, - description=description, - severity=severity, - enabled=enabled, - alert_source=alert_source, - transaction_id=transaction_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # Validate body params - req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) - assert req_body['name'] == 'testString' - assert req_body['type'] == 'Webhook' - assert req_body['endpoint'] == 'testString' - assert req_body['description'] == 'testString' - assert req_body['severity'] == ['low'] - assert req_body['enabled'] == True - assert req_body['alert_source'] == [notification_channel_alert_source_item_model] - - - @responses.activate - def test_create_notification_channel_required_params(self): - """ - test_create_notification_channel_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"channel_id": "channel_id", "status_code": 11}' - responses.add(responses.POST, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = {} - notification_channel_alert_source_item_model['provider_name'] = 'testString' - notification_channel_alert_source_item_model['finding_types'] = ['testString'] - - # Set up parameter values - account_id = 'testString' - name = 'testString' - type = 'Webhook' - endpoint = 'testString' - description = 'testString' - severity = ['low'] - enabled = True - alert_source = [notification_channel_alert_source_item_model] - - # Invoke method - response = _service.create_notification_channel( - account_id, - name, - type, - endpoint, - description=description, - severity=severity, - enabled=enabled, - alert_source=alert_source, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # Validate body params - req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) - assert req_body['name'] == 'testString' - assert req_body['type'] == 'Webhook' - assert req_body['endpoint'] == 'testString' - assert req_body['description'] == 'testString' - assert req_body['severity'] == ['low'] - assert req_body['enabled'] == True - assert req_body['alert_source'] == [notification_channel_alert_source_item_model] - - - @responses.activate - def test_create_notification_channel_value_error(self): - """ - test_create_notification_channel_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"channel_id": "channel_id", "status_code": 11}' - responses.add(responses.POST, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = {} - notification_channel_alert_source_item_model['provider_name'] = 'testString' - notification_channel_alert_source_item_model['finding_types'] = ['testString'] - - # Set up parameter values - account_id = 'testString' - name = 'testString' - type = 'Webhook' - endpoint = 'testString' - description = 'testString' - severity = ['low'] - enabled = True - alert_source = [notification_channel_alert_source_item_model] - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - "name": name, - "type": type, - "endpoint": endpoint, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.create_notification_channel(**req_copy) - - - -class TestDeleteNotificationChannels(): - """ - Test Class for delete_notification_channels - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_delete_notification_channels_all_params(self): - """ - delete_notification_channels() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"message": "message"}' - responses.add(responses.DELETE, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - request_body = ['testString'] - transaction_id = 'testString' - - # Invoke method - response = _service.delete_notification_channels( - account_id, - request_body, - transaction_id=transaction_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # Validate body params - req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) - assert req_body == request_body - - - @responses.activate - def test_delete_notification_channels_required_params(self): - """ - test_delete_notification_channels_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"message": "message"}' - responses.add(responses.DELETE, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - request_body = ['testString'] - - # Invoke method - response = _service.delete_notification_channels( - account_id, - request_body, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # Validate body params - req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) - assert req_body == request_body - - - @responses.activate - def test_delete_notification_channels_value_error(self): - """ - test_delete_notification_channels_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels') - mock_response = '{"message": "message"}' - responses.add(responses.DELETE, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - request_body = ['testString'] - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - "request_body": request_body, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.delete_notification_channels(**req_copy) - - - -class TestDeleteNotificationChannel(): - """ - Test Class for delete_notification_channel - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_delete_notification_channel_all_params(self): - """ - delete_notification_channel() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel_id": "channel_id", "message": "message"}' - responses.add(responses.DELETE, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - transaction_id = 'testString' - - # Invoke method - response = _service.delete_notification_channel( - account_id, - channel_id, - transaction_id=transaction_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_delete_notification_channel_required_params(self): - """ - test_delete_notification_channel_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel_id": "channel_id", "message": "message"}' - responses.add(responses.DELETE, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - - # Invoke method - response = _service.delete_notification_channel( - account_id, - channel_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_delete_notification_channel_value_error(self): - """ - test_delete_notification_channel_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel_id": "channel_id", "message": "message"}' - responses.add(responses.DELETE, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - "channel_id": channel_id, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.delete_notification_channel(**req_copy) - - - -class TestGetNotificationChannel(): - """ - Test Class for get_notification_channel - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_get_notification_channel_all_params(self): - """ - get_notification_channel() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel": {"channel_id": "channel_id", "name": "name", "description": "description", "type": "Webhook", "severity": {"critical": true, "high": true, "medium": true, "low": false}, "endpoint": "endpoint", "enabled": false, "alert_source": [{"provider_name": "VA", "finding_types": [{"anyKey": "anyValue"}]}], "frequency": "frequency"}}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - transaction_id = 'testString' - - # Invoke method - response = _service.get_notification_channel( - account_id, - channel_id, - transaction_id=transaction_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_get_notification_channel_required_params(self): - """ - test_get_notification_channel_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel": {"channel_id": "channel_id", "name": "name", "description": "description", "type": "Webhook", "severity": {"critical": true, "high": true, "medium": true, "low": false}, "endpoint": "endpoint", "enabled": false, "alert_source": [{"provider_name": "VA", "finding_types": [{"anyKey": "anyValue"}]}], "frequency": "frequency"}}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - - # Invoke method - response = _service.get_notification_channel( - account_id, - channel_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_get_notification_channel_value_error(self): - """ - test_get_notification_channel_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel": {"channel_id": "channel_id", "name": "name", "description": "description", "type": "Webhook", "severity": {"critical": true, "high": true, "medium": true, "low": false}, "endpoint": "endpoint", "enabled": false, "alert_source": [{"provider_name": "VA", "finding_types": [{"anyKey": "anyValue"}]}], "frequency": "frequency"}}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - "channel_id": channel_id, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.get_notification_channel(**req_copy) - - - -class TestUpdateNotificationChannel(): - """ - Test Class for update_notification_channel - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_update_notification_channel_all_params(self): - """ - update_notification_channel() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel_id": "channel_id", "status_code": 11}' - responses.add(responses.PUT, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = {} - notification_channel_alert_source_item_model['provider_name'] = 'testString' - notification_channel_alert_source_item_model['finding_types'] = ['testString'] - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - name = 'testString' - type = 'Webhook' - endpoint = 'testString' - description = 'testString' - severity = ['low'] - enabled = True - alert_source = [notification_channel_alert_source_item_model] - transaction_id = 'testString' - - # Invoke method - response = _service.update_notification_channel( - account_id, - channel_id, - name, - type, - endpoint, - description=description, - severity=severity, - enabled=enabled, - alert_source=alert_source, - transaction_id=transaction_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # Validate body params - req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) - assert req_body['name'] == 'testString' - assert req_body['type'] == 'Webhook' - assert req_body['endpoint'] == 'testString' - assert req_body['description'] == 'testString' - assert req_body['severity'] == ['low'] - assert req_body['enabled'] == True - assert req_body['alert_source'] == [notification_channel_alert_source_item_model] - - - @responses.activate - def test_update_notification_channel_required_params(self): - """ - test_update_notification_channel_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel_id": "channel_id", "status_code": 11}' - responses.add(responses.PUT, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = {} - notification_channel_alert_source_item_model['provider_name'] = 'testString' - notification_channel_alert_source_item_model['finding_types'] = ['testString'] - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - name = 'testString' - type = 'Webhook' - endpoint = 'testString' - description = 'testString' - severity = ['low'] - enabled = True - alert_source = [notification_channel_alert_source_item_model] - - # Invoke method - response = _service.update_notification_channel( - account_id, - channel_id, - name, - type, - endpoint, - description=description, - severity=severity, - enabled=enabled, - alert_source=alert_source, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # Validate body params - req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) - assert req_body['name'] == 'testString' - assert req_body['type'] == 'Webhook' - assert req_body['endpoint'] == 'testString' - assert req_body['description'] == 'testString' - assert req_body['severity'] == ['low'] - assert req_body['enabled'] == True - assert req_body['alert_source'] == [notification_channel_alert_source_item_model] - - - @responses.activate - def test_update_notification_channel_value_error(self): - """ - test_update_notification_channel_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString') - mock_response = '{"channel_id": "channel_id", "status_code": 11}' - responses.add(responses.PUT, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model = {} - notification_channel_alert_source_item_model['provider_name'] = 'testString' - notification_channel_alert_source_item_model['finding_types'] = ['testString'] - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - name = 'testString' - type = 'Webhook' - endpoint = 'testString' - description = 'testString' - severity = ['low'] - enabled = True - alert_source = [notification_channel_alert_source_item_model] - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - "channel_id": channel_id, - "name": name, - "type": type, - "endpoint": endpoint, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.update_notification_channel(**req_copy) - - - -class TestTestNotificationChannel(): - """ - Test Class for test_notification_channel - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_test_notification_channel_all_params(self): - """ - test_notification_channel() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString/test') - mock_response = '{"test": "test"}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - transaction_id = 'testString' - - # Invoke method - response = _service.test_notification_channel( - account_id, - channel_id, - transaction_id=transaction_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_test_notification_channel_required_params(self): - """ - test_test_notification_channel_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString/test') - mock_response = '{"test": "test"}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - - # Invoke method - response = _service.test_notification_channel( - account_id, - channel_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_test_notification_channel_value_error(self): - """ - test_test_notification_channel_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/channels/testString/test') - mock_response = '{"test": "test"}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - channel_id = 'testString' - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - "channel_id": channel_id, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.test_notification_channel(**req_copy) - - - -class TestGetPublicKey(): - """ - Test Class for get_public_key - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_get_public_key_all_params(self): - """ - get_public_key() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/public_key') - mock_response = '{"public_key": "public_key"}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - transaction_id = 'testString' - - # Invoke method - response = _service.get_public_key( - account_id, - transaction_id=transaction_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_get_public_key_required_params(self): - """ - test_get_public_key_required_params() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/public_key') - mock_response = '{"public_key": "public_key"}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - - # Invoke method - response = _service.get_public_key( - account_id, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - - - @responses.activate - def test_get_public_key_value_error(self): - """ - test_get_public_key_value_error() - """ - # Set up mock - url = self.preprocess_url(_base_url + '/v1/testString/notifications/public_key') - mock_response = '{"public_key": "public_key"}' - responses.add(responses.GET, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Set up parameter values - account_id = 'testString' - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "account_id": account_id, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - _service.get_public_key(**req_copy) - - - -# endregion -############################################################################## -# End of Service: NotificationChannel -############################################################################## - - -############################################################################## -# Start of Model Tests -############################################################################## -# region -class TestChannelAlertSourceItem(): - """ - Test Class for ChannelAlertSourceItem - """ - - def test_channel_alert_source_item_serialization(self): - """ - Test serialization/deserialization for ChannelAlertSourceItem - """ - - # Construct a json representation of a ChannelAlertSourceItem model - channel_alert_source_item_model_json = {} - channel_alert_source_item_model_json['provider_name'] = 'VA' - channel_alert_source_item_model_json['finding_types'] = [{ 'foo': 'bar' }] - - # Construct a model instance of ChannelAlertSourceItem by calling from_dict on the json representation - channel_alert_source_item_model = ChannelAlertSourceItem.from_dict(channel_alert_source_item_model_json) - assert channel_alert_source_item_model != False - - # Construct a model instance of ChannelAlertSourceItem by calling from_dict on the json representation - channel_alert_source_item_model_dict = ChannelAlertSourceItem.from_dict(channel_alert_source_item_model_json).__dict__ - channel_alert_source_item_model2 = ChannelAlertSourceItem(**channel_alert_source_item_model_dict) - - # Verify the model instances are equivalent - assert channel_alert_source_item_model == channel_alert_source_item_model2 - - # Convert model instance back to dict and verify no loss of data - channel_alert_source_item_model_json2 = channel_alert_source_item_model.to_dict() - assert channel_alert_source_item_model_json2 == channel_alert_source_item_model_json - -class TestChannelDelete(): - """ - Test Class for ChannelDelete - """ - - def test_channel_delete_serialization(self): - """ - Test serialization/deserialization for ChannelDelete - """ - - # Construct a json representation of a ChannelDelete model - channel_delete_model_json = {} - channel_delete_model_json['channel_id'] = 'testString' - channel_delete_model_json['message'] = 'testString' - - # Construct a model instance of ChannelDelete by calling from_dict on the json representation - channel_delete_model = ChannelDelete.from_dict(channel_delete_model_json) - assert channel_delete_model != False - - # Construct a model instance of ChannelDelete by calling from_dict on the json representation - channel_delete_model_dict = ChannelDelete.from_dict(channel_delete_model_json).__dict__ - channel_delete_model2 = ChannelDelete(**channel_delete_model_dict) - - # Verify the model instances are equivalent - assert channel_delete_model == channel_delete_model2 - - # Convert model instance back to dict and verify no loss of data - channel_delete_model_json2 = channel_delete_model.to_dict() - assert channel_delete_model_json2 == channel_delete_model_json - -class TestChannelGet(): - """ - Test Class for ChannelGet - """ - - def test_channel_get_serialization(self): - """ - Test serialization/deserialization for ChannelGet - """ - - # Construct dict forms of any model objects needed in order to build this model. - - channel_severity_model = {} # ChannelSeverity - channel_severity_model['critical'] = True - channel_severity_model['high'] = True - channel_severity_model['medium'] = True - channel_severity_model['low'] = True - - channel_alert_source_item_model = {} # ChannelAlertSourceItem - channel_alert_source_item_model['provider_name'] = 'VA' - channel_alert_source_item_model['finding_types'] = [{ 'foo': 'bar' }] - - channel_model = {} # Channel - channel_model['channel_id'] = 'testString' - channel_model['name'] = 'testString' - channel_model['description'] = 'testString' - channel_model['type'] = 'Webhook' - channel_model['severity'] = channel_severity_model - channel_model['endpoint'] = 'testString' - channel_model['enabled'] = True - channel_model['alert_source'] = [channel_alert_source_item_model] - channel_model['frequency'] = 'testString' - - # Construct a json representation of a ChannelGet model - channel_get_model_json = {} - channel_get_model_json['channel'] = channel_model - - # Construct a model instance of ChannelGet by calling from_dict on the json representation - channel_get_model = ChannelGet.from_dict(channel_get_model_json) - assert channel_get_model != False - - # Construct a model instance of ChannelGet by calling from_dict on the json representation - channel_get_model_dict = ChannelGet.from_dict(channel_get_model_json).__dict__ - channel_get_model2 = ChannelGet(**channel_get_model_dict) - - # Verify the model instances are equivalent - assert channel_get_model == channel_get_model2 - - # Convert model instance back to dict and verify no loss of data - channel_get_model_json2 = channel_get_model.to_dict() - assert channel_get_model_json2 == channel_get_model_json - -class TestChannelInfo(): - """ - Test Class for ChannelInfo - """ - - def test_channel_info_serialization(self): - """ - Test serialization/deserialization for ChannelInfo - """ - - # Construct a json representation of a ChannelInfo model - channel_info_model_json = {} - channel_info_model_json['channel_id'] = 'testString' - channel_info_model_json['status_code'] = 38 - - # Construct a model instance of ChannelInfo by calling from_dict on the json representation - channel_info_model = ChannelInfo.from_dict(channel_info_model_json) - assert channel_info_model != False - - # Construct a model instance of ChannelInfo by calling from_dict on the json representation - channel_info_model_dict = ChannelInfo.from_dict(channel_info_model_json).__dict__ - channel_info_model2 = ChannelInfo(**channel_info_model_dict) - - # Verify the model instances are equivalent - assert channel_info_model == channel_info_model2 - - # Convert model instance back to dict and verify no loss of data - channel_info_model_json2 = channel_info_model.to_dict() - assert channel_info_model_json2 == channel_info_model_json - -class TestChannelSeverity(): - """ - Test Class for ChannelSeverity - """ - - def test_channel_severity_serialization(self): - """ - Test serialization/deserialization for ChannelSeverity - """ - - # Construct a json representation of a ChannelSeverity model - channel_severity_model_json = {} - channel_severity_model_json['critical'] = True - channel_severity_model_json['high'] = True - channel_severity_model_json['medium'] = True - channel_severity_model_json['low'] = True - - # Construct a model instance of ChannelSeverity by calling from_dict on the json representation - channel_severity_model = ChannelSeverity.from_dict(channel_severity_model_json) - assert channel_severity_model != False - - # Construct a model instance of ChannelSeverity by calling from_dict on the json representation - channel_severity_model_dict = ChannelSeverity.from_dict(channel_severity_model_json).__dict__ - channel_severity_model2 = ChannelSeverity(**channel_severity_model_dict) - - # Verify the model instances are equivalent - assert channel_severity_model == channel_severity_model2 - - # Convert model instance back to dict and verify no loss of data - channel_severity_model_json2 = channel_severity_model.to_dict() - assert channel_severity_model_json2 == channel_severity_model_json - -class TestChannelsDelete(): - """ - Test Class for ChannelsDelete - """ - - def test_channels_delete_serialization(self): - """ - Test serialization/deserialization for ChannelsDelete - """ - - # Construct a json representation of a ChannelsDelete model - channels_delete_model_json = {} - channels_delete_model_json['message'] = 'testString' - - # Construct a model instance of ChannelsDelete by calling from_dict on the json representation - channels_delete_model = ChannelsDelete.from_dict(channels_delete_model_json) - assert channels_delete_model != False - - # Construct a model instance of ChannelsDelete by calling from_dict on the json representation - channels_delete_model_dict = ChannelsDelete.from_dict(channels_delete_model_json).__dict__ - channels_delete_model2 = ChannelsDelete(**channels_delete_model_dict) - - # Verify the model instances are equivalent - assert channels_delete_model == channels_delete_model2 - - # Convert model instance back to dict and verify no loss of data - channels_delete_model_json2 = channels_delete_model.to_dict() - assert channels_delete_model_json2 == channels_delete_model_json - -class TestChannelsList(): - """ - Test Class for ChannelsList - """ - - def test_channels_list_serialization(self): - """ - Test serialization/deserialization for ChannelsList - """ - - # Construct dict forms of any model objects needed in order to build this model. - - channel_severity_model = {} # ChannelSeverity - channel_severity_model['critical'] = True - channel_severity_model['high'] = True - channel_severity_model['medium'] = True - channel_severity_model['low'] = True - - channel_alert_source_item_model = {} # ChannelAlertSourceItem - channel_alert_source_item_model['provider_name'] = 'VA' - channel_alert_source_item_model['finding_types'] = [{ 'foo': 'bar' }] - - channel_model = {} # Channel - channel_model['channel_id'] = 'testString' - channel_model['name'] = 'testString' - channel_model['description'] = 'testString' - channel_model['type'] = 'Webhook' - channel_model['severity'] = channel_severity_model - channel_model['endpoint'] = 'testString' - channel_model['enabled'] = True - channel_model['alert_source'] = [channel_alert_source_item_model] - channel_model['frequency'] = 'testString' - - # Construct a json representation of a ChannelsList model - channels_list_model_json = {} - channels_list_model_json['channels'] = [channel_model] - - # Construct a model instance of ChannelsList by calling from_dict on the json representation - channels_list_model = ChannelsList.from_dict(channels_list_model_json) - assert channels_list_model != False - - # Construct a model instance of ChannelsList by calling from_dict on the json representation - channels_list_model_dict = ChannelsList.from_dict(channels_list_model_json).__dict__ - channels_list_model2 = ChannelsList(**channels_list_model_dict) - - # Verify the model instances are equivalent - assert channels_list_model == channels_list_model2 - - # Convert model instance back to dict and verify no loss of data - channels_list_model_json2 = channels_list_model.to_dict() - assert channels_list_model_json2 == channels_list_model_json - -class TestNotificationChannelAlertSourceItem(): - """ - Test Class for NotificationChannelAlertSourceItem - """ - - def test_notification_channel_alert_source_item_serialization(self): - """ - Test serialization/deserialization for NotificationChannelAlertSourceItem - """ - - # Construct a json representation of a NotificationChannelAlertSourceItem model - notification_channel_alert_source_item_model_json = {} - notification_channel_alert_source_item_model_json['provider_name'] = 'testString' - notification_channel_alert_source_item_model_json['finding_types'] = ['testString'] - - # Construct a model instance of NotificationChannelAlertSourceItem by calling from_dict on the json representation - notification_channel_alert_source_item_model = NotificationChannelAlertSourceItem.from_dict(notification_channel_alert_source_item_model_json) - assert notification_channel_alert_source_item_model != False - - # Construct a model instance of NotificationChannelAlertSourceItem by calling from_dict on the json representation - notification_channel_alert_source_item_model_dict = NotificationChannelAlertSourceItem.from_dict(notification_channel_alert_source_item_model_json).__dict__ - notification_channel_alert_source_item_model2 = NotificationChannelAlertSourceItem(**notification_channel_alert_source_item_model_dict) - - # Verify the model instances are equivalent - assert notification_channel_alert_source_item_model == notification_channel_alert_source_item_model2 - - # Convert model instance back to dict and verify no loss of data - notification_channel_alert_source_item_model_json2 = notification_channel_alert_source_item_model.to_dict() - assert notification_channel_alert_source_item_model_json2 == notification_channel_alert_source_item_model_json - -class TestPublicKeyGet(): - """ - Test Class for PublicKeyGet - """ - - def test_public_key_get_serialization(self): - """ - Test serialization/deserialization for PublicKeyGet - """ - - # Construct a json representation of a PublicKeyGet model - public_key_get_model_json = {} - public_key_get_model_json['public_key'] = 'testString' - - # Construct a model instance of PublicKeyGet by calling from_dict on the json representation - public_key_get_model = PublicKeyGet.from_dict(public_key_get_model_json) - assert public_key_get_model != False - - # Construct a model instance of PublicKeyGet by calling from_dict on the json representation - public_key_get_model_dict = PublicKeyGet.from_dict(public_key_get_model_json).__dict__ - public_key_get_model2 = PublicKeyGet(**public_key_get_model_dict) - - # Verify the model instances are equivalent - assert public_key_get_model == public_key_get_model2 - - # Convert model instance back to dict and verify no loss of data - public_key_get_model_json2 = public_key_get_model.to_dict() - assert public_key_get_model_json2 == public_key_get_model_json - -class TestTestChannel(): - """ - Test Class for TestChannel - """ - - def test_test_channel_serialization(self): - """ - Test serialization/deserialization for TestChannel - """ - - # Construct a json representation of a TestChannel model - test_channel_model_json = {} - test_channel_model_json['test'] = 'testString' - - # Construct a model instance of TestChannel by calling from_dict on the json representation - test_channel_model = TestNotificationChannel.from_dict(test_channel_model_json) - assert test_channel_model != False - - # Construct a model instance of TestChannel by calling from_dict on the json representation - test_channel_model_dict = TestNotificationChannel.from_dict(test_channel_model_json).__dict__ - test_channel_model2 = TestNotificationChannel(**test_channel_model_dict) - - # Verify the model instances are equivalent - assert test_channel_model == test_channel_model2 - - # Convert model instance back to dict and verify no loss of data - test_channel_model_json2 = test_channel_model.to_dict() - assert test_channel_model_json2 == test_channel_model_json - -class TestChannel(): - """ - Test Class for Channel - """ - - def test_channel_serialization(self): - """ - Test serialization/deserialization for Channel - """ - - # Construct dict forms of any model objects needed in order to build this model. - - channel_severity_model = {} # ChannelSeverity - channel_severity_model['critical'] = True - channel_severity_model['high'] = True - channel_severity_model['medium'] = True - channel_severity_model['low'] = True - - channel_alert_source_item_model = {} # ChannelAlertSourceItem - channel_alert_source_item_model['provider_name'] = 'VA' - channel_alert_source_item_model['finding_types'] = [{ 'foo': 'bar' }] - - # Construct a json representation of a Channel model - channel_model_json = {} - channel_model_json['channel_id'] = 'testString' - channel_model_json['name'] = 'testString' - channel_model_json['description'] = 'testString' - channel_model_json['type'] = 'Webhook' - channel_model_json['severity'] = channel_severity_model - channel_model_json['endpoint'] = 'testString' - channel_model_json['enabled'] = True - channel_model_json['alert_source'] = [channel_alert_source_item_model] - channel_model_json['frequency'] = 'testString' - - # Construct a model instance of Channel by calling from_dict on the json representation - channel_model = Channel.from_dict(channel_model_json) - assert channel_model != False - - # Construct a model instance of Channel by calling from_dict on the json representation - channel_model_dict = Channel.from_dict(channel_model_json).__dict__ - channel_model2 = Channel(**channel_model_dict) - - # Verify the model instances are equivalent - assert channel_model == channel_model2 - - # Convert model instance back to dict and verify no loss of data - channel_model_json2 = channel_model.to_dict() - assert channel_model_json2 == channel_model_json - - -# endregion -############################################################################## -# End of Model Tests -##############################################################################