blob: ae07a3963b5f31276db2646caa15475d7c5325ff [file] [log] [blame]
# Copyright 2018 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Tests for gs_archive_server."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import base64
import gzip
import md5
import os
import StringIO
import unittest
import cherrypy
import mock
import pytest
import requests
from cherrypy.test import helper
import gs_archive_server
from chromite.lib import cros_logging as logging
_DIR = '/gs_archive_server_test'
# Some REAL files and info on Google Storage.
'a_plain_file': {
'path': '%s/' % _DIR,
'mime': 'application/octet-stream',
'size': 139,
'a_tar_file': {
'path': '%s/control_files.tar' % _DIR,
'members_md5': '0d5d60e9f10d41c60dd85a7f0081de5d',
# a tgz file with only one file "bar" which content is "foo\n"
_A_TGZ_FILE = base64.b64decode(
_A_TAR_FILE = gzip.GzipFile(fileobj=StringIO.StringIO(_A_TGZ_FILE)).read()
class UnmockedGSArchiveServerTest(helper.CPWebCase):
"""Some integration tests using cherrypy test framework."""
def setup_server():
"""An API used by cherrypy to setup test environment."""
def test_download_a_file(self):
"""Test normal files downloading."""
tested_file = _TEST_DATA['a_plain_file']
self.getPage('/download%(path)s' % tested_file)
self.assertHeader('Content-Type', tested_file['mime'])
self.assertEquals(len(self.body), tested_file['size'])
def test_download_a_non_existing_file(self):
"""Test downloading non-existing files."""
def test_download_against_unauthorized_bucket(self):
"""Test downloading from unauthorized bucket."""
class MockedGSArchiveServerTest(unittest.TestCase):
"""Unit test of GsArchiveServer using mock objects."""
def setUp(self):
"""Setup method."""
self.server = gs_archive_server.GsArchiveServer('')
def test_list_member(self):
"""Test list_member RPC."""
with mock.patch.object(self.server, '_caching_server') as caching_server:
rsp = mock.MagicMock() = rsp
rsp.iter_content.return_value = (_A_TAR_FILE[:100], _A_TAR_FILE[100:])
csv = list(self.server.list_member('baz.tar'))
self.assertEquals(len(csv), 1)
(filename, record_start, record_size, record_end,
content_start, size, content_end) = csv[0].split(',')
self.assertEquals(filename, 'bar')
self.assertEquals(record_start, '0')
self.assertEquals(record_size, '1024')
self.assertEquals(record_end, '1023') # 1024 - 1
self.assertEquals(content_start, '512')
self.assertEquals(size, '4')
self.assertEquals(content_end, '515\n') # 512 + 4 - 1
# test char quoting in file name
'index_tar_member_testing.tgz')) as f:
rsp.iter_content.return_value =
members = next(self.server.list_member('baz.tar'))
for csv in members.rstrip('\n').split('\n'):
# each line can be split into 7 elements, even ',' in filename
elements = csv.split(',')
self.assertEquals(len(elements), 7)
# elements from 1 to 6 are integers
_ = [int(d) for d in elements[1:7]]
def testing_server_setup():
"""Check if testing server is setup."""
rsp = requests.get(_TESTING_SERVER)
if rsp.status_code >= 500:
'Testing server %s has internal errors. Some tests are skipped!',
return False
return True
except Exception:
logging.warn('No testings server detected. Some tests are skipped!')
return False
@unittest.skipUnless(testing_server_setup(), 'Testing servers not available!')
class GsCacheBackendFunctionalTest(unittest.TestCase):
"""This is a functional blackbox test
These tests depend on a full setup of the server and proxy server.
If either of they is not available, all tests in this class are skipped.
def _get_page(self, url, headers=None, expect_status=200):
headers = headers.copy() if headers else {}
if not os.environ.get('WITH_CACHE', None):
headers['x-no-cache'] = '1' # bypass all caching to test the whole flow
rsp = requests.get('%s%s' % (_TESTING_SERVER, url), headers=headers,
self.assertEquals(rsp.status_code, expect_status)
return rsp
def _verify_md5(self, content, expected_md5):
"""Verify the md5 sum of input content equals to expteced value."""
m =
self.assertEquals(m.hexdigest(), expected_md5)
def test_download_plain_file(self):
"""Test download RPC."""
tested_file = _TEST_DATA['a_plain_file']
rsp = self._get_page('/download%(path)s' % tested_file)
self.assertEquals(rsp.headers['Content-Length'], str(tested_file['size']))
def test_list_member(self):
"""Test list member of a tar file."""
tested_file = _TEST_DATA['a_tar_file']
rsp = self._get_page('/list_member%(path)s' % tested_file)
self.assertEquals(rsp.headers['Content-Type'], 'text/csv;charset=utf-8')
self._verify_md5(rsp.content, tested_file['members_md5'])
if __name__ == "__main__":