src/test/py/bazel/bzlmod/bzlmod_credentials_test.py

# pylint: disable=g-backslash-continuation
# Copyright 2023 The Bazel Authors. All rights reserved.
#
# 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.
"""Tests using credentials to connect to the bzlmod registry."""

import base64
import os
import tempfile
import unittest

from src.test.py.bazel import test_base
from src.test.py.bazel.bzlmod.test_utils import BazelRegistry
from src.test.py.bazel.bzlmod.test_utils import StaticHTTPServer


class BzlmodCredentialsTest(test_base.TestBase):
  """Test class for using credentials to connect to the bzlmod registry."""

  def setUp(self):
    test_base.TestBase.setUp(self)
    self.registries_work_dir = tempfile.mkdtemp(dir=self._test_cwd)
    self.registry_root = os.path.join(self.registries_work_dir, 'main')
    self.main_registry = BazelRegistry(self.registry_root)
    self.main_registry.createCcModule('aaa', '1.0')

    self.ScratchFile(
        '.bazelrc',
        [
            # In ipv6 only network, this has to be enabled.
            # 'startup --host_jvm_args=-Djava.net.preferIPv6Addresses=true',
            'common --enable_bzlmod',
            # Disable yanked version check so we are not affected BCR changes.
            'common --allow_yanked_versions=all',
        ],
    )
    self.ScratchFile('WORKSPACE')
    # The existence of WORKSPACE.bzlmod prevents WORKSPACE prefixes or suffixes
    # from being used; this allows us to test built-in modules actually work
    self.ScratchFile('WORKSPACE.bzlmod')
    self.ScratchFile(
        'MODULE.bazel',
        [
            'bazel_dep(name = "aaa", version = "1.0")',
        ],
    )
    self.ScratchFile(
        'BUILD',
        [
            'cc_binary(',
            '  name = "main",',
            '  srcs = ["main.cc"],',
            '  deps = ["@aaa//:lib_aaa"],',
            ')',
        ],
    )
    self.ScratchFile(
        'main.cc',
        [
            '#include "aaa.h"',
            'int main() {',
            '    hello_aaa("main function");',
            '}',
        ],
    )
    self.ScratchFile(
        'credhelper',
        [
            '#!/usr/bin/env python3',
            'import sys',
            'if "127.0.0.1" in sys.stdin.read():',
            '    print("""{"headers":{"Authorization":["Bearer TOKEN"]}}""")',
            'else:',
            '    print("""{}""")',
        ],
        executable=True,
    )
    self.ScratchFile(
        '.netrc',
        [
            'machine 127.0.0.1',
            'login foo',
            'password bar',
        ],
    )

  def testUnauthenticated(self):
    with StaticHTTPServer(self.registry_root) as static_server:
      _, stdout, _ = self.RunBazel([
          'run',
          '--registry=' + static_server.getURL(),
          '--registry=https://bcr.bazel.build',
          '//:main',
      ])
      self.assertIn('main function => aaa@1.0', stdout)

  def testMissingCredentials(self):
    with StaticHTTPServer(
        self.registry_root, expected_auth='Bearer TOKEN'
    ) as static_server:
      _, _, stderr = self.RunBazel(
          [
              'run',
              '--registry=' + static_server.getURL(),
              '--registry=https://bcr.bazel.build',
              '//:main',
          ],
          allow_failure=True,
      )
      self.assertIn('GET returned 401 Unauthorized', '\n'.join(stderr))

  def testCredentialsFromHelper(self):
    with StaticHTTPServer(
        self.registry_root, expected_auth='Bearer TOKEN'
    ) as static_server:
      _, stdout, _ = self.RunBazel([
          'run',
          '--credential_helper=%workspace%/credhelper',
          '--registry=' + static_server.getURL(),
          '--registry=https://bcr.bazel.build',
          '//:main',
      ])
      self.assertIn('main function => aaa@1.0', stdout)

  def testCredentialsFromNetrc(self):
    expected_auth = 'Basic ' + base64.b64encode(b'foo:bar').decode('ascii')

    with StaticHTTPServer(
        self.registry_root, expected_auth=expected_auth
    ) as static_server:
      _, stdout, _ = self.RunBazel(
          [
              'run',
              '--registry=' + static_server.getURL(),
              '--registry=https://bcr.bazel.build',
              '//:main',
          ],
          env_add={'NETRC': self.Path('.netrc')},
      )
      self.assertIn('main function => aaa@1.0', stdout)

  def testCredentialsFromHelperOverrideNetrc(self):
    with StaticHTTPServer(
        self.registry_root, expected_auth='Bearer TOKEN'
    ) as static_server:
      _, stdout, _ = self.RunBazel(
          [
              'run',
              '--credential_helper=%workspace%/credhelper',
              '--registry=' + static_server.getURL(),
              '--registry=https://bcr.bazel.build',
              '//:main',
          ],
          env_add={'NETRC': self.Path('.netrc')},
      )
      self.assertIn('main function => aaa@1.0', stdout)


if __name__ == '__main__':
  unittest.main()