code: def test_should_read_logs_with_pagination(
      self, mock_client, mock_get_creds_and_project_id
  ):
      mock_client.return_value.list_log_entries.side_effect = [
          mock.MagicMock(
              pages=iter(
                  [
                      _create_list_log_entries_response_mock(
                          ["MSG1", "MSG2"], "TOKEN1"
                      )
                  ]
              )
          ),
          mock.MagicMock(
              pages=iter(
                  [
                      _create_list_log_entries_response_mock(
                          ["MSG3", "MSG4"], None
                      )
                  ]
              )
          ),
      ]
      mock_get_creds_and_project_id.return_value = (
          "creds",
          "project_id",
      )
      stackdriver_task_handler = self._setup_handler()
  
      logs, metadata1 = stackdriver_task_handler.read(self.ti, 3)
      mock_client.return_value.list_log_entries.assert_called_once_with(
          request=ListLogEntriesRequest(
              resource_names=["projects/project_id"],
              filter=(
                  '''resource.type="global"
  logName="projects/project_id/logs/airflow"
  labels.task_id="task_for_testing_stackdriver_task_handler"
  labels.dag_id="dag_for_testing_stackdriver_file_task_handler"
  labels.execution_date="2016-01-01T00:00:00+00:00"
  labels.try_number="3"'''
              ),
              order_by="timestamp asc",
              page_size=1000,
              page_token=None,
          )
      )
      assert [(("default-hostname", "MSG1\nMSG2"),)] == logs
      assert [
          {"end_of_log": False, "next_page_token": "TOKEN1"}
      ] == metadata1
  
      mock_client.return_value.list_log_entries.return_value.next_page_token = (
          None
      )
      logs, metadata2 = stackdriver_task_handler.read(
          self.ti, 3, metadata1[0]
      )
  
      mock_client.return_value.list_log_entries.assert_called_with(
          request=ListLogEntriesRequest(
              resource_names=["projects/project_id"],
              filter=(
                  'resource.type="global"\n'
                  'logName="projects/project_id/logs/airflow"\n'
                  'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
                  'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
                  'labels.execution_date="2016-01-01T00:00:00+00:00"\n'
                  'labels.try_number="3"'
              ),
              order_by="timestamp asc",
              page_size=1000,
              page_token="TOKEN1",
          )
      )
      assert [(("default-hostname", "MSG3\nMSG4"),)] == logs
      assert [{"end_of_log": True}] == metadata2
  
imports: from __future__ import annotations
  import logging
  from unittest import mock
  from urllib.parse import parse_qs, urlsplit
  import pytest
  from google.cloud.logging import Resource
  from google.cloud.logging_v2.types import ListLogEntriesRequest, ListLogEntriesResponse, LogEntry
  from airflow.providers.google.cloud.log.stackdriver_task_handler import StackdriverTaskHandler
  from airflow.utils import timezone
  from airflow.utils.state import TaskInstanceState
  from tests.test_utils.db import clear_db_dags, clear_db_runs
label_desc: Use image node_id 8 for calling the TestStackdriverLoggingHandlerTask obj's underlying member method code with example usage: obj.test_should_read_logs_with_pagination(mock_client, mock_get_creds_and_project_id) without return types