Source code
Revision control
Copy as Markdown
Other Tools
#!/usr/bin/env vpython3
# Copyright 2022 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""File for testing serve_repo.py."""
import argparse
import json
import unittest
import unittest.mock as mock
from types import SimpleNamespace
import serve_repo
from common import REPO_ALIAS, register_device_args
_REPO_DIR = 'test_repo_dir'
_REPO_NAME = 'test_repo_name'
_TARGET = 'test_target'
_NO_SERVERS_LIST = json.dumps({'ok': {'data': []}})
_SERVERS_LIST = json.dumps(
{'ok': {
'data': [{
'name': _REPO_NAME,
'repo_path': _REPO_DIR
}]
}})
_WRONG_SERVERS_LIST = json.dumps(
{'ok': {
'data': [{
'name': 'wrong_name',
'repo_path': _REPO_DIR
}]
}})
# Tests private functions.
# pylint: disable=protected-access
class ServeRepoTest(unittest.TestCase):
"""Unittests for serve_repo.py."""
@mock.patch('serve_repo.run_ffx_command')
def test_start_server(self, mock_ffx) -> None:
"""Test |_start_serving| function for start."""
mock_ffx.side_effect = [
SimpleNamespace(returncode=1, stderr='err1', stdout=''),
SimpleNamespace(returncode=0, stdout=_SERVERS_LIST, stderr=''),
SimpleNamespace(returncode=0, stdout=_SERVERS_LIST, stderr=''),
SimpleNamespace(returncode=0, stdout='', stderr='')
]
serve_repo._start_serving(_REPO_DIR, _REPO_NAME, _TARGET)
self.assertEqual(mock_ffx.call_count, 3)
first_call = mock_ffx.call_args_list[0]
self.assertEqual(
mock.call(cmd=[
'repository', 'server', 'start', '--background', '--address',
'[::]:0', '--repository', _REPO_NAME, '--repo-path', _REPO_DIR,
'--no-device'
],
check=False), first_call)
second_call = mock_ffx.call_args_list[1]
self.assertEqual(
mock.call(cmd=[
'--machine', 'json', 'repository', 'server', 'list', '--name',
_REPO_NAME
],
check=False,
capture_output=True), second_call)
third_call = mock_ffx.call_args_list[2]
self.assertEqual(
mock.call(cmd=[
'target', 'repository', 'register', '-r', _REPO_NAME,
'--alias', REPO_ALIAS
],
target_id=_TARGET), third_call)
@mock.patch('serve_repo.run_ffx_command')
def test_assert_server_running(self, mock_ffx) -> None:
"""Test |_assert_server_running| function for start."""
mock_ffx.side_effect = [
SimpleNamespace(returncode=0, stdout=_SERVERS_LIST)
]
# Raises an error if there is a problem, so no need to check for
# RuntimeError.
try:
serve_repo._assert_server_running(_REPO_NAME)
except RuntimeError as err:
self.fail(f'Unexpected error: {err}')
@mock.patch('serve_repo.run_ffx_command')
def test_is_server_not_running(self, mock_ffx) -> None:
"""Test |_assert_server_running| function for start with no server."""
mock_ffx.return_value = SimpleNamespace(returncode=0,
stdout=_NO_SERVERS_LIST,
stderr='')
with self.assertRaises(RuntimeError):
serve_repo._assert_server_running(_REPO_NAME)
@mock.patch('serve_repo.run_ffx_command')
def test_is_wrong_server_running(self, mock_ffx) -> None:
"""Test |_assert_server_running| function for start with no server."""
mock_ffx.return_value = SimpleNamespace(returncode=0,
stdout=_WRONG_SERVERS_LIST,
stderr='')
with self.assertRaises(RuntimeError):
serve_repo._assert_server_running(_REPO_NAME)
@mock.patch('serve_repo.run_ffx_command')
def test_is_server_not_running_bad_ffx(self, mock_ffx) -> None:
"""Test |_assert_server_running| function for start with bad ffx."""
mock_ffx.return_value = SimpleNamespace(returncode=1,
stderr='Some error',
stdout='')
with self.assertRaises(RuntimeError):
serve_repo._assert_server_running(_REPO_NAME)
@mock.patch('serve_repo.run_ffx_command')
def test_is_server_not_running_bad_json(self, mock_ffx) -> None:
"""Test |_assert_server_running| function for start with bad ffx."""
mock_ffx.return_value = SimpleNamespace(returncode=0,
stderr='',
stdout='{"some": bad...')
with self.assertRaises(RuntimeError):
serve_repo._assert_server_running(_REPO_NAME)
@mock.patch('serve_repo.run_ffx_command')
def test_stop_server(self, mock_ffx) -> None:
"""Test |_stop_serving| function for stop."""
serve_repo._stop_serving(_REPO_NAME, _TARGET)
self.assertEqual(mock_ffx.call_count, 2)
first_call = mock_ffx.call_args_list[0]
self.assertEqual(
mock.call(
cmd=['target', 'repository', 'deregister', '-r', _REPO_NAME],
target_id=_TARGET,
check=False), first_call)
second_call = mock_ffx.call_args_list[1]
self.assertEqual(
mock.call(cmd=['repository', 'server', 'stop', _REPO_NAME],
check=False), second_call)
@mock.patch('serve_repo._start_serving')
@mock.patch('serve_repo._stop_serving')
def test_serve_repository(self, mock_stop, mock_start) -> None:
"""Tests |serve_repository| context manager."""
parser = argparse.ArgumentParser()
serve_repo.register_serve_args(parser)
register_device_args(parser)
with serve_repo.serve_repository(
parser.parse_args([
'--repo', _REPO_DIR, '--repo-name', _REPO_NAME,
'--target-id', _TARGET
])):
self.assertEqual(mock_start.call_count, 1)
self.assertEqual(mock_stop.call_count, 1)
if __name__ == '__main__':
unittest.main()