Filter logs for specific logger in TestGetGcpCredentialsAndProjectId.test_disable_logging

[Taragolis]

TestGetGcpCredentialsAndProjectId.test_disable_logging time to time failed with selected error

  ____________ TestGetGcpCredentialsAndProjectId.test_disable_logging ____________
  
  self = <tests.providers.google.cloud.utils.test_credentials_provider.TestGetGcpCredentialsAndProjectId testMethod=test_disable_logging>
  mock_default = <MagicMock name='from_service_account_file' id='140574344364880'>
  mock_info = <MagicMock name='from_service_account_info' id='140574409759184'>
  mock_file = <MagicMock name='default' id='140574344364432'>
  
      @mock.patch("google.auth.default", return_value=("CREDENTIALS", "PROJECT_ID"))
      @mock.patch(
          'google.oauth2.service_account.Credentials.from_service_account_info',
      )
      @mock.patch(
          'google.oauth2.service_account.Credentials.from_service_account_file',
      )
      def test_disable_logging(self, mock_default, mock_info, mock_file):
          # assert not logs
          with self.assertLogs(level="DEBUG") as logs:
              logging.debug('nothing')
              get_credentials_and_project_id(disable_logging=True)
          assert logs.output == ['DEBUG:root:nothing']
      
          # assert no debug logs emitted from get_credentials_and_project_id
          with self.assertLogs(level="DEBUG") as logs:
              logging.debug('nothing')
              get_credentials_and_project_id(
                  keyfile_dict={'private_key': 'PRIVATE_KEY'},
                  disable_logging=True,
              )
          assert logs.output == ['DEBUG:root:nothing']
      
          # assert no debug logs emitted from get_credentials_and_project_id
          with self.assertLogs(level="DEBUG") as logs:
              logging.debug('nothing')
              get_credentials_and_project_id(
                  key_path='KEY.json',
                  disable_logging=True,
              )
  >       assert logs.output == ['DEBUG:root:nothing']
  E       AssertionError: assert ['DEBUG:root:...eede10>()]>>'] == ['DEBUG:root:nothing']
  E         Left contains one more item: 'ERROR:asyncio:Task was destroyed but it is pending!\ntask: <Task pending coro=<<async_generator_asend without __name__>()> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fda0feede10>()]>>'
  E         Use -v to get the full diff
  
  tests/providers/google/cloud/utils/test_credentials_provider.py:352: AssertionError

Due to the code of airflow.providers.google.cloud.utils.credentials_provider._CredentialProvider logging should disable only within the class, so asyncio error log not related to disabling log.

Add some workaround for check only logs with specific logger.

[Taragolis]

cc: @dstandish

[dstandish]

Hey @Taragolis , building off your work, i got inspired, and figured out we could wrap this logic in a context mgr and make it all even cleaner.

No obligation to accept it but here is the code

Suggested change

	assert_no_logs(logs.records)
	assert_no_logs(logs.records)
	@contextmanager
	def assert_no_logs(self, name, level):
	with self.assertLogs(level=level) as logs:
	logging.log(level=level, msg='nothing')
	yield
	records = [log_record for log_record in logs.records if log_record.name == name]
	if not records:
	return
	raise AssertionError(f"Did not expect any log message from logger={name!r}, but got: {records}")

[dstandish]

then the code becomes this:

        # assert no logs
        with self.assert_no_logs(name=logger_name, level="DEBUG") as logs:
            get_credentials_and_project_id(disable_logging=True)

        # assert no debug logs emitted from get_credentials_and_project_id
        with self.assert_no_logs(name=logger_name, level="DEBUG") as logs:
            get_credentials_and_project_id(
                keyfile_dict={'private_key': 'PRIVATE_KEY'},
                disable_logging=True,
            )

        # assert no debug logs emitted from get_credentials_and_project_id
        with self.assert_no_logs(name=logger_name, level="DEBUG") as logs:
            get_credentials_and_project_id(
                key_path='KEY.json',
                disable_logging=True,
            )