oss-fuzz/infra/cifuzz/filestore/github_actions/github_actions_test.py

# Copyright 2021 Google LLC
#
# 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 for github_actions."""
import os
import shutil
import sys
import tarfile
import tempfile
import unittest
from unittest import mock

from pyfakefs import fake_filesystem_unittest

# pylint: disable=wrong-import-position
INFRA_DIR = os.path.dirname(
    os.path.dirname(os.path.dirname(os.path.dirname(
        os.path.abspath(__file__)))))
sys.path.append(INFRA_DIR)

from filestore import github_actions
import test_helpers

# pylint: disable=protected-access,no-self-use


class GithubActionsFilestoreTest(fake_filesystem_unittest.TestCase):
  """Tests for GithubActionsFilestore."""

  def setUp(self):
    test_helpers.patch_environ(self)
    self.github_token = 'example githubtoken'
    self.owner = 'exampleowner'
    self.repo = 'examplerepo'
    os.environ['GITHUB_REPOSITORY'] = f'{self.owner}/{self.repo}'
    self.config = test_helpers.create_run_config(github_token=self.github_token)
    self.artifact_name = 'corpus'
    self.corpus_dir = '/corpus-dir'
    self.testcase = os.path.join(self.corpus_dir, 'testcase')

  def _get_expected_http_headers(self):
    return {
        'Authorization': f'token {self.github_token}',
        'Accept': 'application/vnd.github.v3+json',
    }

  @mock.patch('filestore.github_actions.github_api.list_artifacts')
  def test_list_artifacts(self, mocked_list_artifacts):
    """Tests that _list_artifacts works as intended."""
    filestore = github_actions.GithubActionsFilestore(self.config)
    filestore._list_artifacts()
    mocked_list_artifacts.assert_called_with(self.owner, self.repo,
                                             self._get_expected_http_headers())

  @mock.patch('logging.warning')
  @mock.patch('filestore.github_actions.GithubActionsFilestore._list_artifacts',
              return_value=None)
  @mock.patch('filestore.github_actions.github_api.find_artifact',
              return_value=None)
  def test_download_latest_build_no_artifact(self, _, __, mocked_warning):
    """Tests that download_latest_build returns None and doesn't exception when
    find_artifact can't find an artifact."""
    filestore = github_actions.GithubActionsFilestore(self.config)
    name = 'build-name'
    build_dir = 'build-dir'
    self.assertIsNone(filestore.download_latest_build(name, build_dir))
    mocked_warning.assert_called_with('Could not download artifact: %s.',
                                      'cifuzz-' + name)

  @mock.patch('logging.warning')
  @mock.patch('filestore.github_actions.GithubActionsFilestore._list_artifacts',
              return_value=None)
  @mock.patch('filestore.github_actions.github_api.find_artifact',
              return_value=None)
  def test_download_corpus_no_artifact(self, _, __, mocked_warning):
    """Tests that download_corpus_build returns None and doesn't exception when
    find_artifact can't find an artifact."""
    filestore = github_actions.GithubActionsFilestore(self.config)
    name = 'corpus-name'
    dst_dir = 'corpus-dir'
    self.assertFalse(filestore.download_corpus(name, dst_dir))
    mocked_warning.assert_called_with('Could not download artifact: %s.',
                                      'cifuzz-' + name)

  @mock.patch('filestore.github_actions.tar_directory')
  @mock.patch('third_party.github_actions_toolkit.artifact.artifact_client'
              '.upload_artifact')
  def test_upload_directory(self, mocked_upload_artifact, mocked_tar_directory):
    """Tests that upload_directory invokes tar_directory and
    artifact_client.upload_artifact properly."""
    self._create_corpus_dir()
    filestore = github_actions.GithubActionsFilestore(self.config)
    filestore.upload_directory(self.artifact_name, self.corpus_dir)

    # Don't assert what second argument will be since it's a temporary
    # directory.
    self.assertEqual(mocked_tar_directory.call_args_list[0][0][0],
                     self.corpus_dir)

    # Don't assert what second and third arguments will be since they are
    # temporary directories.
    expected_artifact_name = 'cifuzz-' + self.artifact_name
    self.assertEqual(mocked_upload_artifact.call_args_list[0][0][0],
                     expected_artifact_name)

    # Assert artifacts list contains one tarfile.
    artifacts_list = mocked_upload_artifact.call_args_list[0][0][1]
    self.assertEqual(len(artifacts_list), 1)
    self.assertEqual(os.path.basename(artifacts_list[0]),
                     expected_artifact_name + '.tar')

  def _create_corpus_dir(self):
    """Sets up pyfakefs and creates a corpus directory containing
    self.testcase."""
    self.setUpPyfakefs()
    self.fs.create_file(self.testcase, contents='hi')

  @mock.patch('filestore.github_actions.GithubActionsFilestore._find_artifact')
  @mock.patch('http_utils.download_and_unpack_zip')
  def test_download_artifact(self, mocked_download_and_unpack_zip,
                             mocked_find_artifact):
    """Tests that _download_artifact works as intended."""
    artifact_download_url = 'http://example.com/download'
    artifact_listing = {
        'expired': False,
        'name': self.artifact_name,
        'archive_download_url': artifact_download_url
    }
    mocked_find_artifact.return_value = artifact_listing

    self._create_corpus_dir()
    with tempfile.TemporaryDirectory() as temp_dir:
      # Create a tarball.
      archive_path = os.path.join(temp_dir, f'cifuzz-{self.artifact_name}.tar')
      github_actions.tar_directory(self.corpus_dir, archive_path)

      artifact_download_dst_dir = os.path.join(temp_dir, 'dst')
      os.mkdir(artifact_download_dst_dir)

      def mock_download_and_unpack_zip(url, download_artifact_temp_dir,
                                       headers):
        self.assertEqual(url, artifact_download_url)
        self.assertEqual(headers, self._get_expected_http_headers())
        shutil.copy(
            archive_path,
            os.path.join(download_artifact_temp_dir,
                         os.path.basename(archive_path)))
        return True

      mocked_download_and_unpack_zip.side_effect = mock_download_and_unpack_zip
      filestore = github_actions.GithubActionsFilestore(self.config)
      self.assertTrue(
          filestore._download_artifact(self.artifact_name,
                                       artifact_download_dst_dir))
      mocked_find_artifact.assert_called_with(f'cifuzz-{self.artifact_name}')
      self.assertTrue(
          os.path.exists(
              os.path.join(artifact_download_dst_dir,
                           os.path.basename(self.testcase))))

  @mock.patch('filestore.github_actions.github_api.list_artifacts')
  def test_find_artifact(self, mocked_list_artifacts):
    """Tests that _find_artifact works as intended."""
    artifact_listing_1 = {
        'expired': False,
        'name': 'other',
        'archive_download_url': 'http://download1'
    }
    artifact_listing_2 = {
        'expired': False,
        'name': self.artifact_name,
        'archive_download_url': 'http://download2'
    }
    artifact_listing_3 = {
        'expired': True,
        'name': self.artifact_name,
        'archive_download_url': 'http://download3'
    }
    artifact_listing_4 = {
        'expired': False,
        'name': self.artifact_name,
        'archive_download_url': 'http://download4'
    }
    artifacts = [
        artifact_listing_1, artifact_listing_2, artifact_listing_3,
        artifact_listing_4
    ]
    mocked_list_artifacts.return_value = artifacts
    filestore = github_actions.GithubActionsFilestore(self.config)
    # Test that find_artifact will return the most recent unexpired artifact
    # with the correct name.
    self.assertEqual(filestore._find_artifact(self.artifact_name),
                     artifact_listing_2)
    mocked_list_artifacts.assert_called_with(self.owner, self.repo,
                                             self._get_expected_http_headers())


class TarDirectoryTest(unittest.TestCase):
  """Tests for tar_directory."""

  def test_tar_directory(self):
    """Tests that tar_directory writes the archive to the correct location and
    archives properly."""
    with tempfile.TemporaryDirectory() as temp_dir:
      archive_path = os.path.join(temp_dir, 'myarchive.tar')
      archived_dir = os.path.join(temp_dir, 'toarchive')
      os.mkdir(archived_dir)
      archived_filename = 'file1'
      archived_file_path = os.path.join(archived_dir, archived_filename)
      with open(archived_file_path, 'w') as file_handle:
        file_handle.write('hi')
      github_actions.tar_directory(archived_dir, archive_path)
      self.assertTrue(os.path.exists(archive_path))

      # Now check it archives correctly.
      unpacked_directory = os.path.join(temp_dir, 'unpacked')
      with tarfile.TarFile(archive_path) as artifact_tarfile:
        artifact_tarfile.extractall(unpacked_directory)
      unpacked_archived_file_path = os.path.join(unpacked_directory,
                                                 archived_filename)
      self.assertTrue(os.path.exists(unpacked_archived_file_path))