# coding: utf-8 """ BGP Updates API Retrieve BGP updates observed by BGP monitors tracking prefixes in your ThousandEyes account group. Generated by OpenAPI Generator (https://openapi-generator.tech) Do not edit the class manually. """ # noqa: E501 import json import unittest import thousandeyes_sdk.bgp_updates.models from thousandeyes_sdk.core.exceptions import ApiException from thousandeyes_sdk.bgp_updates.api.bgp_updates_api import BGPUpdatesApi from .integration_test_utils import IntegrationTestBase from .test_utils import assert_constructed_model_matches_example_json class TestBGPUpdatesApiIntegration(IntegrationTestBase): """BGPUpdatesApi integration test stubs""" def setUp(self) -> None: super().setUp() self.api = BGPUpdatesApi(self.api_client) def test_get_bgp_updates_happy_path(self) -> None: """Integration test for get_bgp_updates success path""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] response_body_json = """ { "_links" : { "next" : { "hreflang" : "hreflang", "templated" : true, "profile" : "profile", "name" : "name", "href" : "https://api.thousandeyes.com/v7/link/to/resource/id", "type" : "type", "deprecation" : "deprecation", "title" : "title" }, "previous" : { "hreflang" : "hreflang", "templated" : true, "profile" : "profile", "name" : "name", "href" : "https://api.thousandeyes.com/v7/link/to/resource/id", "type" : "type", "deprecation" : "deprecation", "title" : "title" }, "self" : { "hreflang" : "hreflang", "templated" : true, "profile" : "profile", "name" : "name", "href" : "https://api.thousandeyes.com/v7/link/to/resource/id", "type" : "type", "deprecation" : "deprecation", "title" : "title" } }, "updates" : [ { "monitorId" : "101", "prefix" : "192.0.2.0/24", "asPath" : [ 64512, 64513 ], "originAs" : 64512, "monitor" : { "monitorId" : "101", "ipAddress" : "10.0.0.1", "asn" : 64512, "monitorType" : "private", "monitorName" : "Private monitor", "countryId" : "US" }, "rpkiStatus" : "Valid", "communities" : [ "64512:100" ], "updateType" : "announcement", "timestamp" : "2024-05-31T16:08:37Z" }, { "monitorId" : "101", "prefix" : "192.0.2.0/24", "asPath" : [ 64512, 64513 ], "originAs" : 64512, "monitor" : { "monitorId" : "101", "ipAddress" : "10.0.0.1", "asn" : 64512, "monitorType" : "private", "monitorName" : "Private monitor", "countryId" : "US" }, "rpkiStatus" : "Valid", "communities" : [ "64512:100" ], "updateType" : "announcement", "timestamp" : "2024-05-31T16:08:37Z" } ] } """ expected_response = json.loads(response_body_json) response = self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates"), ) assert_constructed_model_matches_example_json(response, expected_response) def test_get_bgp_updates_error_400(self) -> None: """Integration test for get_bgp_updates error path (HTTP 400)""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] error_body_json = """ { "instance" : "instance", "detail" : "detail", "type" : "type", "title" : "title", "errors" : [ { "code" : "code", "field" : "field", "message" : "message" }, { "code" : "code", "field" : "field", "message" : "message" } ], "status" : 0 } """ expected_error = json.loads(error_body_json) with self.assertRaises( ApiException.exception_class_for_http_status(400) ) as context: self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates", error_status="400"), ) assert_constructed_model_matches_example_json(context.exception.data, expected_error) def test_get_bgp_updates_error_401(self) -> None: """Integration test for get_bgp_updates error path (HTTP 401)""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] error_body_json = """ { "error_description" : "Invalid access token", "error" : "invalid_token" } """ expected_error = json.loads(error_body_json) with self.assertRaises( ApiException.exception_class_for_http_status(401) ) as context: self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates", error_status="401"), ) assert_constructed_model_matches_example_json(context.exception.data, expected_error) def test_get_bgp_updates_error_403(self) -> None: """Integration test for get_bgp_updates error path (HTTP 403)""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] error_body_json = """ { "instance" : "instance", "detail" : "detail", "type" : "type", "title" : "title", "status" : 6 } """ expected_error = json.loads(error_body_json) with self.assertRaises( ApiException.exception_class_for_http_status(403) ) as context: self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates", error_status="403"), ) assert_constructed_model_matches_example_json(context.exception.data, expected_error) def test_get_bgp_updates_error_404(self) -> None: """Integration test for get_bgp_updates error path (HTTP 404)""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] error_body_json = """ { "instance" : "instance", "detail" : "detail", "type" : "type", "title" : "title", "status" : 6 } """ expected_error = json.loads(error_body_json) with self.assertRaises( ApiException.exception_class_for_http_status(404) ) as context: self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates", error_status="404"), ) assert_constructed_model_matches_example_json(context.exception.data, expected_error) def test_get_bgp_updates_error_429(self) -> None: """Integration test for get_bgp_updates error path (HTTP 429)""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] error_body_json = """ { "instance" : "instance", "detail" : "detail", "type" : "type", "title" : "title", "status" : 6 } """ expected_error = json.loads(error_body_json) with self.assertRaises( ApiException.exception_class_for_http_status(429) ) as context: self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates", error_status="429"), ) assert_constructed_model_matches_example_json(context.exception.data, expected_error) def test_get_bgp_updates_error_500(self) -> None: """Integration test for get_bgp_updates error path (HTTP 500)""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] error_body_json = """ { "instance" : "instance", "detail" : "detail", "type" : "type", "title" : "title", "status" : 6 } """ expected_error = json.loads(error_body_json) with self.assertRaises( ApiException.exception_class_for_http_status(500) ) as context: self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates", error_status="500"), ) assert_constructed_model_matches_example_json(context.exception.data, expected_error) def test_get_bgp_updates_error_502(self) -> None: """Integration test for get_bgp_updates error path (HTTP 502)""" aid = '1234' max = 20 cursor = 'cursor_example' expand = ["monitor"] start_date = '2022-07-17T22:00:54Z' end_date = '2022-07-18T22:00:54Z' prefixes = ["192.0.2.0/24"] origin_ases = [64512] as_paths = ["64512 64513"] rpki_statuses = [ 'Valid' ] update_type = [ 'announcement' ] monitor_ids = ["101"] communities = ["64512:100"] error_body_json = """ { "instance" : "instance", "detail" : "detail", "type" : "type", "title" : "title", "status" : 6 } """ expected_error = json.loads(error_body_json) with self.assertRaises( ApiException.exception_class_for_http_status(502) ) as context: self.api.get_bgp_updates( aid=aid, max=max, cursor=cursor, expand=expand, start_date=start_date, end_date=end_date, prefixes=prefixes, origin_ases=origin_ases, as_paths=as_paths, rpki_statuses=rpki_statuses, update_type=update_type, monitor_ids=monitor_ids, communities=communities, _headers=self.te_headers("get_bgp_updates", error_status="502"), ) assert_constructed_model_matches_example_json(context.exception.data, expected_error) if __name__ == '__main__': unittest.main()