From 31791b05ab8b0cc1e154e73bbd74467b0759f96c Mon Sep 17 00:00:00 2001 From: mduda Date: Thu, 3 Sep 2026 10:38:27 +0200 Subject: [PATCH] feat: [DPS-45125] ACLP Logs - add new TrafficPeak destination type --- .github/workflows/e2e-test-pr.yml | 13 +- .github/workflows/e2e-test.yml | 11 +- linode_api4/groups/monitor.py | 31 ++- linode_api4/objects/monitor.py | 30 ++- test/fixtures/monitor_streams_4.json | 35 +++ .../monitor_streams_destinations_3.json | 28 +++ .../models/monitor/test_monitor_logs.py | 200 ++++++++++++++---- test/unit/objects/monitor_test.py | 121 +++++++++++ 8 files changed, 423 insertions(+), 46 deletions(-) create mode 100644 test/fixtures/monitor_streams_4.json create mode 100644 test/fixtures/monitor_streams_destinations_3.json diff --git a/.github/workflows/e2e-test-pr.yml b/.github/workflows/e2e-test-pr.yml index d89cda3b1..8d3cc6127 100644 --- a/.github/workflows/e2e-test-pr.yml +++ b/.github/workflows/e2e-test-pr.yml @@ -10,6 +10,14 @@ on: options: - 'true' - 'false' + run_traffic_peak_tests: + description: 'Set this parameter to "true" to run ACLP logs TrafficPeak destination related test cases' + required: false + default: 'false' + type: choice + options: + - 'true' + - 'false' run_db_fork_tests: description: 'Set this parameter to "true" to run fork database related test cases' required: false @@ -112,12 +120,13 @@ jobs: run: | timestamp=$(date +'%Y%m%d%H%M') report_filename="${timestamp}_sdk_test_report.xml" - make test-int RUN_DB_FORK_TESTS="$RUN_DB_FORK_TESTS" RUN_DB_TESTS="$RUN_DB_TESTS" RUN_ACLP_LOGS_STREAM_TESTS="$RUN_ACLP_LOGS_STREAM_TESTS" TEST_ARGS="--junitxml=${report_filename}" TEST_SUITE="$TEST_SUITE" + make test-int RUN_DB_FORK_TESTS="$RUN_DB_FORK_TESTS" RUN_DB_TESTS="$RUN_DB_TESTS" RUN_ACLP_LOGS_STREAM_TESTS="$RUN_ACLP_LOGS_STREAM_TESTS" RUN_TRAFFIC_PEAK_TESTS="$RUN_TRAFFIC_PEAK_TESTS" TEST_ARGS="--junitxml=${report_filename}" TEST_SUITE="$TEST_SUITE" env: LINODE_TOKEN: ${{ secrets.LINODE_TOKEN }} RUN_DB_FORK_TESTS: ${{ github.event.inputs.run_db_fork_tests }} RUN_DB_TESTS: ${{ github.event.inputs.run_db_tests }} RUN_ACLP_LOGS_STREAM_TESTS: ${{ github.event.inputs.run_aclp_logs_stream_tests }} + RUN_TRAFFIC_PEAK_TESTS: ${{ github.event.inputs.run_traffic_peak_tests }} TEST_SUITE: ${{ github.event.inputs.test_suite }} - name: Upload test results @@ -218,4 +227,4 @@ jobs: fi done env: - LINODE_CLI_TOKEN: ${{ secrets.LINODE_TOKEN }} \ No newline at end of file + LINODE_CLI_TOKEN: ${{ secrets.LINODE_TOKEN }} diff --git a/.github/workflows/e2e-test.yml b/.github/workflows/e2e-test.yml index 46742cbae..5bfa572f9 100644 --- a/.github/workflows/e2e-test.yml +++ b/.github/workflows/e2e-test.yml @@ -11,6 +11,14 @@ on: options: - 'true' - 'false' + run_traffic_peak_tests: + description: 'Set this parameter to "true" to run TrafficPeak related test cases' + required: false + default: 'false' + type: choice + options: + - 'true' + - 'false' run_db_fork_tests: description: 'Set this parameter to "true" to run fork database related test cases' required: false @@ -107,12 +115,13 @@ jobs: run: | timestamp=$(date +'%Y%m%d%H%M') report_filename="${timestamp}_sdk_test_report.xml" - make test-int RUN_DB_FORK_TESTS="$RUN_DB_FORK_TESTS" RUN_DB_TESTS="$RUN_DB_TESTS" RUN_ACLP_LOGS_STREAM_TESTS="$RUN_ACLP_LOGS_STREAM_TESTS" TEST_SUITE="$TEST_SUITE" TEST_ARGS="--junitxml=${report_filename}" + make test-int RUN_DB_FORK_TESTS="$RUN_DB_FORK_TESTS" RUN_DB_TESTS="$RUN_DB_TESTS" RUN_ACLP_LOGS_STREAM_TESTS="$RUN_ACLP_LOGS_STREAM_TESTS" RUN_TRAFFIC_PEAK_TESTS="$RUN_TRAFFIC_PEAK_TESTS" TEST_SUITE="$TEST_SUITE" TEST_ARGS="--junitxml=${report_filename}" env: LINODE_TOKEN: ${{ env.LINODE_TOKEN }} RUN_DB_FORK_TESTS: ${{ github.event.inputs.run_db_fork_tests }} RUN_DB_TESTS: ${{ github.event.inputs.run_db_tests }} RUN_ACLP_LOGS_STREAM_TESTS: ${{ github.event.inputs.run_aclp_logs_stream_tests }} + RUN_TRAFFIC_PEAK_TESTS: ${{ github.event.inputs.run_traffic_peak_tests }} TEST_SUITE: ${{ github.event.inputs.test_suite }} - name: Upload Test Report as Artifact diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index 46f37561f..6be127f95 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -22,6 +22,7 @@ AkamaiObjectStorageLogsDestinationDetails, CustomHTTPSLogsDestinationDetails, LogsStreamDetails, + TrafficPeakLogsDestinationDetails, ) __all__ = [ @@ -455,6 +456,7 @@ def destination_create( details: Union[ AkamaiObjectStorageLogsDestinationDetails, CustomHTTPSLogsDestinationDetails, + TrafficPeakLogsDestinationDetails, ], ) -> LogsDestination: """ @@ -495,17 +497,40 @@ def destination_create( ) ) + For a ``traffic_peak`` destination:: + + new_destination = client.monitor.destination_create( + label="traffic_peak_logs_destination", + type="traffic_peak", + details=TrafficPeakLogsDestinationDetails( + endpoint_url="https://my-site.com", + authentication=TrafficPeakDestinationAuthentication( + details=BasicAuthenticationDetails( + basic_authentication_user="user", + basic_authentication_password="pass", + ), + ), + data_compression="gzip", + content_type="application/json", + custom_headers=[ + CustomHeader(name="header", value="header_value") + ], + ) + ) + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-destination :param label: The name for this logs destination. :type label: str - :param type: The type of destination — ``akamai_object_storage`` or ``custom_https``. + :param type: The type of destination — ``akamai_object_storage``, ``custom_https``, or ``traffic_peak``. :type type: str or LogsDestinationType :param details: A typed details object matching the destination type. Use :class:`AkamaiObjectStorageLogsDestinationDetails` for ``akamai_object_storage`` or :class:`CustomHTTPSLogsDestinationDetails` - for ``custom_https``. - :type details: AkamaiObjectStorageLogsDestinationDetails or CustomHTTPSLogsDestinationDetails + for ``custom_https``. Use + :class:`TrafficPeakLogsDestinationDetails` for + ``traffic_peak``. + :type details: AkamaiObjectStorageLogsDestinationDetails, CustomHTTPSLogsDestinationDetails, or TrafficPeakLogsDestinationDetails :returns: The newly created logs destination. :rtype: LogsDestination diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index c23e4cead..afaf4a6a6 100644 --- a/linode_api4/objects/monitor.py +++ b/linode_api4/objects/monitor.py @@ -27,6 +27,8 @@ "ContentType", "CustomHeader", "CustomHTTPSLogsDestinationDetails", + "TrafficPeakDestinationAuthentication", + "TrafficPeakLogsDestinationDetails", "DataCompressionType", "DestinationAuthentication", "LogsDestinationDetailsBase", @@ -158,6 +160,7 @@ class LogsDestinationType(StrEnum): akamai_object_storage = "akamai_object_storage" custom_https = "custom_https" + traffic_peak = "traffic_peak" class AuthenticationType(StrEnum): @@ -588,6 +591,15 @@ class DestinationAuthentication(JSONObject): details: Optional[BasicAuthenticationDetails] = None +@dataclass +class TrafficPeakDestinationAuthentication(JSONObject): + """ + Authentication details for a TrafficPeak destination. + """ + + details: Optional[BasicAuthenticationDetails] = None + + @dataclass class CustomHeader(JSONObject): """ @@ -625,7 +637,8 @@ def load_by_type( Factory method that instantiates the correct details subclass based on the destination type string. - :param dest_type: The destination type (e.g. "akamai_object_storage", "custom_https"). + :param dest_type: The destination type (``akamai_object_storage``, + ``custom_https``, or ``traffic_peak``). :param json_dict: The raw JSON dict for the details block. :returns: A populated subclass instance, or None if json_dict is empty/None. """ @@ -638,6 +651,8 @@ def load_by_type( ) elif dest_type == LogsDestinationType.custom_https: return CustomHTTPSLogsDestinationDetails.from_json(json_dict) + elif dest_type == LogsDestinationType.traffic_peak: + return TrafficPeakLogsDestinationDetails.from_json(json_dict) return None @@ -656,6 +671,19 @@ class CustomHTTPSLogsDestinationDetails(LogsDestinationDetailsBase): client_certificate_details: Optional[ClientCertificateDetails] = None +@dataclass +class TrafficPeakLogsDestinationDetails(LogsDestinationDetailsBase): + """ + Represents the details block for TrafficPeak LogsDestination type. + """ + + endpoint_url: str = "" + authentication: Optional[TrafficPeakDestinationAuthentication] = None + data_compression: Optional[DataCompressionType] = None + content_type: Optional[ContentType] = None + custom_headers: Optional[List[CustomHeader]] = None + + @dataclass class AkamaiObjectStorageLogsDestinationDetails(LogsDestinationDetailsBase): """ diff --git a/test/fixtures/monitor_streams_4.json b/test/fixtures/monitor_streams_4.json new file mode 100644 index 000000000..9c64f503c --- /dev/null +++ b/test/fixtures/monitor_streams_4.json @@ -0,0 +1,35 @@ +{ + "id": 4, + "label": "traffic-peak-stream", + "type": "audit_logs", + "status": "active", + "destinations": [ + { + "id": 3, + "label": "traffic-peak-destination", + "type": "traffic_peak", + "details": { + "endpoint_url": "https://example.com/", + "authentication": { + "details": { + "basic_authentication_user": "user", + "basic_authentication_password": "password" + } + }, + "data_compression": "none", + "content_type": "application/json", + "custom_headers": [ + { + "name": "header", + "value": "header_value" + } + ] + } + } + ], + "created": "2026-09-01T00:00:00", + "updated": "2026-09-01T00:00:00", + "created_by": "tester", + "updated_by": "tester", + "version": 1 +} diff --git a/test/fixtures/monitor_streams_destinations_3.json b/test/fixtures/monitor_streams_destinations_3.json new file mode 100644 index 000000000..1215adf04 --- /dev/null +++ b/test/fixtures/monitor_streams_destinations_3.json @@ -0,0 +1,28 @@ +{ + "id": 3, + "version": 1, + "type": "traffic_peak", + "label": "traffic-peak-destination", + "status": "active", + "created_by": "tester", + "created": "2026-09-01T00:00:00", + "updated_by": "tester", + "updated": "2026-09-01T00:00:00", + "details": { + "endpoint_url": "https://example.com/", + "authentication": { + "details": { + "basic_authentication_user": "user", + "basic_authentication_password": "password" + } + }, + "data_compression": "none", + "content_type": "application/json", + "custom_headers": [ + { + "name": "header", + "value": "header_value" + } + ] + } +} diff --git a/test/integration/models/monitor/test_monitor_logs.py b/test/integration/models/monitor/test_monitor_logs.py index 9d4148c17..28825a594 100644 --- a/test/integration/models/monitor/test_monitor_logs.py +++ b/test/integration/models/monitor/test_monitor_logs.py @@ -17,9 +17,13 @@ ) from linode_api4.objects.monitor import ( AkamaiObjectStorageLogsDestinationDetails, + BasicAuthenticationDetails, + CustomHeader, LogsDestination, LogsStream, LogsStreamStatus, + TrafficPeakDestinationAuthentication, + TrafficPeakLogsDestinationDetails, ) _RUN_ACLP_LOGS_STREAM_TESTS = "RUN_ACLP_LOGS_STREAM_TESTS" @@ -28,6 +32,12 @@ not in {"yes", "true"}, reason=f"{_RUN_ACLP_LOGS_STREAM_TESTS} environment variable must be set to 'yes' or 'true'", ) +_RUN_TRAFFIC_PEAK_TESTS = "RUN_TRAFFIC_PEAK_TESTS" +_SKIP_TRAFFIC_PEAK_TESTS = pytest.mark.skipif( + os.getenv(_RUN_TRAFFIC_PEAK_TESTS, "").strip().lower() + not in {"yes", "true"}, + reason=f"{_RUN_TRAFFIC_PEAK_TESTS} environment variable must be set to 'yes' or 'true'", +) _STREAM_FIXTURE_CLEANUP_WAIT = 2700 _STREAM_FIXTURE_PROVISIONING_WAIT = 3600 @@ -87,17 +97,22 @@ def _delete_destination_with_bucket( client: LinodeClient, dest: LogsDestination, bucket: ObjectStorageBucket ): """Helper that deletes a logs destination and its backing OBJ bucket.""" + _delete_destination(client, dest) + _empty_bucket(client, bucket) + send_request_when_resource_available(timeout=100, func=bucket.delete) + + +def _delete_destination(client: LinodeClient, destination: LogsDestination): + """Helper that deletes a logs destination after its stream is detached.""" def no_stream_attached(): streams = client.monitor.streams() return all( - all(d.id != dest.id for d in s.destinations) for s in streams + all(d.id != destination.id for d in s.destinations) for s in streams ) wait_for_condition(30, _STREAM_FIXTURE_CLEANUP_WAIT, no_stream_attached) - send_request_when_resource_available(timeout=100, func=dest.delete) - _empty_bucket(client, bucket) - send_request_when_resource_available(timeout=100, func=bucket.delete) + send_request_when_resource_available(timeout=100, func=destination.delete) def _skip_if_streams_exist(client: LinodeClient): @@ -112,6 +127,40 @@ def _skip_if_streams_exist(client: LinodeClient): ) +def _wait_for_stream_updatable(client: LinodeClient, stream_id: int): + """ + Blocks until the stream with the given id reaches active or inactive status. + Updating destinations or other attributes puts the stream + back into a transitional state, and attempting to delete or modify it while + transitioning results in [400] errors. + """ + + def is_stream_updatable(): + stream = client.load(LogsStream, stream_id) + return stream.status in ( + LogsStreamStatus.active, + LogsStreamStatus.inactive, + ) + + wait_for_condition( + 30, _STREAM_FIXTURE_PROVISIONING_WAIT, is_stream_updatable + ) + + +def _delete_stream(client: LinodeClient, stream: LogsStream): + _wait_for_stream_updatable(client, stream.id) + send_request_when_resource_available(timeout=100, func=stream.delete) + + # The delete request returns 200 but stream deletion is async on the backend. + # Wait until the stream is fully gone before teardown continues, so that + # dependent fixtures can proceed with teardown. + def is_stream_deleted(): + existing = client.monitor.streams() + return all(s.id != stream.id for s in existing) + + wait_for_condition(30, _STREAM_FIXTURE_CLEANUP_WAIT, is_stream_deleted) + + def _empty_bucket(client: LinodeClient, bucket: ObjectStorageBucket): """ Helper function clearing objects in the test bucket so it can be deleted. @@ -280,6 +329,113 @@ def test_fails_to_create_destination_empty_required_fields( ) +@pytest.fixture(scope="function") +def create_traffic_peak_destination(test_linode_client: LinodeClient): + destination = test_linode_client.monitor.destination_create( + label=get_test_label(), + type="traffic_peak", + details=TrafficPeakLogsDestinationDetails( + endpoint_url="https://example.com/", + authentication=TrafficPeakDestinationAuthentication( + details=BasicAuthenticationDetails( + basic_authentication_user="user", + basic_authentication_password="password", + ) + ), + data_compression="none", + content_type="application/json", + custom_headers=[ + CustomHeader(name="X-Source", value="linode-api4-python") + ], + ), + ) + yield destination + _delete_destination(test_linode_client, destination) + + +@_SKIP_TRAFFIC_PEAK_TESTS +def test_create_and_get_traffic_peak_destination( + test_linode_client: LinodeClient, + create_traffic_peak_destination: LogsDestination, +): + """ + Test that fetching a TrafficPeak destination by ID returns the correct + destination with expected fields. + """ + destination = test_linode_client.load( + LogsDestination, create_traffic_peak_destination.id + ) + + assert isinstance(destination, LogsDestination) + assert destination.type == "traffic_peak" + assert isinstance(destination.details, TrafficPeakLogsDestinationDetails) + assert destination.details.endpoint_url == "https://example.com/" + assert isinstance( + destination.details.authentication, + TrafficPeakDestinationAuthentication, + ) + assert destination.details.authentication.details is not None + assert destination.details.data_compression == "none" + assert destination.details.content_type == "application/json" + assert destination.details.custom_headers[0].name == "X-Source" + + +@_SKIP_TRAFFIC_PEAK_TESTS +def test_list_traffic_peak_destinations( + test_linode_client: LinodeClient, + create_traffic_peak_destination: LogsDestination, +): + """ + Test that listing destinations returns a PaginatedList containing the + previously created TrafficPeak destination. + """ + destinations = test_linode_client.monitor.destinations( + LogsDestination.id == create_traffic_peak_destination.id + ) + + assert isinstance(destinations, PaginatedList) + assert len(destinations) == 1 + assert destinations[0].id == create_traffic_peak_destination.id + assert destinations[0].type == "traffic_peak" + + +@pytest.fixture(scope="function") +def create_traffic_peak_stream( + test_linode_client: LinodeClient, + create_traffic_peak_destination: LogsDestination, +): + _skip_if_streams_exist(test_linode_client) + stream = test_linode_client.monitor.stream_create( + label=get_test_label(), + destinations=[create_traffic_peak_destination.id], + type=LogsStreamType.audit_logs, + ) + yield stream + _delete_stream(test_linode_client, stream) + + +@_SKIP_TRAFFIC_PEAK_TESTS +@_SKIP_STREAM_TESTS +def test_create_stream_with_traffic_peak_destination( + test_linode_client: LinodeClient, + create_traffic_peak_stream: LogsStream, +): + """ + Test that loading a stream by ID returns the correct TrafficPeak destination + with expected fields. + """ + stream = test_linode_client.load(LogsStream, create_traffic_peak_stream.id) + destination = stream.destinations[0] + + assert destination.type == "traffic_peak" + assert isinstance(destination.details, TrafficPeakLogsDestinationDetails) + assert destination.details.endpoint_url == "https://example.com/" + assert isinstance( + destination.details.authentication, + TrafficPeakDestinationAuthentication, + ) + + @pytest.fixture(scope="session") def invalid_destination_error(test_linode_client: LinodeClient): """ @@ -346,41 +502,7 @@ def create_stream( assert stream.id is not None assert stream.status == LogsStreamStatus.provisioning yield stream - _stream_teardown(test_linode_client, stream) - - -def _wait_for_stream_updatable(client: LinodeClient, stream_id: int): - """ - Blocks until the stream with the given id reaches active or inactive status. - Updating destinations or other attributes puts the stream - back into a transitional state, and attempting to delete or modify it while - transitioning results in [400] errors. - """ - - def is_stream_updatable(): - stream = client.load(LogsStream, stream_id) - return stream.status in ( - LogsStreamStatus.active, - LogsStreamStatus.inactive, - ) - - wait_for_condition( - 30, _STREAM_FIXTURE_PROVISIONING_WAIT, is_stream_updatable - ) - - -def _stream_teardown(test_linode_client: LinodeClient, stream: LogsStream): - _wait_for_stream_updatable(test_linode_client, stream.id) - send_request_when_resource_available(timeout=100, func=stream.delete) - - # The delete request returns 200 but stream deletion is async on the backend. - # Wait until the stream is fully gone before teardown continues, so that - # dependent fixtures (e.g. create_secondary_destination) can proceed with teardown. - def is_stream_deleted(): - existing = test_linode_client.monitor.streams() - return all(s.id != stream.id for s in existing) - - wait_for_condition(30, _STREAM_FIXTURE_CLEANUP_WAIT, is_stream_deleted) + _delete_stream(test_linode_client, stream) @pytest.fixture(scope="session") diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py index c0999e485..fca248f7e 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -12,11 +12,15 @@ ) from linode_api4.objects.monitor import ( AkamaiObjectStorageLogsDestinationDetails, + BasicAuthenticationDetails, + CustomHeader, CustomHTTPSLogsDestinationDetails, DestinationAuthentication, LogsDestinationDetailsBase, LogsStreamDetails, LogsStreamType, + TrafficPeakDestinationAuthentication, + TrafficPeakLogsDestinationDetails, ) @@ -467,6 +471,123 @@ def test_create_custom_https_destination(self): self.assertIsInstance(result.details, CustomHTTPSLogsDestinationDetails) +class TrafficPeakLogsDestinationTest(ClientBaseCase): + """ + Tests methods for traffic_peak type LogsDestination. + """ + + def test_load_traffic_peak_destination(self): + """ + Test that loading a traffic_peak destination deserializes all nested fields correctly. + """ + destination = self.client.load(LogsDestination, 3) + + self.assertEqual(destination.type, "traffic_peak") + self.assertIsInstance( + destination.details, TrafficPeakLogsDestinationDetails + ) + self.assertEqual( + destination.details.endpoint_url, "https://example.com/" + ) + self.assertIsInstance( + destination.details.authentication, + TrafficPeakDestinationAuthentication, + ) + self.assertEqual( + destination.details.authentication.details.basic_authentication_user, + "user", + ) + self.assertEqual(destination.details.data_compression, "none") + self.assertEqual(destination.details.content_type, "application/json") + self.assertEqual(destination.details.custom_headers[0].name, "header") + self.assertEqual( + destination.details.custom_headers[0].value, "header_value" + ) + + def test_stream_with_traffic_peak_destination(self): + """ + Test that a LogsStreamDestination with type traffic_peak is deserialized correctly. + """ + stream = self.client.load(LogsStream, 4) + destination = stream.destinations[0] + + self.assertEqual(destination.type, "traffic_peak") + self.assertIsInstance( + destination.details, TrafficPeakLogsDestinationDetails + ) + self.assertEqual( + destination.details.endpoint_url, "https://example.com/" + ) + + def test_create_traffic_peak_destination(self): + """ + Test that destination_create with type=traffic_peak sends the correct payload. + """ + create_response = { + "id": 3, + "label": "traffic-peak-destination", + "type": "traffic_peak", + "status": "active", + "details": { + "endpoint_url": "https://example.com/", + "authentication": { + "details": { + "basic_authentication_user": "user", + "basic_authentication_password": "password", + } + }, + "data_compression": "none", + "content_type": "application/json", + "custom_headers": [{"name": "header", "value": "header_value"}], + }, + "created": "2026-09-01T00:00:00", + "updated": "2026-09-01T00:00:00", + "created_by": "tester", + "updated_by": "tester", + "version": 1, + } + + with self.mock_post(create_response) as mock: + result = self.client.monitor.destination_create( + label="traffic-peak-destination", + type="traffic_peak", + details=TrafficPeakLogsDestinationDetails( + endpoint_url="https://example.com/", + authentication=TrafficPeakDestinationAuthentication( + details=BasicAuthenticationDetails( + basic_authentication_user="user", + basic_authentication_password="password", + ), + ), + data_compression="none", + content_type="application/json", + custom_headers=[ + CustomHeader(name="header", value="header_value") + ], + ), + ) + + self.assertEqual(mock.call_url, "/monitor/streams/destinations") + self.assertEqual(mock.call_data["type"], "traffic_peak") + self.assertEqual( + mock.call_data["details"], + { + "endpoint_url": "https://example.com/", + "authentication": { + "details": { + "basic_authentication_user": "user", + "basic_authentication_password": "password", + }, + }, + "data_compression": "none", + "content_type": "application/json", + "custom_headers": [{"name": "header", "value": "header_value"}], + }, + ) + self.assertIsInstance(result, LogsDestination) + self.assertIsInstance(result.details, TrafficPeakLogsDestinationDetails) + + class LogsStreamTest(ClientBaseCase): """ Tests methods for LogsStream class.