Google Git
Sign in
chromium / infra / luci / luci-py / HEAD / . / appengine / swarming / handlers_prpc_test.py
blob: 1dc7ee17694117284b19b979a20fea17e73b7d4c [file] [log] [blame]
#!/usr/bin/env vpython
# coding=utf-8
# Copyright 2018 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.
import datetime
import logging
import os
import random
import sys
import unittest
import json
import cgi
import mock
import base64
from parameterized import parameterized
import test_env_handlers
from google.appengine.ext import ndb
import google.protobuf as proto
import webapp2
import webtest
from test_support import test_case
from components import auth
from components import ereporter2
from components.prpc import encoding
from components import utils
from proto.api_v2 import swarming_pb2
from proto.internals import rbe_pb2
from proto.config import realms_pb2
from server import acl
from server import bot_code
from server import bot_management
from server import bot_management
from server import config
from server import rbe
from server import service_accounts
from server import task_pack
from server import task_request
from server import task_result
from server import task_scheduler
from server import task_to_run
import handlers_bot
import handlers_prpc
import message_conversion_prpc
# 2010-01-02T03:04:05Z
def fmtdate(d, fmt='%Y-%m-%dT%H:%M:%SZ'):
"""Formats a datetime.datetime instance to the format generated by the API.
API datetime responses return at UTC+0 hence the Z at the end."""
return unicode(d.strftime(fmt))
def _decode(raw, dst):
assert raw[:5] == ')]}\'\n', raw[:5]
return encoding.get_decoder(encoding.Encoding.JSON)(raw[5:], dst)
def _decode_dict(dct, out):
return proto.json_format.ParseDict(dct, out)
def _encode(d):
raw = encoding.get_encoder(encoding.Encoding.JSON)(d)
assert raw[:5] == ')]}\'\n', raw[:5]
return raw[5:]
def _bot_event(event_type, bot_id, **kwargs):
args = {
'authenticated_as': 'bot:whitelisted-ip',
'dimensions': {
u'id': [bot_id],
u'pool': [u'default']
},
'external_ip': '8.8.4.4',
'state': {
'ram': 65
},
'version': '123456789',
'quarantined': False,
'maintenance_msg': None,
'task_id': None,
'task_name': None,
'register_dimensions': event_type.startswith('request_')
}
args.update(kwargs)
return bot_management.bot_event(event_type, bot_id, **args)
def apply_default_for_task_props(props):
props.cipd_input.client_package.package_name = 'infra/tools/cipd/${platform}'
props.cipd_input.client_package.version = 'git_revision:deadbeef'
props.cipd_input.packages.extend([
swarming_pb2.CipdPackage(package_name='rm',
path='bin',
version='git_revision:deadbeef')
])
props.cipd_input.server = 'https://pool.config.cipd.example.com'
props.command[:] = ['python', '-c', 'print(1)']
props.containment.containment_type = swarming_pb2.Containment.AUTO
props.dimensions.extend([
swarming_pb2.StringPair(key='os', value='Amiga'),
swarming_pb2.StringPair(key='pool', value='default')
])
props.execution_timeout_secs = 3600
props.grace_period_secs = 30
props.idempotent = False
props.io_timeout_secs = 1200
props.outputs[:] = ['foo', 'path/to/foobar']
def apply_defaults_for_request(request):
"""Applies some default expectations to a TaskRequestResponse
To be used for expectations.
"""
# This assumes:
# self.mock(random, 'getrandbits', lambda _: 0x88)
request.authenticated = 'user:user@example.com'
request.expiration_secs = 86400
request.task_id = '5cee488008810'
request.name = 'job1'
request.priority = 20
request.service_account = 'none'
request.tags[:] = [
u'a:tag',
u'authenticated:user:user@example.com',
u'os:Amiga',
u'pool:default',
u'priority:20',
u'realm:none',
u'service_account:none',
u'swarming.pool.template:none',
u'swarming.pool.version:pools_cfg_rev',
u'user:joe@localhost',
]
request.user = 'joe@localhost'
request.bot_ping_tolerance_secs = 600
class PrpcTest(test_env_handlers.AppTestBase):
no_run = 1
service = None
def setUp(self):
super(PrpcTest, self).setUp()
routes = handlers_prpc.get_routes(debug=True) + handlers_bot.get_routes()
self.app = webtest.TestApp(
webapp2.WSGIApplication(routes, debug=True),
extra_environ={
'REMOTE_ADDR': self.source_ip,
'SERVER_SOFTWARE': os.environ['SERVER_SOFTWARE'],
},
)
self._headers = {
'Content-Type': encoding.Encoding.JSON[1],
'Accept': encoding.Encoding.JSON[1],
}
def apply_defaults_for_result_summary(self, result):
"""Applies default expectations to a TaskResultResponse initialized from a
TaskResultSummary ndb entity.
To be used for expectations.
"""
# This assumes:
# self.mock(random, 'getrandbits', lambda _: 0x88)
del result.bot_dimensions[:]
result.bot_dimensions.extend([
swarming_pb2.StringListPair(key='id', value=['bot1']),
swarming_pb2.StringListPair(key='os', value=['Amiga']),
swarming_pb2.StringListPair(key='pool', value=['default'])
])
result.bot_id = 'bot1'
result.bot_version = self.bot_version
result.current_task_slice = 0
result.failure = False
result.internal_failure = False
result.name = 'job1'
result.run_id = '5cee488008811'
result.server_versions[:] = [u'v1a']
result.state = swarming_pb2.TaskState.COMPLETED
result.tags[:] = [
'a:tag',
'os:Amiga',
'pool:default',
'priority:20',
'realm:none',
'service_account:none',
'swarming.pool.template:no_config',
'user:joe@localhost',
]
result.task_id = '5cee488008810'
result.user = 'joe@localhost'
def apply_defaults_for_run_result(self, result):
"""Returns a serialized swarming_rpcs.TaskResult initialized from a
TaskRunResult.
To be used for expectations.
"""
result.bot_dimensions.extend([
swarming_pb2.StringListPair(key='id', value=['bot1']),
swarming_pb2.StringListPair(key='os', value=['Amiga']),
swarming_pb2.StringListPair(key='pool', value=['default'])
])
result.bot_id = 'bot1'
result.bot_version = self.bot_version
result.costs_usd.extend([0.0])
result.current_task_slice = 0
result.failure = False
result.internal_failure = False
result.name = 'job1'
result.run_id = '5cee488008811'
result.server_versions.extend(['v1a'])
result.state = swarming_pb2.TaskState.RUNNING
result.task_id = '5cee488008811'
return result
def gen_perf_stats_prpc(self, stats):
"""Returns a serialized swarming_rpcs.PerformanceStats.
To be used for expectations.
"""
stats.bot_overhead = 0.1
stats.cache_trim.duration = 0.1
stats.isolated_download.duration = 1.0
stats.isolated_download.initial_size = 100000
stats.isolated_download.initial_number_items = 10
stats.isolated_download.items_cold = b'x\234\023\001\000\000\025\000\025'
stats.isolated_download.items_hot = b'x\234\223\343\002\000\000H\000)'
stats.isolated_download.num_items_cold = 1
stats.isolated_download.total_bytes_items_cold = 20
stats.isolated_download.num_items_hot = 2
stats.isolated_download.total_bytes_items_hot = 70
stats.isolated_upload.duration = 2.0
stats.isolated_upload.items_cold = b'x\234cdT\003\000\000.\000)'
stats.isolated_upload.items_hot = b'x\234cdd\324\007\000\000<\0003'
stats.isolated_upload.num_items_cold = 3
stats.isolated_upload.total_bytes_items_cold = 43
stats.isolated_upload.num_items_hot = 4
stats.isolated_upload.total_bytes_items_hot = 56
stats.cleanup.duration = 0.1
stats.package_installation.duration = 3.0
stats.named_caches_install.duration = 0.1
stats.named_caches_uninstall.duration = 0.1
def _new_task_request_prpc(self, use_default_slice=False):
ntr = swarming_pb2.NewTaskRequest(expiration_secs=24 * 60 * 60,
name='job1',
priority=20,
tags=[u'a:tag'],
user='joe@localhost',
bot_ping_tolerance_secs=600)
if use_default_slice:
ts = swarming_pb2.TaskSlice(expiration_secs=180, wait_for_capacity=True)
# Hack to get min line length.
props = ts.properties
props.cipd_input.client_package.package_name = ('infra/tools/'
'cipd/${platform}')
props.cipd_input.client_package.version = 'git_revision:deadbeef'
props.cipd_input.packages.extend([
swarming_pb2.CipdPackage(package_name='rm',
path='bin',
version='git_revision:deadbeef')
])
props.cipd_input.server = 'https://pool.config.cipd.example.com'
props.command[:] = ['python', '-c', 'print(1)']
props.containment.containment_type = swarming_pb2.Containment.AUTO
props.dimensions.extend([
swarming_pb2.StringPair(key='os', value='Amiga'),
swarming_pb2.StringPair(key='pool', value='default')
])
props.execution_timeout_secs = 3600
props.grace_period_secs = 30
props.idempotent = False
props.io_timeout_secs = 1200
props.outputs[:] = ['foo', 'path/to/foobar']
props.command[:] = ['python', 'run_test.py']
ntr.task_slices.extend([ts])
ntr.ClearField('expiration_secs')
return ntr
def _client_create_task_prpc(self, request):
"""Creates a task via the pRPC API."""
response = self.post_prpc('NewTask', request, service='swarming.v2.Tasks')
actual = swarming_pb2.TaskRequestMetadataResponse()
_decode(response.body, actual)
return actual, actual.task_id
def _client_cancel_task_prpc(self, task_id):
"""Cancels a pending task via pRPC API."""
response = self.post_prpc('CancelTask',
swarming_pb2.TaskCancelRequest(
task_id=task_id, kill_running=False),
service='swarming.v2.Tasks')
actual = swarming_pb2.CancelResponse()
_decode(response.body, actual)
self.assertTrue(actual.canceled)
def post_prpc(self, rpc, request, expect_errors=False, service=None):
if not service:
assert self.service, "Child classes must define service"
service = self.service
return self.app.post('/prpc/%s/%s' % (service, rpc),
_encode(request),
self._headers,
expect_errors=expect_errors)
class BotServicePrpcTest(PrpcTest):
"""Tests the pRPC handlers."""
def setUp(self):
super(BotServicePrpcTest, self).setUp()
self.service = "swarming.v2.Bots"
self.now = datetime.datetime(2010, 1, 2, 3, 4, 5)
self.mock_now(self.now)
self.mock_default_pool_acl([])
self.mock_tq_tasks()
def test_bot_get(self):
self.set_as_bot()
self.do_handshake()
self.set_as_privileged_user()
request = swarming_pb2.BotRequest(bot_id="bot1", )
resp = self.post_prpc("GetBot", request)
actual_info = swarming_pb2.BotInfo()
_decode(resp.body, actual_info)
expected = message_conversion_prpc.bot_info_response(
bot_management.get_info_key('bot1').get())
self.assertEqual(expected, actual_info)
def test_bot_in_maintenance(self):
self.set_as_privileged_user()
_bot_event('request_sleep', 'bot1', maintenance_msg='boiling water')
request = swarming_pb2.BotRequest(bot_id="bot1", )
resp = self.post_prpc("GetBot", request)
actual_info = swarming_pb2.BotInfo()
_decode(resp.body, actual_info)
expected = message_conversion_prpc.bot_info_response(
bot_management.get_info_key('bot1').get())
self.assertTrue(actual_info.maintenance_msg)
self.assertEqual(expected, actual_info)
def test_bot_get_not_found(self):
self.set_as_privileged_user()
request = swarming_pb2.BotRequest(bot_id="bot1", )
resp = self.post_prpc("GetBot", request, expect_errors=True)
self.assertEqual(resp.status, '404 Not Found')
self.assertEqual(resp.body, 'bot1 not found.')
def test_get_deleted_bot(self):
bot_id = "bot1"
_bot_event('bot_connected', bot_id=bot_id, state={'foo': 0})
self.set_as_admin()
request = swarming_pb2.BotRequest(bot_id=bot_id, )
resp = self.post_prpc("DeleteBot", request)
expected = swarming_pb2.DeleteResponse(deleted=True)
actual = swarming_pb2.DeleteResponse(deleted=True)
_decode(resp.body, actual)
self.assertEqual(expected, actual)
# Now check whether the bot still exists, it will return not found
request = swarming_pb2.BotRequest(bot_id=bot_id, )
resp = self.post_prpc("GetBot", request)
self.assertEqual(resp.status, '200 OK')
ghost_bot = swarming_pb2.BotInfo()
_decode(resp.body, ghost_bot)
self.assertTrue(ghost_bot.deleted)
def test_unauthorized_bot_not_found(self):
self.set_as_anonymous()
request = swarming_pb2.BotRequest(bot_id="bot1", )
resp = self.post_prpc("GetBot", request, expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
self.assertEqual(resp.body, 'Access is denied.')
@parameterized.expand([
'GetBot',
'ListBotEvents',
])
def test_get_ok_realm(self, api):
# non-privileged user with realm permission.
self.mock_auth_db([auth.Permission('swarming.pools.listBots')])
self.set_as_user()
_bot_event('bot_connected', bot_id='id1')
resp = self.post_prpc(api,
swarming_pb2.BotRequest(bot_id='id1'),
expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
@parameterized.expand([
'GetBot',
'ListBotEvents',
])
def test_get_forbidden(self, api):
self.mock_auth_db([])
# non-privileged user with no realm permission.
self.set_as_user()
# alive bot
_bot_event('bot_connected', bot_id='id1')
resp = self.post_prpc(api,
swarming_pb2.BotRequest(bot_id='id1'),
expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
# deleted bot
with mock.patch('server.acl._is_admin', return_value=True):
self.post_prpc('DeleteBot', swarming_pb2.BotRequest(bot_id='id1'))
resp = self.post_prpc(api,
swarming_pb2.BotRequest(bot_id='id1'),
expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
def test_delete_ok(self):
"""Assert that delete finds and deletes a bot."""
self.set_as_admin()
self.mock(acl, '_is_admin', lambda *_args, **_kwargs: True)
state = {
'dict': {
'random': 'values'
},
'float': 0.,
'list': ['of', 'things'],
'str': u'uni',
}
_bot_event('request_sleep', bot_id='id1', state=state)
# delete the bot
resp = self.post_prpc('DeleteBot', swarming_pb2.BotRequest(bot_id='id1'))
actual = swarming_pb2.DeleteResponse()
_decode(resp.body, actual)
self.assertEqual(swarming_pb2.DeleteResponse(deleted=True), actual)
# is it gone?
resp = self.post_prpc('DeleteBot',
swarming_pb2.BotRequest(bot_id='id1'),
expect_errors=True)
self.assertEqual(resp.status, '404 Not Found')
def test_events(self):
# Run one task, push an event manually.
second = datetime.timedelta(seconds=1)
self.mock(random, 'getrandbits', lambda _: 0x88)
first_ticker = test_case.Ticker(self.now, datetime.timedelta(seconds=1))
t1 = self.mock_now(first_ticker())
self.set_as_bot()
params = self.do_handshake()
t2 = self.mock_now(first_ticker())
self.bot_poll(params=params)
self.set_as_user()
self.client_create_task_raw()
self.set_as_bot()
t3 = self.mock_now(first_ticker())
res = self.bot_poll(params=params)
now_60 = first_ticker.last() + datetime.timedelta(seconds=60)
second_ticker = test_case.Ticker(now_60, datetime.timedelta(seconds=1))
t4 = self.mock_now(second_ticker())
resp = self.bot_complete_task(task_id=res['manifest']['task_id'])
self.assertEqual({u'must_stop': False, u'ok': True}, resp)
params['event'] = 'bot_rebooting'
params['message'] = 'for the best'
t5 = self.mock_now(second_ticker())
resp = self.post_json('/swarming/api/v1/bot/event', params)
self.assertEqual({}, resp)
start = first_ticker.first()
end = second_ticker.last()
self.set_as_privileged_user()
request = swarming_pb2.BotEventsRequest(bot_id="bot1", limit=200)
request.start.FromDatetime(start)
request.end.FromDatetime(end + second)
resp = self.post_prpc("ListBotEvents", request)
dimensions = [
{
u'key': u'id',
u'value': [u'bot1']
},
{
u'key': u'os',
u'value': [u'Amiga']
},
{
u'key': u'pool',
u'value': [u'default']
},
]
state_dict = {
'bot_group_cfg_version': 'default',
'running_time': 1234.,
'sleep_streak': 0,
'started_ts': 1410990411.111,
}
state = unicode(
json.dumps(state_dict, sort_keys=True, separators=(',', ':')))
state_dict.pop('bot_group_cfg_version')
state_no_cfg_ver = unicode(
json.dumps(state_dict, sort_keys=True, separators=(',', ':')))
items = [
{
u'authenticated_as': u'bot:whitelisted-ip',
u'dimensions': dimensions,
u'event_type': u'bot_rebooting',
u'external_ip': unicode(self.source_ip),
u'message': u'for the best',
u'quarantined': False,
u'state': state,
u'ts': fmtdate(t5),
u'version': unicode(self.bot_version),
},
{
u'authenticated_as': u'bot:whitelisted-ip',
u'dimensions': dimensions,
u'event_type': u'task_completed',
u'external_ip': unicode(self.source_ip),
u'quarantined': False,
u'state': state,
u'task_id': res['manifest']['task_id'],
u'ts': fmtdate(t4),
u'version': unicode(self.bot_version),
},
{
u'authenticated_as': u'bot:whitelisted-ip',
u'dimensions': dimensions,
u'event_type': u'request_task',
u'external_ip': unicode(self.source_ip),
u'quarantined': False,
u'state': state,
u'task_id': res['manifest']['task_id'],
u'ts': fmtdate(t3),
u'version': unicode(self.bot_version),
},
{
u'authenticated_as': u'bot:whitelisted-ip',
u'dimensions': dimensions,
u'event_type': u'request_sleep',
u'external_ip': unicode(self.source_ip),
u'quarantined': False,
u'state': state,
u'ts': fmtdate(t2),
u'version': unicode(self.bot_version),
},
{
u'authenticated_as': u'bot:whitelisted-ip',
u'dimensions': dimensions,
u'event_type': u'bot_connected',
u'external_ip': unicode(self.source_ip),
u'quarantined': False,
u'state': state_no_cfg_ver,
u'ts': fmtdate(t1),
u'version': u'123',
},
]
expected = swarming_pb2.BotEventsResponse()
_decode_dict(dict(items=items), expected)
actual = swarming_pb2.BotEventsResponse()
_decode(resp.body, actual)
# Now test with a subset.
start = second_ticker.first()
end = second_ticker.last()
request = swarming_pb2.BotEventsRequest(bot_id="bot1", limit=200)
request.start.FromDatetime(start)
request.end.FromDatetime(end + second)
_encode(request)
resp = self.post_prpc("ListBotEvents", request)
actual = swarming_pb2.BotEventsResponse()
_decode(resp.body, actual)
# actual.items[:] converts to a list so it can be compared with expected
# items
self.assertEqual(expected.items[:2], actual.items[:])
def test_terminate_admin(self):
self.set_as_bot()
self.bot_poll()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_admin()
request = swarming_pb2.TerminateRequest(bot_id="bot1")
resp = self.post_prpc('TerminateBot', request)
actual = swarming_pb2.TerminateResponse()
_decode(resp.body, actual)
expected = swarming_pb2.TerminateResponse(task_id='5cee488008810')
self.assertEqual(expected, actual)
def test_terminate_privileged_user(self):
self.set_as_bot()
self.bot_poll()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_privileged_user()
bot_id = "bot1"
reason = "hello\n\n\tworld"
request = swarming_pb2.TerminateRequest(bot_id=bot_id, reason=reason)
resp = self.post_prpc('TerminateBot', request)
actual = swarming_pb2.TerminateResponse()
_decode(resp.body, actual)
expected = swarming_pb2.TerminateResponse(task_id='5cee488008810')
self.assertEqual(expected, actual)
request_key, _ = task_pack.get_request_and_result_keys(actual.task_id)
request = request_key.get()
self.assertEqual("Terminate bot1: hello world", request.name)
def test_terminate_with_extremely_long_reason(self):
self.set_as_bot()
self.bot_poll()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_privileged_user()
bot_id = "bot1"
reason = "hello world" * 10000
request = swarming_pb2.TerminateRequest(bot_id=bot_id, reason=reason)
resp = self.post_prpc('TerminateBot', request, expect_errors=True)
self.assertEqual(resp.status, '400 Bad Request')
def test_terminate_user(self):
self.set_as_bot()
self.bot_poll()
self.mock(random, 'getrandbits', lambda _: 0x88)
# without realm permission.
self.set_as_user()
self.mock_auth_db([])
request = swarming_pb2.TerminateRequest(bot_id="bot1")
resp = self.post_prpc('TerminateBot', request, expect_errors=True)
expected = cgi.escape(
'user "user@example.com" does not have '
'permission "swarming.pools.terminateBot"',
quote=True)
self.assertEqual(expected, resp.body)
# give permission.
self.mock_auth_db([auth.Permission('swarming.pools.terminateBot')])
request = swarming_pb2.BotRequest(bot_id="bot1")
resp = self.post_prpc('TerminateBot', request)
actual = swarming_pb2.TerminateResponse()
_decode(resp.body, actual)
expected = swarming_pb2.TerminateResponse(task_id='5cee488008810')
self.assertEqual(expected, actual)
def test_terminate_via_rbe(self):
self.set_as_bot()
self.bot_poll()
self.mock(random, 'getrandbits', lambda _: 0x88)
def mocked_rbe_config(*_args):
return rbe.RBEBotConfig(
instance='some/rbe/instance',
hybrid_mode=False)
self.mock(rbe, 'get_rbe_config_for_bot', mocked_rbe_config)
self.set_as_admin()
request = swarming_pb2.TerminateRequest(bot_id="bot1")
resp = self.post_prpc('TerminateBot', request)
actual = swarming_pb2.TerminateResponse()
_decode(resp.body, actual)
expected = swarming_pb2.TerminateResponse(task_id='5cee488008810')
self.assertEqual(expected, actual)
key, _ = task_pack.get_request_and_result_keys(actual.task_id)
req = key.get()
self.assertEqual(req.rbe_instance, 'some/rbe/instance')
def test_tasks_ok(self):
"""Asserts that tasks produces bot information."""
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_bot()
self.bot_poll()
self.set_as_user()
self.client_create_task_raw()
self.set_as_bot()
res = self.bot_poll()
response = self.bot_complete_task(task_id=res['manifest']['task_id'])
self.assertEqual({u'must_stop': False, u'ok': True}, response)
now_1 = self.mock_now(self.now, 1)
self.mock(random, 'getrandbits', lambda _: 0x55)
self.set_as_user()
self.client_create_task_raw(name='philbert')
self.set_as_bot()
res = self.bot_poll()
response = self.bot_complete_task(exit_code=1,
task_id=res['manifest']['task_id'])
self.assertEqual({u'must_stop': False, u'ok': True}, response)
start = self.now + datetime.timedelta(seconds=0.5)
end = now_1 + datetime.timedelta(seconds=0.5)
self.set_as_privileged_user()
request = swarming_pb2.BotTasksRequest()
request.bot_id = 'bot1'
request.start.FromDatetime(start)
request.end.FromDatetime(end)
request.sort = swarming_pb2.QUERY_CREATED_TS
request.state = swarming_pb2.QUERY_ALL
request.include_performance_stats = True
request.limit = 100
response = self.post_prpc('ListBotTasks', request)
actual = swarming_pb2.TaskListResponse()
_decode(response.body, actual)
expected = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_run_result(expected)
expected.bot_idle_since_ts.FromDatetime(now_1)
expected.completed_ts.FromDatetime(now_1)
expected.modified_ts.FromDatetime(now_1)
expected.started_ts.FromDatetime(now_1)
expected.created_ts.FromDatetime(now_1)
expected.duration = 0.1
expected.name = 'philbert'
expected.exit_code = 1
expected.failure = True
expected.run_id = '5cee870005511'
expected.state = swarming_pb2.TaskState.COMPLETED
expected.task_id = '5cee870005511'
expected.costs_usd[:] = [0.1]
self.gen_perf_stats_prpc(expected.performance_stats)
self.assertEqual(expected, actual.items[0])
def test_all_task_sorts(self):
self.set_as_privileged_user()
for sort in (swarming_pb2.QUERY_CREATED_TS, swarming_pb2.QUERY_STARTED_TS,
swarming_pb2.QUERY_COMPLETED_TS):
request = swarming_pb2.BotTasksRequest()
request.bot_id = 'bot1'
request.sort = sort
request.state = swarming_pb2.QUERY_ALL
request.include_performance_stats = True
request.limit = 100
response = self.post_prpc('ListBotTasks', request)
self.assertEqual(response.status, '200 OK')
def test_abandoned_has_error(self):
"""abandoned_ts does not have an index with bot_id. So we test that it
specifically will return a 400 bad request.
"""
self.set_as_privileged_user()
request = swarming_pb2.BotTasksRequest()
request.bot_id = 'bot1'
request.sort = swarming_pb2.QUERY_ABANDONED_TS
request.state = swarming_pb2.QUERY_ALL
request.include_performance_stats = True
request.limit = 100
response = self.post_prpc('ListBotTasks', request, expect_errors=True)
self.assertEqual(response.status, '400 Bad Request')
def test_all_task_state_filters(self):
self.set_as_privileged_user()
for state in swarming_pb2.StateQuery.DESCRIPTOR.values_by_number.keys():
# state cannot be pending since it is a disallowed state for
# a TaskRunResult. See task_result._validate_state_not_pending
if state not in (swarming_pb2.QUERY_PENDING,
swarming_pb2.QUERY_PENDING_RUNNING,
swarming_pb2.QUERY_DEDUPED):
request = swarming_pb2.BotTasksRequest()
request.bot_id = 'bot1'
request.state = state
request.include_performance_stats = True
request.limit = 100
response = self.post_prpc('ListBotTasks', request)
self.assertEqual(response.status, '200 OK')
def test_all_task_state_filters_disallowed_states(self):
self.set_as_privileged_user()
# State cannot be any of these states, since for `bot.list_tasks`
# a TaskRunResult. See task_result._DISALLOWED_STATES
for state in (swarming_pb2.QUERY_PENDING,
swarming_pb2.QUERY_PENDING_RUNNING,
swarming_pb2.QUERY_DEDUPED):
request = swarming_pb2.BotTasksRequest()
request.bot_id = 'bot1'
request.state = state
request.include_performance_stats = True
request.limit = 100
response = self.post_prpc('ListBotTasks', request, expect_errors=True)
self.assertEqual(response.status, '400 Bad Request')
def test_dimensions_not_found(self):
self.set_as_privileged_user()
request = swarming_pb2.BotsDimensionsRequest(pool='unknown')
response = self.post_prpc('GetBotDimensions', request, expect_errors=True)
self.assertEqual(response.status, '404 Not Found')
def test_dimensions_forbidden(self):
self.set_as_user()
self.mock_auth_db([])
# the user doesn't have permission to get dimensions.
request = swarming_pb2.BotsDimensionsRequest()
response = self.post_prpc('GetBotDimensions', request, expect_errors=True)
self.assertEqual(response.status, '403 Forbidden')
request = swarming_pb2.BotsDimensionsRequest(pool='default')
response = self.post_prpc('GetBotDimensions', request, expect_errors=True)
self.assertEqual(response.status, '403 Forbidden')
# pool permission isn't sufficient to get all dimensions.
self.mock_auth_db([
auth.Permission('swarming.pools.listBots'),
])
request = swarming_pb2.BotsDimensionsRequest()
response = self.post_prpc('GetBotDimensions', request, expect_errors=True)
self.assertEqual(response.status, '403 Forbidden')
def test_dimensions_pool(self):
self.set_as_user()
self.mock_auth_db([
auth.Permission('swarming.pools.listBots'),
])
bot_management.DimensionAggregation(
key=bot_management.get_aggregation_key('default'),
dimensions=[
bot_management.DimensionValues(dimension='foo',
values=['alpha', 'beta']),
],
ts=self.now).put()
expected = swarming_pb2.BotsDimensions(
bots_dimensions=[
swarming_pb2.StringListPair(key='foo', value=['alpha', 'beta']),
],
ts=message_conversion_prpc.date(self.now),
)
request = swarming_pb2.BotsDimensionsRequest(pool='default')
response = self.post_prpc('GetBotDimensions', request)
actual = swarming_pb2.BotsDimensions()
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_dimensions_all(self):
"""Asserts that BotsDimensions is returned with the right data."""
self.set_as_privileged_user()
bot_management.DimensionAggregation(
key=bot_management.DimensionAggregation.KEY,
dimensions=[
bot_management.DimensionValues(dimension='foo',
values=['alpha', 'beta']),
bot_management.DimensionValues(dimension='bar',
values=['gamma', 'delta',
'epsilon']),
],
ts=self.now).put()
expected = swarming_pb2.BotsDimensions(
bots_dimensions=[
swarming_pb2.StringListPair(key='foo', value=['alpha', 'beta']),
swarming_pb2.StringListPair(key='bar',
value=['gamma', 'delta', 'epsilon']),
],
ts=message_conversion_prpc.date(self.now),
)
request = swarming_pb2.BotsDimensionsRequest()
response = self.post_prpc('GetBotDimensions', request)
actual = swarming_pb2.BotsDimensions()
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_list_ok(self):
"""Asserts that BotInfo is returned for the appropriate set of bots."""
self.set_as_privileged_user()
then = datetime.datetime(2009, 1, 2, 3, 4, 5)
self.mock_now(then)
# Add four bot events, corresponding to one dead bot, one quarantined bot,
# one bot in maintenance, and one good bot
_bot_event('request_sleep', bot_id='id3')
self.mock_now(self.now)
_bot_event('request_sleep', bot_id='id1')
_bot_event('request_sleep', bot_id='id2', quarantined=True)
_bot_event('request_sleep', bot_id='id4', maintenance_msg='very busy')
def _default_bot(bot_id, the_date):
out = swarming_pb2.BotInfo(
authenticated_as='bot:whitelisted-ip',
bot_id=bot_id,
deleted=False,
dimensions=[
swarming_pb2.StringListPair(
key='id',
value=[bot_id],
),
swarming_pb2.StringListPair(
key='pool',
value=['default'],
)
],
external_ip='8.8.4.4',
first_seen_ts=message_conversion_prpc.date(the_date),
is_dead=False,
last_seen_ts=message_conversion_prpc.date(the_date),
quarantined=False,
state='{"ram":65}',
version='123456789')
return out
# setup expected bots
bot1 = _default_bot('id1', self.now)
bot2 = _default_bot('id2', self.now)
bot2.quarantined = True
bot3 = _default_bot('id3', then)
bot4 = _default_bot('id4', self.now)
bot4.maintenance_msg = 'very busy'
expected = swarming_pb2.BotInfoListResponse(
items=[bot1, bot2, bot3, bot4],
death_timeout=config.settings().bot_death_timeout_secs,
now=message_conversion_prpc.date(self.now))
def _verify(items,
quarantined=swarming_pb2.NULL,
in_maintenance=swarming_pb2.NULL,
is_dead=swarming_pb2.NULL,
is_busy=swarming_pb2.NULL,
dimensions=None):
if dimensions is None:
dimensions = []
request = swarming_pb2.BotsRequest(limit=100,
quarantined=quarantined,
in_maintenance=in_maintenance,
is_dead=is_dead,
is_busy=is_busy,
dimensions=dimensions)
response = self.post_prpc('ListBots', request)
actual = swarming_pb2.BotInfoListResponse()
_decode(response.body, actual)
expected.ClearField('items')
expected.items.extend(items)
self.assertEqual(expected.items, actual.items)
self.assertEqual(expected.death_timeout, actual.death_timeout)
_verify([bot1, bot2, bot3, bot4])
# This will mark bot3 as dead since it was created before self.now
self.assertEqual(1, bot_management.cron_update_bot_info())
bot3.is_dead = True
expected.items[2].is_dead = True
_verify([bot1, bot2, bot3, bot4])
# All bots should be returned if we don't care about quarantined
_verify([bot1, bot2, bot3, bot4], quarantined=swarming_pb2.NULL)
# All bots should be returned if we don't care about is_dead
_verify([bot1, bot2, bot3, bot4], is_dead=swarming_pb2.NULL)
# Only bot1 corresponds to these two dimensions
_verify(items=[bot1],
dimensions=[
swarming_pb2.StringPair(key='pool', value='default'),
swarming_pb2.StringPair(key='id', value='id1'),
])
# Only bot1 corresponds to being not dead and not quarantined and
# not in maintenance and this dimension
_verify(items=[bot1],
dimensions=[swarming_pb2.StringPair(key='pool', value='default')],
quarantined=swarming_pb2.FALSE,
in_maintenance=swarming_pb2.FALSE,
is_dead=swarming_pb2.FALSE)
# exclude bot2 only, which is quarantined
_verify(items=[bot1, bot3, bot4], quarantined=swarming_pb2.FALSE)
# exclude bot3 only, which is dead
_verify(items=[bot1, bot2, bot4], is_dead=swarming_pb2.FALSE)
# only bot2 is quarantined
_verify(items=[bot2], quarantined=swarming_pb2.TRUE)
# only bot4 is in maintenance
_verify(items=[bot4], in_maintenance=swarming_pb2.TRUE)
# quarantined:true can be paired with other dimensions and still work
_verify(items=[bot2],
quarantined=swarming_pb2.TRUE,
dimensions=[swarming_pb2.StringPair(key='pool', value='default')])
# in_maintenance:true can be paired with other dimensions and still work
_verify(items=[bot4],
in_maintenance=swarming_pb2.TRUE,
dimensions=[swarming_pb2.StringPair(key='pool', value='default')])
# only bot3 is dead
_verify(items=[bot3], is_dead=swarming_pb2.TRUE)
# is_dead:true can be paired with other dimensions and still work
_verify(items=[bot3],
is_dead=swarming_pb2.TRUE,
dimensions=[swarming_pb2.StringPair(key='pool', value='default')])
# only 1 bot is "ready for work"
_verify(items=[bot1],
is_busy=swarming_pb2.FALSE,
is_dead=swarming_pb2.FALSE,
quarantined=swarming_pb2.FALSE)
# not:existing is a dimension that doesn't exist, nothing returned.
_verify(items=[],
dimensions=[swarming_pb2.StringPair(key='not', value='existing')])
# quarantined:true can be paired with other non-existing dimensions and
# still work
_verify(items=[],
quarantined=swarming_pb2.TRUE,
dimensions=[swarming_pb2.StringPair(key='not', value='existing')])
# in_maintenance:true can be paired with other non-existing dimensions and
# still work
_verify(items=[],
in_maintenance=swarming_pb2.TRUE,
dimensions=[swarming_pb2.StringPair(key='not', value='existing')])
# is_dead:true can be paired with other non-existing dimensions and
# still work
_verify(items=[],
is_dead=swarming_pb2.TRUE,
dimensions=[swarming_pb2.StringPair(key='not', value='existing')])
# No bot is both dead and quarantined
_verify(items=[], is_dead=swarming_pb2.TRUE, quarantined=swarming_pb2.TRUE)
# OR dimension finds bot1 and bot2
_verify(items=[bot1, bot2],
dimensions=[swarming_pb2.StringPair(key='id', value='id1|id2')])
def test_count_ok(self):
self.set_as_privileged_user()
then = datetime.datetime(2009, 1, 2, 3, 4, 5)
self.mock_now(then)
_bot_event('request_sleep', bot_id='id3', quarantined=True)
self.mock_now(self.now)
_bot_event('request_task', bot_id='id1', task_id='987')
_bot_event('request_sleep', bot_id='id2', quarantined=True)
_bot_event('request_sleep', bot_id='id4', maintenance_msg='very busy')
def _verify(dimensions,
count=0,
quarantined=0,
maintenance=0,
dead=0,
busy=0):
request = swarming_pb2.BotsCountRequest(dimensions=dimensions)
resp = self.post_prpc('CountBots', request)
actual = swarming_pb2.BotsCount()
_decode(resp.body, actual)
self.assertEqual(count, actual.count)
self.assertEqual(quarantined, actual.quarantined)
self.assertEqual(maintenance, actual.maintenance)
self.assertEqual(dead, actual.dead)
self.assertEqual(busy, actual.busy)
_verify(dimensions=[],
count=4,
quarantined=2,
maintenance=1,
dead=0,
busy=4)
self.assertEqual(1, bot_management.cron_update_bot_info())
_verify(dimensions=[],
count=4,
quarantined=2,
maintenance=1,
dead=1,
busy=4)
_verify(dimensions=[
swarming_pb2.StringPair(key='pool', value='default'),
swarming_pb2.StringPair(key='id', value='id1')
],
count=1,
busy=1)
_verify(dimensions=[swarming_pb2.StringPair(key='id', value='id1|id2')],
count=2,
quarantined=1,
maintenance=0,
dead=0,
busy=2)
_verify(dimensions=[
swarming_pb2.StringPair(key='pool', value='default'),
swarming_pb2.StringPair(key='id', value='id3')
],
count=1,
quarantined=1,
dead=1,
busy=1)
_verify(dimensions=[
swarming_pb2.StringPair(key='pool', value='default'),
swarming_pb2.StringPair(key='id', value='id4')
],
busy=1,
maintenance=1,
count=1)
_verify(dimensions=[swarming_pb2.StringPair(key='non', value='existing')])
def test_count_bad_request(self):
self.set_as_privileged_user()
request = swarming_pb2.BotsCountRequest(
dimensions=[swarming_pb2.StringPair(key='a', value='')])
resp = self.post_prpc('CountBots', request, expect_errors=True)
self.assertEqual(resp.status, '400 Bad Request')
@parameterized.expand([
('ListBots', swarming_pb2.BotsRequest(limit=10)),
('CountBots', swarming_pb2.BotsCountRequest()),
])
def test_ok_realm(self, rpc, request):
# non-privileged user, with realm permission.
self.set_as_user()
self.mock_auth_db([auth.Permission('swarming.pools.listBots')])
request.dimensions.extend(
[swarming_pb2.StringPair(key='pool', value='default')])
self.post_prpc(rpc, request)
@parameterized.expand([
('ListBots', swarming_pb2.BotsRequest(limit=10)),
('CountBots', swarming_pb2.BotsCountRequest()),
])
def test_forbidden_realm(self, rpc, request):
# non-privileged user, with no permissions.
self.mock_auth_db([])
self.set_as_user()
# the user needs to specify a pool dimension.
resp = self.post_prpc(rpc, request, expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
# the user needs to have permissions of the requested pools.
request.dimensions.extend(
[swarming_pb2.StringPair(key='pool', value='default')])
response = self.post_prpc(rpc, request, expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
expected = cgi.escape(
'user "user@example.com" does not have permission '
'"swarming.pools.listBots"',
quote=True)
self.assertEqual(expected, response.body)
# the user needs to specify known pools.
request.ClearField('dimensions')
request.dimensions.extend(
[swarming_pb2.StringPair(key='pool', value='unknown')])
response = self.post_prpc(rpc, request, expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
expected = cgi.escape('No such pool or no permission to use it: unknown',
quote=True)
self.assertEqual(expected, response.body)
class TaskServicePrpcTest(PrpcTest):
def setUp(self):
super(TaskServicePrpcTest, self).setUp()
self.service = "swarming.v2.Tasks"
self.now = datetime.datetime(2010, 1, 2, 3, 4, 5)
self.mock_now(self.now)
self.mock_default_pool_acl([])
self.mock_tq_tasks()
self.mock(service_accounts, 'has_token_server', lambda: True)
def test_result_unknown(self):
"""Asserts that result raises 404 for unknown task IDs."""
self.set_as_privileged_user()
resp = self.post_prpc('GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id='12310'),
expect_errors=True)
self.assertEqual(resp.status, '404 Not Found')
def test_result_long(self):
"""Asserts that result raises 400 for wildly invalid task IDs."""
self.set_as_privileged_user()
resp = self.post_prpc('GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id='12310' *
10),
expect_errors=True)
self.assertEqual(resp.status, '400 Bad Request')
def test_result_ok(self):
"""Asserts that result produces a result entity."""
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_bot()
self.bot_poll()
# pending task
self.set_as_user()
_, task_id = self.client_create_task_raw()
response = self.post_prpc(
'GetResult', swarming_pb2.TaskIdWithPerfRequest(task_id=task_id))
expected = swarming_pb2.TaskResultResponse()
expected.created_ts.FromDatetime(self.now)
expected.modified_ts.FromDatetime(self.now)
expected.failure = False
expected.internal_failure = False
expected.name = 'job1'
expected.server_versions[:] = ['v1a']
expected.state = swarming_pb2.TaskState.PENDING
expected.tags[:] = [
u'a:tag',
u'authenticated:user:user@example.com',
u'os:Amiga',
u'pool:default',
u'priority:20',
u'realm:none',
u'service_account:none',
u'swarming.pool.template:none',
u'swarming.pool.version:pools_cfg_rev',
u'user:joe@localhost',
]
expected.task_id = '5cee488008810'
expected.user = 'joe@localhost'
actual = swarming_pb2.TaskResultResponse()
_decode(response.body, actual)
self.assertEqual(expected, actual)
# no bot started: running task
run_id = task_id[:-1] + '1'
response = self.post_prpc(
'GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id=run_id),
expect_errors=True)
self.assertEqual(response.status, '404 Not Found')
# run as bot
self.set_as_bot()
self.bot_poll()
self.set_as_user()
response = self.post_prpc(
'GetResult', swarming_pb2.TaskIdWithPerfRequest(task_id=run_id))
expected = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_run_result(expected)
expected.bot_idle_since_ts.FromDatetime(self.now)
expected.created_ts.FromDatetime(self.now)
expected.modified_ts.FromDatetime(self.now)
expected.started_ts.FromDatetime(self.now)
expected.current_task_slice = 0
actual = swarming_pb2.TaskResultResponse()
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_result_completed_task(self):
"""Tests that completed tasks are correctly reported."""
self.set_as_bot()
self.bot_poll()
self.set_as_user()
self.client_create_task_raw()
self.set_as_bot()
task_id = self.bot_run_task()
# First ask without perf metadata.
self.set_as_user()
response = self.post_prpc(
'GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id=task_id,
include_performance_stats=True))
expected = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_run_result(expected)
self.gen_perf_stats_prpc(expected.performance_stats)
expected.bot_idle_since_ts.FromDatetime(self.now)
expected.completed_ts.FromDatetime(self.now)
expected.modified_ts.FromDatetime(self.now)
expected.created_ts.FromDatetime(self.now)
expected.started_ts.FromDatetime(self.now)
expected.state = swarming_pb2.COMPLETED
expected.costs_usd[:] = [0.1]
expected.run_id = task_id
expected.task_id = task_id
expected.duration = 0.1
actual = swarming_pb2.TaskResultResponse()
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_request_unknown(self):
"""Asserts that 404 is raised for unknown tasks."""
self.set_as_user()
response = self.post_prpc('GetRequest',
swarming_pb2.TaskIdRequest(task_id='12310'),
expect_errors=True)
self.assertEqual(response.status, '404 Not Found')
def test_request_ok(self):
"""Asserts that request produces a task request."""
self.set_as_user()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.mock_default_pool_acl(['service-account@example.com'])
self.mock_auth_db([
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
])
_, task_id = self.client_create_task_raw(
properties={'secret_bytes': 'zekret'},
service_account='service-account@example.com',
realm='test:task_realm')
expected_props = swarming_pb2.TaskProperties()
apply_default_for_task_props(expected_props)
expected_props.secret_bytes = b'<REDACTED>'
expected_props.command[:] = ['python', 'run_test.py']
expected = swarming_pb2.TaskRequestResponse(
properties=expected_props,
task_slices=[
swarming_pb2.TaskSlice(expiration_secs=86400,
properties=expected_props,
wait_for_capacity=False)
])
apply_defaults_for_request(expected)
expected.created_ts.FromDatetime(self.now)
expected.service_account = 'service-account@example.com'
expected.bot_ping_tolerance_secs = 600
expected.expiration_secs = 86400
expected.realm = 'test:task_realm'
expected.tags[:] = [
u'a:tag',
u'authenticated:user:user@example.com',
u'os:Amiga',
u'pool:default',
u'priority:20',
u'realm:test:task_realm',
u'service_account:service-account@example.com',
u'swarming.pool.template:none',
u'swarming.pool.version:pools_cfg_rev',
u'user:joe@localhost',
]
response = self.post_prpc('GetRequest',
swarming_pb2.TaskIdRequest(task_id=task_id))
actual = swarming_pb2.TaskRequestResponse()
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_cancel_pending(self):
"""Asserts that task cancellation goes smoothly."""
# catch PubSub notification
# Create and cancel a task as a non-privileged user.
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_bot()
self.bot_poll()
self.set_as_user()
_, task_id = self.client_create_task_raw(
pubsub_topic='projects/abc/topics/def', pubsub_userdata='blah')
expected = swarming_pb2.CancelResponse(canceled=True, was_running=False)
response = self.post_prpc(
'CancelTask',
swarming_pb2.TaskCancelRequest(task_id=task_id, kill_running=False))
actual = swarming_pb2.CancelResponse()
_decode(response.body, actual)
self.assertEqual(expected, actual)
# determine that the task's state updates correctly
expected = swarming_pb2.TaskResultResponse(
abandoned_ts=message_conversion_prpc.date(self.now),
completed_ts=message_conversion_prpc.date(self.now),
created_ts=message_conversion_prpc.date(self.now),
current_task_slice=0,
failure=False,
internal_failure=False,
modified_ts=message_conversion_prpc.date(self.now),
name='job1',
server_versions=['v1a'],
state='CANCELED',
tags=[
'a:tag',
'authenticated:user:user@example.com',
'os:Amiga',
'pool:default',
'priority:20',
'realm:none',
'service_account:none',
'swarming.pool.template:none',
'swarming.pool.version:pools_cfg_rev',
'user:joe@localhost',
],
task_id=task_id,
user=u'joe@localhost',
)
response = self.post_prpc(
'GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id=task_id,
include_performance_stats=False))
actual = swarming_pb2.TaskResultResponse()
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_cancel_forbidden(self):
"""Asserts that non-privileged non-owner can't cancel tasks."""
# Create a task as an admin.
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_admin()
_, task_id = self.client_create_task_raw(
pubsub_topic='projects/abc/topics/def', pubsub_userdata='blah')
# Attempt to cancel as non-privileged user -> HTTP 403.
self.set_as_user()
self.mock_auth_db([])
response = self.post_prpc('CancelTask',
swarming_pb2.TaskCancelRequest(
task_id=task_id, kill_running=False),
expect_errors=True)
self.assertEqual(response.status, '403 Forbidden')
def test_cancel_with_realm_permission(self):
# someone creates tasks with/without realm.
self.set_as_privileged_user()
self.mock_auth_db([
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
])
_, task_id_with_realm = self.client_create_task_raw(realm='test:task_realm')
_, task_id_without_realm = self.client_create_task_raw(realm=None)
def assertTaskIsNotAccessible(task_id):
response = self.post_prpc('CancelTask',
swarming_pb2.TaskCancelRequest(
task_id=task_id, kill_running=False),
expect_errors=True)
self.assertEqual(response.status, '403 Forbidden')
self.assertEqual(
cgi.escape('Task "%s" is not accessible' % task_id, quote=True),
response.body)
# non-privileged user can't cancel the both tasks without permission.
self.set_as_user()
assertTaskIsNotAccessible(task_id_with_realm)
assertTaskIsNotAccessible(task_id_without_realm)
# the user can cancel to the both tasks with swarming.pools.cancelTask
# permission.
self.mock_auth_db([auth.Permission('swarming.pools.cancelTask')])
self.post_prpc(
'CancelTask',
swarming_pb2.TaskCancelRequest(task_id=task_id_with_realm,
kill_running=False))
self.post_prpc(
'CancelTask',
swarming_pb2.TaskCancelRequest(task_id=task_id_without_realm,
kill_running=False))
# the user can cancel with swarming.tasks.cancel permission.
self.mock_auth_db([auth.Permission('swarming.tasks.cancel')])
self.post_prpc(
'CancelTask',
swarming_pb2.TaskCancelRequest(task_id=task_id_with_realm,
kill_running=False))
# but, not accessible to the task without realm.
assertTaskIsNotAccessible(task_id_without_realm)
def test_cancel_running(self):
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_bot()
self.bot_poll()
self.set_as_user()
_, task_id = self.client_create_task_raw(properties=dict(
command=['python', 'runtest.py']))
self.set_as_bot()
params = self.do_handshake()
data = self.post_json('/swarming/api/v1/bot/poll', params)
run_id = data['manifest']['task_id']
def _params(**kwargs):
out = {
'cost_usd': 0.1,
'duration': None,
'exit_code': None,
'id': 'bot1',
'output': None,
'output_chunk_start': 0,
'task_id': run_id,
}
out.update(**kwargs)
return out
self.set_as_bot()
params = _params(output=base64.b64encode('Oh '))
response = self.post_json('/swarming/api/v1/bot/task_update', params)
self.assertEqual({u'must_stop': False, u'ok': True}, response)
self.set_as_user()
expected = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_result_summary(expected)
expected.bot_idle_since_ts.FromDatetime(self.now)
expected.costs_usd[:] = [0.1]
expected.created_ts.FromDatetime(self.now)
expected.modified_ts.FromDatetime(self.now)
expected.started_ts.FromDatetime(self.now)
expected.state = swarming_pb2.TaskState.RUNNING
expected.tags[:] = [
u'a:tag', u'authenticated:user:user@example.com', u'os:Amiga',
u'pool:default', u'priority:20', u'realm:none', u'service_account:none',
u'swarming.pool.template:none', u'swarming.pool.version:pools_cfg_rev',
u'user:joe@localhost'
]
actual = swarming_pb2.TaskResultResponse()
response = self.post_prpc(
'GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id=task_id,
include_performance_stats=False))
_decode(response.body, actual)
self.assertEqual(expected, actual)
# Denied if kill_running == False.
response = self.post_prpc('CancelTask',
swarming_pb2.TaskCancelRequest(
task_id=task_id, kill_running=False),
expect_errors=True)
actual = swarming_pb2.CancelResponse()
_decode(response.body, actual)
self.assertEqual(
swarming_pb2.CancelResponse(canceled=False, was_running=True), actual)
# Works if kill_running == True.
response = self.post_prpc('CancelTask',
swarming_pb2.TaskCancelRequest(task_id=task_id,
kill_running=True),
expect_errors=True)
actual = swarming_pb2.CancelResponse()
_decode(response.body, actual)
self.assertEqual(
swarming_pb2.CancelResponse(canceled=True, was_running=True), actual)
self.set_as_bot()
params = _params(output=base64.b64encode('hi'), output_chunk_start=3)
response = self.post_json('/swarming/api/v1/bot/task_update', params)
self.assertEqual({u'must_stop': True, u'ok': True}, response)
# abandoned_ts is set but state isn't changed yet.
self.set_as_user()
expected = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_result_summary(expected)
expected.abandoned_ts.FromDatetime(self.now)
expected.bot_idle_since_ts.FromDatetime(self.now)
expected.costs_usd[:] = [0.1]
expected.created_ts.FromDatetime(self.now)
expected.modified_ts.FromDatetime(self.now)
expected.started_ts.FromDatetime(self.now)
expected.state = swarming_pb2.TaskState.RUNNING
expected.tags[:] = [
'a:tag',
'authenticated:user:user@example.com',
'os:Amiga',
'pool:default',
'priority:20',
'realm:none',
'service_account:none',
'swarming.pool.template:none',
'swarming.pool.version:pools_cfg_rev',
'user:joe@localhost',
]
actual = swarming_pb2.TaskResultResponse()
response = self.post_prpc(
'GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id=task_id,
include_performance_stats=False))
_decode(response.body, actual)
self.assertEqual(expected, actual)
# Bot terminates the task.
self.set_as_bot()
params = _params(output=base64.b64encode(' again'),
output_chunk_start=6,
duration=0.1,
exit_code=0)
response = self.post_json('/swarming/api/v1/bot/task_update', params)
self.assertEqual({u'must_stop': True, u'ok': True}, response)
self.set_as_user()
expected = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_result_summary(expected)
expected.abandoned_ts.FromDatetime(self.now)
expected.bot_idle_since_ts.FromDatetime(self.now)
expected.completed_ts.FromDatetime(self.now)
expected.costs_usd[:] = [0.1]
expected.created_ts.FromDatetime(self.now)
expected.duration = 0.1
expected.exit_code = 0
expected.modified_ts.FromDatetime(self.now)
expected.started_ts.FromDatetime(self.now)
expected.state = swarming_pb2.TaskState.KILLED
expected.tags[:] = [
u'a:tag',
u'authenticated:user:user@example.com',
u'os:Amiga',
u'pool:default',
u'priority:20',
u'realm:none',
u'service_account:none',
u'swarming.pool.template:none',
u'swarming.pool.version:pools_cfg_rev',
u'user:joe@localhost',
]
actual = swarming_pb2.TaskResultResponse()
response = self.post_prpc(
'GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id=task_id,
include_performance_stats=False))
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_cancel_with_too_long_key(self):
"""Asserts that result raises 400 for wildly invalid task IDs."""
self.set_as_privileged_user()
resp = self.post_prpc('CancelTask',
swarming_pb2.TaskCancelRequest(task_id='12310' * 10,
kill_running=False),
expect_errors=True)
self.assertEqual(resp.status, '400 Bad Request')
def test_stdout_ok(self):
"""Asserts that stdout reports a task's output."""
self.set_as_bot()
self.bot_poll()
self.set_as_user()
self.client_create_task_raw()
# task_id determined by bot run
self.set_as_bot()
task_id = self.bot_run_task()
self.set_as_privileged_user()
run_id = task_id[:-1] + '1'
expected = swarming_pb2.TaskOutputResponse(
output=u'rÉsult string'.encode('utf-8'),
state=swarming_pb2.TaskState.COMPLETED)
for i in (task_id, run_id):
response = self.post_prpc('GetStdout',
swarming_pb2.TaskIdWithOffsetRequest(task_id=i))
actual = swarming_pb2.TaskOutputResponse()
_decode(response.body, actual)
self.assertEqual(expected, actual)
# Partial fetch.
response = self.post_prpc(
'GetStdout',
swarming_pb2.TaskIdWithOffsetRequest(task_id=task_id,
offset=1,
length=2))
actual = swarming_pb2.TaskOutputResponse()
_decode(response.body, actual)
# This is because it's counting in bytes, not in unicode characters:
expected = swarming_pb2.TaskOutputResponse(
output=u'É'.encode('utf-8'), state=swarming_pb2.TaskState.COMPLETED)
self.assertEqual(expected, actual)
response = self.post_prpc(
'GetStdout',
swarming_pb2.TaskIdWithOffsetRequest(task_id=task_id,
offset=3,
length=5))
actual = swarming_pb2.TaskOutputResponse()
_decode(response.body, actual)
expected = swarming_pb2.TaskOutputResponse(
output=u'sult '.encode('utf-8'), state=swarming_pb2.TaskState.COMPLETED)
self.assertEqual(expected, actual)
def test_stdout_empty(self):
"""Asserts that incipient tasks produce no output."""
self.set_as_user()
_, task_id = self.client_create_task_raw()
response = self.post_prpc(
'GetStdout', swarming_pb2.TaskIdWithOffsetRequest(task_id=task_id))
actual = swarming_pb2.TaskOutputResponse()
_decode(response.body, actual)
expected = swarming_pb2.TaskOutputResponse(
state=swarming_pb2.TaskState.NO_RESOURCE)
self.assertEqual(expected, actual)
def test_result_run_not_found(self):
self.set_as_user()
_, task_id = self.client_create_task_raw()
run_id = task_id[:-1] + '1'
response = self.post_prpc(
'GetStdout',
swarming_pb2.TaskIdWithOffsetRequest(task_id=run_id),
expect_errors=True)
self.assertEqual(response.status, '404 Not Found')
def test_task_deduped(self):
"""Asserts that task deduplication works as expected."""
self.set_as_user()
self.set_as_bot()
self.bot_poll()
self.set_as_user()
_, task_id_1 = self.client_create_task_raw(properties=dict(idempotent=True))
self.set_as_bot()
task_id_bot = self.bot_run_task()
self.assertEqual(task_id_1, task_id_bot[:-1] + '0')
self.assertEqual('1', task_id_bot[-1:])
# second task; this one's results should be returned immediately
self.set_as_user()
_, task_id_2 = self.client_create_task_raw(name='second',
user='jack@localhost',
properties=dict(idempotent=True))
self.set_as_bot()
resp = self.bot_poll()
self.assertEqual('sleep', resp['cmd'])
self.set_as_user()
# results shouldn't change, even if the second task wasn't executed
response = self.post_prpc(
'GetStdout', swarming_pb2.TaskIdWithOffsetRequest(task_id=task_id_2))
actual = swarming_pb2.TaskOutputResponse()
_decode(response.body, actual)
expected = swarming_pb2.TaskOutputResponse(
output=u'rÉsult string'.encode('utf-8'),
state=swarming_pb2.TaskState.COMPLETED)
self.assertEqual(expected, actual)
@parameterized.expand(['GetRequest', 'GetResult', 'GetStdout'])
def test_get_with_realm_permission(self, api):
# someone creates tasks with/without realm.
self.set_as_privileged_user()
self.mock_auth_db([
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
])
_, task_id_with_realm = self.client_create_task_raw(realm='test:task_realm')
_, task_id_without_realm = self.client_create_task_raw(realm=None)
def create_request(task_id):
if api == 'GetRequest':
return swarming_pb2.TaskIdRequest(task_id=task_id)
if api == 'GetResult':
return swarming_pb2.TaskIdWithPerfRequest(task_id=task_id)
if api == 'GetStdout':
return swarming_pb2.TaskIdWithOffsetRequest(task_id=task_id)
raise Exception('Unknown api %s' % api)
def assertTaskIsNotAccessible(task_id):
request = create_request(task_id)
response = self.post_prpc(api, request, expect_errors=True)
self.assertEqual(
cgi.escape('Task "%s" is not accessible' % task_id, quote=True),
response.body)
# non-privileged user can't access to the both tasks without permission.
self.set_as_user()
assertTaskIsNotAccessible(task_id_with_realm)
assertTaskIsNotAccessible(task_id_without_realm)
# the user can access to the both tasks with swarming.pools.listTasks
# permission.
self.mock_auth_db([auth.Permission('swarming.pools.listTasks')])
self.post_prpc(api, create_request(task_id_with_realm))
self.post_prpc(api, create_request(task_id_without_realm))
# the user can access with swarming.tasks.get permission.
self.mock_auth_db([auth.Permission('swarming.tasks.get')])
self.post_prpc(api, create_request(task_id_with_realm))
# but, not accessible to the task with no realm.
assertTaskIsNotAccessible(task_id_without_realm)
def test_new_ok(self):
# It can create a new task.
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_privileged_user()
ntr = self._new_task_request_prpc(use_default_slice=True)
response = self.post_prpc('NewTask', ntr)
actual = swarming_pb2.TaskRequestMetadataResponse()
_decode(response.body, actual)
self.assertEqual(u'5cee488008810', actual.task_id)
def test_new_task_performance_stats_are_empty(self):
# first create a new task.
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_privileged_user()
ntr = self._new_task_request_prpc(use_default_slice=True)
response = self.post_prpc('NewTask', ntr)
actual = swarming_pb2.TaskRequestMetadataResponse()
_decode(response.body, actual)
# A task which has yet to run will have performance stats which
# are empty.
response = self.post_prpc(
'GetResult',
swarming_pb2.TaskIdWithPerfRequest(task_id=actual.task_id,
include_performance_stats=True))
actual = swarming_pb2.TaskResultResponse()
_decode(response.body, actual)
self.assertEqual(swarming_pb2.PENDING, actual.state)
self.assertFalse(actual.HasField('performance_stats'))
def fail_on_using_legacy_acls(self):
def err(*_args, **_kwargs):
raise AssertionError('Must not be called')
self.mock(task_scheduler, 'check_schedule_request_acl_caller', err)
self.mock(task_scheduler, 'check_schedule_request_acl_service_account', err)
def test_new_ok_in_realms_mode(self):
self.set_as_user()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.mock(service_accounts, 'has_token_server', lambda: True)
self.mock_auth_db([
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
])
# Legacy pool ACLs must no be used in this mode, only realm ACLs are used.
self.fail_on_using_legacy_acls()
ntr = self._new_task_request_prpc()
ntr.properties.command.extend([u'echo', u'hi'])
ntr.properties.dimensions.extend(
[swarming_pb2.StringPair(key=u'pool', value=u'default')])
ntr.properties.execution_timeout_secs = 30
ntr.service_account = 'service-account@example.com'
ntr.realm = 'test:task_realm'
response = self.post_prpc('NewTask', ntr)
actual = swarming_pb2.TaskRequestMetadataResponse()
_decode(response.body, actual)
self.assertEqual(u'5cee488008810', actual.task_id)
# Get the produced TaskRequest.
key, _ = task_pack.get_request_and_result_keys(actual.task_id)
req = key.get()
# Make sure associated the task with the correct realm.
self.assertEqual('test:task_realm', req.realm)
# Correctly initialized the service account state.
self.assertEqual('service-account@example.com', req.service_account)
def test_new_ok_with_default_task_realm_not_enforced(self):
self.set_as_user()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.mock_default_pool_acl(service_accounts=[],
default_task_realm='test:task_realm',
enforced_realm_permissions=None)
# Realm ACLs should be ignored. Note that they are still called to record
# dry run check results.
self.mock(auth, 'has_permission', lambda *_args, **_kwargs: False)
ntr = self._new_task_request_prpc()
ntr.properties.command.extend([u'echo', u'hi'])
ntr.properties.dimensions.extend(
[swarming_pb2.StringPair(key=u'pool', value=u'default')])
ntr.properties.execution_timeout_secs = 30
response = self.post_prpc('NewTask', ntr)
actual = swarming_pb2.TaskRequestMetadataResponse()
_decode(response.body, actual)
self.assertEqual(u'5cee488008810', actual.task_id)
# Get the produced TaskRequest.
key, _ = task_pack.get_request_and_result_keys(actual.task_id)
req = key.get()
# Make sure associated the task with the correct realm.
self.assertEqual('test:task_realm', req.realm)
def test_new_ok_with_default_task_realm_enforced(self):
self.set_as_user()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.mock(service_accounts, 'has_token_server', lambda: True)
self.mock_auth_db([
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
])
self.mock_default_pool_acl(
service_accounts=['service-account@example.com'],
default_task_realm='test:task_realm',
enforced_realm_permissions={
realms_pb2.REALM_PERMISSION_POOLS_CREATE_TASK,
realms_pb2.REALM_PERMISSION_TASKS_CREATE_IN_REALM,
realms_pb2.REALM_PERMISSION_TASKS_ACT_AS,
})
# Legacy pool ACLs must no be used in this mode, only realm ACLs are used.
self.fail_on_using_legacy_acls()
ntr = self._new_task_request_prpc()
ntr.properties.command.extend([u'echo', u'hi'])
ntr.properties.dimensions.extend(
[swarming_pb2.StringPair(key=u'pool', value=u'default')])
ntr.properties.execution_timeout_secs = 30
ntr.service_account = 'service-account@example.com'
response = self.post_prpc('NewTask', ntr)
actual = swarming_pb2.TaskRequestMetadataResponse()
_decode(response.body, actual)
self.assertEqual(u'5cee488008810', actual.task_id)
# Get the produced TaskRequest.
key, _ = task_pack.get_request_and_result_keys(actual.task_id)
req = key.get()
# Make sure associated the task with the correct realm.
self.assertEqual('test:task_realm', req.realm)
# Correctly initialized the service account state.
self.assertEqual('service-account@example.com', req.service_account)
def test_new_invalid_realm(self):
self.set_as_user()
ntr = self._new_task_request_prpc()
ntr.properties.command.extend([u'echo', u'hi'])
ntr.properties.dimensions.extend(
[swarming_pb2.StringPair(key=u'pool', value=u'default')])
ntr.properties.execution_timeout_secs = 30
ntr.service_account = 'service-account@example.com'
ntr.realm = 'test/invalid'
response = self.post_prpc('NewTask', ntr, expect_errors=True)
expected = ("Bad realm u'test/invalid', want "
"&quot;&lt;project&gt;:&lt;name&gt;&quot;")
actual = response.body
self.assertEqual(expected, actual)
@ndb.tasklet
def _updateToken(self, *_args, **_kwargs):
raise ndb.Return('update-token')
def test_new_ok_with_resultdb_and_realm(self):
self.set_as_user()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.mock_auth_db([
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
])
with mock.patch('server.resultdb.create_invocation_async',
side_effect=self._updateToken):
ntr = self._new_task_request_prpc()
ntr.ClearField('expiration_secs')
task_slice = swarming_pb2.TaskSlice(expiration_secs=180,
wait_for_capacity=True)
apply_default_for_task_props(task_slice.properties)
task_slice.properties.command[:] = ['python', 'run_test.py']
ntr.task_slices.extend([task_slice])
ntr.realm = 'test:task_realm'
ntr.resultdb.enable = True
_, task_id = self._client_create_task_prpc(ntr)
# Get the produced TaskRequest, and verify the realm and resultdb config.
rKey, sKey = task_pack.get_request_and_result_keys(task_id)
request, summary = rKey.get(), sKey.get()
self.assertEqual('test:task_realm', request.realm)
self.assertEqual('invocations/task-test-swarming.appspot.com-5cee488008811',
summary.resultdb_info.invocation)
def test_new_ok_with_resultdb_and_default_task_realm(self):
self.set_as_user()
self.mock(random, 'getrandbits', lambda _: 0x88)
self.mock_auth_db([
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
])
self.mock_default_pool_acl(
service_accounts=['service-account@example.com'],
default_task_realm='test:task_realm',
enforced_realm_permissions={
realms_pb2.REALM_PERMISSION_POOLS_CREATE_TASK,
realms_pb2.REALM_PERMISSION_TASKS_CREATE_IN_REALM,
realms_pb2.REALM_PERMISSION_TASKS_ACT_AS,
})
with mock.patch('server.resultdb.create_invocation_async',
side_effect=self._updateToken):
ntr = self._new_task_request_prpc()
ntr.ClearField('expiration_secs')
task_slice = swarming_pb2.TaskSlice(expiration_secs=180,
wait_for_capacity=True)
apply_default_for_task_props(task_slice.properties)
task_slice.properties.command[:] = ['python', 'run_test.py']
ntr.task_slices.extend([task_slice])
ntr.realm = 'test:task_realm'
ntr.resultdb.enable = True
_, task_id = self._client_create_task_prpc(ntr)
# Get the produced TaskRequest, and verify the realm and resultdb config.
rKey, sKey = task_pack.get_request_and_result_keys(task_id)
request, summary = rKey.get(), sKey.get()
self.assertEqual('test:task_realm', request.realm)
self.assertEqual('invocations/task-test-swarming.appspot.com-5cee488008811',
summary.resultdb_info.invocation)
def _prepare_mass_cancel(self):
# Create 3 tasks: one pending, one running, one complete.
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_bot()
self.do_handshake(do_first_poll=True)
# Completed.
self.set_as_user()
ntr = self._new_task_request_prpc(use_default_slice=True)
ntr.name = "first"
ntr.tags[:] = ['project:yay', 'commit:abcd', 'os:Win']
_, _ = self._client_create_task_prpc(ntr)
self.set_as_bot()
self.bot_run_task()
# Running.
self.set_as_user()
self.mock_now(self.now, 60)
ntr = self._new_task_request_prpc(use_default_slice=True)
ntr.name = 'second'
ntr.user = 'jack@localhost'
ntr.tags[:] = ['project:yay', 'commit:efgh', 'os:Win']
_, running_id = self._client_create_task_prpc(ntr)
self.set_as_bot()
self.bot_poll()
# Pending.
self.set_as_user()
now_120 = self.mock_now(self.now, 120)
ntr = self._new_task_request_prpc(use_default_slice=True)
ntr.name = 'third'
ntr.user = 'jack@localhost'
ntr.tags[:] = ['project:yay', 'commit:ijkhl', 'os:Linux']
_, pending_id = self._client_create_task_prpc(ntr)
return running_id, pending_id, now_120
def test_mass_cancel_pending(self):
_, pending_id, now_120 = self._prepare_mass_cancel()
def enqueue_task(url, name, payload):
self.assertEqual('/internal/taskqueue/important/tasks/cancel', url)
self.assertEqual('cancel-tasks', name)
e = {'tasks': [pending_id], 'kill_running': True}
self.assertEqual(e, json.loads(payload))
return True
self.mock(utils, 'enqueue_task', enqueue_task)
self.set_as_admin()
tcr = swarming_pb2.TasksCancelRequest(
limit=100,
tags=['project:yay'],
start=message_conversion_prpc.date(self.now -
datetime.timedelta(seconds=1)),
end=message_conversion_prpc.date(now_120 +
datetime.timedelta(seconds=1)),
)
resp = self.post_prpc('CancelTasks', tcr)
actual = swarming_pb2.TasksCancelResponse()
_decode(resp.body, actual)
expected = swarming_pb2.TasksCancelResponse(
matched=1, now=message_conversion_prpc.date(now_120))
self.assertEqual(expected, actual)
def test_mass_cancel_running(self):
running_id, pending_id, now_120 = self._prepare_mass_cancel()
def enqueue_task(url, name, payload):
self.assertEqual('/internal/taskqueue/important/tasks/cancel', url)
self.assertEqual('cancel-tasks', name)
e = {'tasks': [pending_id, running_id], 'kill_running': True}
self.assertEqual(e, json.loads(payload))
return True
self.mock(utils, 'enqueue_task', enqueue_task)
self.set_as_admin()
tcr = swarming_pb2.TasksCancelRequest(
limit=100,
tags=['project:yay'],
kill_running=True,
start=message_conversion_prpc.date(self.now -
datetime.timedelta(seconds=1)),
end=message_conversion_prpc.date(now_120 +
datetime.timedelta(seconds=1)),
)
resp = self.post_prpc('CancelTasks', tcr)
actual = swarming_pb2.TasksCancelResponse()
_decode(resp.body, actual)
expected = swarming_pb2.TasksCancelResponse(
matched=2, now=message_conversion_prpc.date(now_120))
self.assertEqual(expected, actual)
def test_mass_cancel_realm_permission(self):
# non-privileged user without realm permission.
self.set_as_user()
self.mock_auth_db([])
# the user needs to specify a pool filter.
tcr = swarming_pb2.TasksCancelRequest(
limit=100,
tags=['project:yay'],
kill_running=True,
)
resp = self.post_prpc('CancelTasks', tcr, expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
self.assertEqual(u'No pool is specified', resp.body)
# the user can't access the tasks without permission.
tcr.tags[:] = ['pool:default']
resp = self.post_prpc('CancelTasks', tcr, expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
self.assertEqual(
cgi.escape((u'user "user@example.com" does not have '
u'permission "swarming.pools.cancelTask"'),
quote=True), resp.body)
# give permission to the user.
self.mock_auth_db([auth.Permission('swarming.pools.cancelTask')])
# the user still needs to specify a pool filter.
tcr.tags[:] = ['foo:bar']
resp = self.post_prpc('CancelTasks', tcr, expect_errors=True)
self.assertEqual(resp.status, '403 Forbidden')
self.assertEqual(u'No pool is specified', resp.body)
# ok if the user has a permission of the specified pool.
tcr.tags[:] = ['pool:default']
self.post_prpc('CancelTasks', tcr)
def _gen_two_tasks(self):
self.mock_now(self.now)
self.mock(random, 'getrandbits', lambda _: 0x88)
self.set_as_bot()
self.bot_poll()
self.set_as_user()
ntr = self._new_task_request_prpc(use_default_slice=True)
ntr.name = "first"
ntr.tags[:] = ['project:yay', 'commit:post', 'pool:default']
ntr.task_slices[0].properties.idempotent = True
_, first_id = self._client_create_task_prpc(ntr)
self.set_as_bot()
self.bot_run_task()
# Hack the datastore so MODIFIED_TS returns in backward order compared to
# CREATED_TS.
now_120 = self.mock_now(self.now, 120)
entity = task_pack.unpack_result_summary_key(first_id).get()
entity.modified_ts = now_120
entity.put()
self.set_as_user()
self.mock(random, 'getrandbits', lambda _: 0x66)
now_60 = self.mock_now(self.now, 60)
ntr = self._new_task_request_prpc(use_default_slice=True)
ntr.name = 'second'
ntr.user = 'jack@localhost'
ntr.tags[:] = ['project:yay', 'commit:pre', 'pool:default']
ntr.task_slices[0].properties.idempotent = True
_, second_id = self._client_create_task_prpc(ntr)
first = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_result_summary(first)
first.task_id = first_id
first.name = "first"
first.bot_idle_since_ts.FromDatetime(self.now)
first.completed_ts.FromDatetime(self.now)
first.costs_usd[:] = [0.1]
first.created_ts.FromDatetime(self.now)
first.duration = 0.1
first.exit_code = 0
first.modified_ts.FromDatetime(now_120)
self.gen_perf_stats_prpc(first.performance_stats)
first.started_ts.FromDatetime(self.now)
first.tags[:] = [
u'authenticated:user:user@example.com',
u'commit:post',
u'os:Amiga',
u'pool:default',
u'priority:20',
u'project:yay',
u'realm:none',
u'service_account:none',
u'swarming.pool.template:none',
u'swarming.pool.version:pools_cfg_rev',
u'user:joe@localhost',
]
first.user = 'joe@localhost'
second = swarming_pb2.TaskResultResponse()
self.apply_defaults_for_result_summary(second)
second.name = "second"
second.bot_idle_since_ts.FromDatetime(self.now)
second.completed_ts.FromDatetime(self.now)
second.created_ts.FromDatetime(now_60)
second.deduped_from = '5cee488008811'
second.cost_saved_usd = 0.1
second.duration = 0.1
second.exit_code = 0
second.modified_ts.FromDatetime(now_60)
second.run_id = '5cee488008811'
second.started_ts.FromDatetime(self.now)
second.tags[:] = [
u'authenticated:user:user@example.com', u'commit:pre', u'os:Amiga',
u'pool:default', u'priority:20', u'project:yay', u'realm:none',
u'service_account:none', u'swarming.pool.template:none',
u'swarming.pool.version:pools_cfg_rev', u'user:jack@localhost'
]
second.task_id = second_id
second.user = 'jack@localhost'
start = self.now - datetime.timedelta(days=1)
end = self.now + datetime.timedelta(days=1)
return first, second, now_120, message_conversion_prpc.date(
start), message_conversion_prpc.date(end)
def test_list_ok(self):
utils.clear_cache(config.settings)
first, second, now_120, start, end = self._gen_two_tasks()
# Basic request. Default sort is CREATED_TS
self.set_as_privileged_user()
request = swarming_pb2.TasksWithPerfRequest(limit=100,
start=start,
end=end,
state=swarming_pb2.QUERY_COMPLETED,
include_performance_stats=True)
expected = swarming_pb2.TaskListResponse(
now=message_conversion_prpc.date(now_120), items=[second, first])
self.mock_now(self.now)
resp = self.post_prpc('ListTasks', request)
actual = swarming_pb2.TaskListResponse()
_decode(resp.body, actual)
self.assertEqual(expected.items, actual.items)
# Only return the second with specific tags
request = swarming_pb2.TasksWithPerfRequest(limit=100,
start=start,
end=end,
state=swarming_pb2.QUERY_COMPLETED,
include_performance_stats=True,
tags=['project:yay', 'commit:pre'])
expected = swarming_pb2.TaskListResponse(items=[second])
resp = self.post_prpc('ListTasks', request)
actual = swarming_pb2.TaskListResponse()
_decode(resp.body, actual)
self.assertEqual(expected.items, actual.items)
# Return both since or tags are used
request = swarming_pb2.TasksWithPerfRequest(
limit=100,
end=end,
start=start,
tags=['commit:post|pre'],
state=swarming_pb2.QUERY_COMPLETED,
include_performance_stats=False,
)
first.ClearField('performance_stats')
expected = swarming_pb2.TaskListResponse(items=[second, first])
resp = self.post_prpc('ListTasks', request)
actual = swarming_pb2.TaskListResponse()
_decode(resp.body, actual)
self.assertEqual(expected.items, actual.items)
request = swarming_pb2.TasksWithPerfRequest(
limit=100,
state=swarming_pb2.QUERY_COMPLETED,
end=end,
start=start,
tags=['foo:bar'],
)
actual = swarming_pb2.TaskListResponse()
resp = self.post_prpc('ListTasks', request)
_decode(resp.body, actual)
self.assertEqual(0, len(actual.items))
# Both state and tag.
request = swarming_pb2.TasksWithPerfRequest(
limit=100,
end=end,
start=start,
tags=['commit:pre'],
state=swarming_pb2.QUERY_COMPLETED_SUCCESS)
expected = swarming_pb2.TaskListResponse(items=[second])
resp = self.post_prpc('ListTasks', request)
actual = swarming_pb2.TaskListResponse()
_decode(resp.body, actual)
self.assertEqual(expected.items, actual.items)
def test_count_indexes(self):
# Asserts that no combination crashes.
_, _, _, start, end = self._gen_two_tasks()
self.set_as_privileged_user()
for state in swarming_pb2.StateQuery.DESCRIPTOR.values_by_number:
for tags in ([], ['a:1'], ['a:1', 'b:2']):
request = swarming_pb2.TasksCountRequest(
start=start,
end=end,
state=state,
tags=tags,
)
self.post_prpc('CountTasks', request)
@parameterized.expand([
('ListTasks', swarming_pb2.TasksWithPerfRequest(limit=10)),
('CountTasks', swarming_pb2.TasksCountRequest()),
])
def test_realm_permissions_count(self, rpc, request):
# non-privileged user without realm permission.
self.set_as_user()
self.mock_auth_db([])
start = self.now - datetime.timedelta(days=1)
# the user needs to specify a pool filter.
request.start.FromDatetime(start)
expected = cgi.escape('No pool is specified', quote=True)
response = self.post_prpc(rpc, request, expect_errors=True)
self.assertEqual(expected, response.body)
# the user can't access the tasks without permission.
request.tags.extend(['pool:default'])
expected = cgi.escape(
'user "user@example.com" does not have '
'permission "swarming.pools.listTasks"',
quote=True)
response = self.post_prpc(rpc, request, expect_errors=True)
self.assertEqual(expected, response.body)
# give permission to the user.
self.mock_auth_db([auth.Permission('swarming.pools.listTasks')])
response = self.post_prpc(rpc, request)
def test_count(self):
_, _, _, the_start, the_end = self._gen_two_tasks()
self.set_as_privileged_user()
def _verify(count,
state=swarming_pb2.QUERY_COMPLETED_SUCCESS,
tags=None,
start=None,
end=None):
if start is None:
start = the_start
if end is None:
end = the_end
if tags is None:
tags = []
request = swarming_pb2.TasksCountRequest(
start=start,
end=end,
tags=tags,
state=state,
)
actual = swarming_pb2.TasksCount()
response = self.post_prpc('CountTasks', request)
_decode(response.body, actual)
self.assertEqual(count, actual.count)
_verify(2)
_verify(1, tags=['commit:pre'])
_verify(0, state=swarming_pb2.QUERY_PENDING)
new_start = message_conversion_prpc.date(
the_start.ToDatetime() + datetime.timedelta(days=5), )
new_end = message_conversion_prpc.date(
the_end.ToDatetime() + datetime.timedelta(days=10), )
_verify(0, start=new_start, end=new_end)
def test_list_request(self):
self.mock_now(self.now)
self.mock(random, 'getrandbits', lambda _: 0x88)
names = ["foo", "bar", "baz"]
unique_tags = ["a:1", "b:2", "c:3"]
expected_requests = []
ntrs = []
ticker = test_case.Ticker(self.now, datetime.timedelta(seconds=60))
times = [ticker()]
for name, tag in zip(names, unique_tags):
tags = ['project:yay', 'commit:post', 'pool:default', tag]
ntr = self._new_task_request_prpc(use_default_slice=True)
ntr.name = name
ntr.tags[:] = tags
ntrs.append(ntr)
expected = swarming_pb2.TaskRequestResponse()
expected.properties.CopyFrom(ntr.task_slices[0].properties)
all_tags = sorted(tags + [
"priority:20",
"os:Amiga",
"realm:none",
"service_account:none",
"swarming.pool.template:none",
"swarming.pool.version:pools_cfg_rev",
"authenticated:user:priv@example.com",
"user:" + ntr.user,
])
expected.tags[:] = all_tags
expected.task_slices.extend(ntr.task_slices)
expected.resultdb.CopyFrom(swarming_pb2.ResultDBCfg())
expected.name = name
expected.service_account = "none"
expected.user = ntr.user
expected.priority = ntr.priority
expected.authenticated = "user:priv@example.com"
expected.expiration_secs = ntr.task_slices[0].expiration_secs
expected.bot_ping_tolerance_secs = ntr.bot_ping_tolerance_secs
expected_requests.append(expected)
self.set_as_privileged_user()
_, first_id = self._client_create_task_prpc(ntrs[0])
expected_requests[0].created_ts.FromDatetime(self.now)
expected_requests[0].task_id = first_id
times.append(ticker())
self.mock_now(times[-1])
self.set_as_bot()
self.bot_poll()
self.bot_run_task()
self.set_as_privileged_user()
for i in range(1, len(ntrs)):
times.append(ticker())
self.mock_now(times[-1])
_, task_id = self._client_create_task_prpc(ntrs[i])
expected_requests[i].task_id = task_id
expected_requests[i].created_ts.FromDatetime(times[-1])
def _verify(present,
start=None,
end=None,
state=swarming_pb2.QUERY_ALL,
sort=swarming_pb2.QUERY_CREATED_TS,
tags=None,
timeoffset=None):
if timeoffset is None:
timeoffset = datetime.timedelta(seconds=1)
if start is None:
start = times[0] - timeoffset
if end is None:
end = times[-1] + timeoffset
if tags is None:
tags = []
request = swarming_pb2.TasksRequest(limit=10,
state=state,
sort=sort,
tags=tags)
request.start.FromDatetime(start)
request.end.FromDatetime(end)
response = self.post_prpc('ListTaskRequests', request)
actual = swarming_pb2.TaskRequestsResponse()
_decode(response.body, actual)
expected_items = [expected_requests[i] for i in present]
self.assertEqual(expected_items, list(actual.items))
_verify([0], end=times[1])
_verify([], tags=["e:1"])
_verify([0], state=swarming_pb2.QUERY_COMPLETED_SUCCESS)
_verify([2, 1, 0])
_verify([0], tags=["a:1"])
def test_list_task_states_ok(self):
self.set_as_privileged_user()
ntr = self._new_task_request_prpc(use_default_slice=True)
_, pending_id = self._client_create_task_prpc(ntr)
_, canceled_id = self._client_create_task_prpc(ntr)
self._client_cancel_task_prpc(canceled_id)
request = swarming_pb2.TaskStatesRequest(
task_id=[canceled_id, pending_id, '1d69b9f088008810'])
response = self.post_prpc('ListTaskStates', request)
actual = swarming_pb2.TaskStates()
_decode(response.body, actual)
expected = swarming_pb2.TaskStates(states=[
swarming_pb2.CANCELED, swarming_pb2.PENDING, swarming_pb2.PENDING
])
self.assertEqual(expected, actual)
def test_list_task_states_invalid(self):
self.set_as_privileged_user()
request = swarming_pb2.TaskStatesRequest(task_id=['invalid'])
response = self.post_prpc('ListTaskStates', request, True)
self.assertEqual(response.status, '400 Bad Request')
expected_error = cgi.escape("Invalid task_id: Invalid key u'invali'",
quote=True)
self.assertEqual(expected_error, response.body)
def test_batch_get_result(self):
self.set_as_privileged_user()
def create_task(realm):
perms = [
auth.Permission('swarming.pools.createTask'),
auth.Permission('swarming.tasks.createInRealm'),
]
self.mock_auth_db(perms, task_realm=realm)
ntr = self._new_task_request_prpc(use_default_slice=True)
ntr.realm = realm
return self._client_create_task_prpc(ntr)[1]
invisible_pending_id = create_task('test:invisible')
invisible_canceled_id = create_task('test:invisible')
self._client_cancel_task_prpc(invisible_canceled_id)
visible_pending_id = create_task('test:visible')
visible_canceled_id = create_task('test:visible')
self._client_cancel_task_prpc(visible_canceled_id)
run_result_id = visible_pending_id[:-1] + '1'
missing_id = '1d69b9f088008810'
# Need to switch to a different user, otherwise all tasks will be visible,
# since whoever created the task is always allowed to see it. Also need
# a non-privileged user that doesn't have acl.can_view_all_tasks() perm.
self.set_as_user()
self.mock_auth_db([auth.Permission('swarming.tasks.get')],
task_realm='test:visible')
request = swarming_pb2.BatchGetResultRequest(task_ids=[
visible_pending_id,
visible_canceled_id,
invisible_pending_id,
invisible_canceled_id,
run_result_id, # won't work, need summary ID
missing_id,
])
response = self.post_prpc('BatchGetResult', request)
actual = swarming_pb2.BatchGetResultResponse()
_decode(response.body, actual)
extract = []
for res in actual.results:
if res.HasField('error'):
extract.append((res.task_id, res.error.message))
else:
self.assertEqual(res.task_id, res.result.task_id)
extract.append((res.task_id, res.result.state))
self.assertEqual(extract, [
(visible_pending_id, swarming_pb2.PENDING),
(visible_canceled_id, swarming_pb2.CANCELED),
(invisible_pending_id, u'No access to see the status of this task'),
(invisible_canceled_id, u'No access to see the status of this task'),
(
run_result_id,
u'Bad task ID "%s": Can\'t reference to a specific try result.' %
run_result_id,
),
(missing_id, u'No such task'),
])
class InternalsServicePrpcTest(PrpcTest):
def setUp(self):
super(InternalsServicePrpcTest, self).setUp()
self.service = 'swarming.internals.rbe.Internals'
@mock.patch('server.task_scheduler.expire_slice')
def test_expire_slice_no_resource(self, expire_slice_mock):
req_key = task_request.new_request_key()
task_id = task_pack.pack_result_summary_key(
task_pack.request_key_to_result_summary_key(req_key))
self.set_as_swarming_itself()
self.post_prpc(
'ExpireSlice',
rbe_pb2.ExpireSliceRequest(
task_id=task_id,
task_to_run_shard=15,
task_to_run_id=32, # slice #2
reason=rbe_pb2.ExpireSliceRequest.NO_RESOURCE,
))
expire_slice_mock.assert_called_once_with(
task_to_run.task_to_run_key_from_parts(req_key, 15, 32),
task_result.State.NO_RESOURCE,
)
@mock.patch('server.task_scheduler.expire_slice')
def test_expire_slice_permission_denied(self, expire_slice_mock):
req_key = task_request.new_request_key()
task_id = task_pack.pack_result_summary_key(
task_pack.request_key_to_result_summary_key(req_key))
self.set_as_swarming_itself()
self.post_prpc(
'ExpireSlice',
rbe_pb2.ExpireSliceRequest(
task_id=task_id,
task_to_run_shard=15,
task_to_run_id=32, # slice #2
reason=rbe_pb2.ExpireSliceRequest.PERMISSION_DENIED,
details='Boo',
))
expire_slice_mock.assert_called_once_with(
task_to_run.task_to_run_key_from_parts(req_key, 15, 32),
task_result.State.EXPIRED,
)
errs = list(ereporter2.Error.query())
self.assertEqual(len(errs), 1)
self.assertEqual(
errs[0].to_dict(include=['category', 'message', 'params']), {
'category': u'PERMISSION_DENIED',
'message': u'Boo',
'params': {
u'slice_index': 2,
u'task_id': task_id,
},
})
@mock.patch('server.task_scheduler.expire_slice')
def test_expire_slice_bot_died(self, expire_slice_mock):
req_key = task_request.new_request_key()
task_id = task_pack.pack_result_summary_key(
task_pack.request_key_to_result_summary_key(req_key))
self.set_as_swarming_itself()
self.post_prpc(
'ExpireSlice',
rbe_pb2.ExpireSliceRequest(
task_id=task_id,
task_to_run_shard=15,
task_to_run_id=32, # slice #2
reason=rbe_pb2.ExpireSliceRequest.BOT_INTERNAL_ERROR,
details='Boo',
))
expire_slice_mock.assert_called_once_with(
task_to_run.task_to_run_key_from_parts(req_key, 15, 32),
task_result.State.BOT_DIED,
)
errs = list(ereporter2.Error.query())
self.assertEqual(len(errs), 1)
self.assertEqual(
errs[0].to_dict(include=['category', 'message', 'params']), {
'category': u'BOT_INTERNAL_ERROR',
'message': u'Boo',
'params': {
u'slice_index': 2,
u'task_id': task_id,
},
})
class SwarmingServicePrpcTest(PrpcTest):
def setUp(self):
super(SwarmingServicePrpcTest, self).setUp()
self.service = 'swarming.v2.Swarming'
self.now = datetime.datetime(2010, 1, 2, 3, 4, 5)
self.mock_now(self.now)
self.mock_tq_tasks()
@parameterized.expand(['GetDetails', 'GetToken'])
def test_forbidden(self, rpc):
self.set_as_anonymous()
response = self.post_prpc(rpc, proto.empty_pb2.Empty(), expect_errors=True)
self.assertEqual(response.status, '403 Forbidden')
def test_details(self):
self.set_as_privileged_user()
expected = swarming_pb2.ServerDetails(
bot_version=bot_code.get_bot_version('https://testbed.example.com')[0],
server_version=utils.get_app_version(),
cas_viewer_server='https://test-cas-viewer-server.com',
)
response = self.post_prpc('GetDetails', proto.empty_pb2.Empty())
actual = swarming_pb2.ServerDetails()
_decode(response.body, actual)
self.assertEqual(expected, actual)
@mock.patch('server.bot_code.generate_bootstrap_token')
def test_get_token_ok(self, token_mock):
self.set_as_admin()
token = 'the_token'
token_mock.return_value = token
response = self.post_prpc('GetToken', proto.empty_pb2.Empty())
actual = swarming_pb2.BootstrapToken()
_decode(response.body, actual)
expected = swarming_pb2.BootstrapToken(bootstrap_token=token)
self.assertEqual(expected, actual)
def test_public_permissions(self):
self.set_as_anonymous()
expected = swarming_pb2.ClientPermissions(cancel_task=False,
cancel_tasks=False,
delete_bot=False,
delete_bots=False,
get_bootstrap_token=False,
get_configs=False,
put_configs=False,
terminate_bot=False)
actual = swarming_pb2.ClientPermissions()
response = self.post_prpc('GetPermissions',
swarming_pb2.PermissionsRequest())
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_user_permissions(self):
self.set_as_user()
expected = swarming_pb2.ClientPermissions(cancel_task=False,
cancel_tasks=False,
delete_bot=False,
delete_bots=False,
get_bootstrap_token=False,
get_configs=False,
put_configs=False,
terminate_bot=False)
actual = swarming_pb2.ClientPermissions()
response = self.post_prpc('GetPermissions',
swarming_pb2.PermissionsRequest())
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_user_permissions_with_bot_id(self):
self.mock_default_pool_acl([])
self.set_as_bot()
self.bot_poll()
self.set_as_user()
self.mock_auth_db([
auth.Permission('swarming.pools.deleteBot'),
auth.Permission('swarming.pools.listBots'),
auth.Permission('swarming.pools.terminateBot'),
])
expected = swarming_pb2.ClientPermissions(cancel_task=False,
cancel_tasks=False,
delete_bot=True,
delete_bots=False,
get_bootstrap_token=False,
list_bots=['default', 'template'],
get_configs=False,
put_configs=False,
terminate_bot=True)
actual = swarming_pb2.ClientPermissions()
response = self.post_prpc('GetPermissions',
swarming_pb2.PermissionsRequest(bot_id='bot1'))
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_user_permissions_with_task_id(self):
self.mock_default_pool_acl([])
self.set_as_admin()
ntr = self._new_task_request_prpc(use_default_slice=True)
_, task_id = self._client_create_task_prpc(ntr)
self.set_as_user()
self.mock_auth_db([
auth.Permission('swarming.pools.cancelTask'),
auth.Permission('swarming.pools.listTasks'),
])
expected = swarming_pb2.ClientPermissions(
cancel_task=True,
cancel_tasks=False,
delete_bot=False,
delete_bots=False,
get_bootstrap_token=False,
list_tasks=['default', 'template'],
get_configs=False,
put_configs=False,
terminate_bot=False)
actual = swarming_pb2.ClientPermissions()
response = self.post_prpc('GetPermissions',
swarming_pb2.PermissionsRequest(task_id=task_id))
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_user_permissions_with_pool_tags(self):
"""Asserts that permissions respond correctly to a user has pool
permissions.
"""
self.mock_default_pool_acl([])
# The user has realm permissions.
self.set_as_user()
self.mock_auth_db([
auth.Permission('swarming.pools.cancelTask'),
auth.Permission('swarming.pools.deleteBot'),
auth.Permission('swarming.pools.listBots'),
auth.Permission('swarming.pools.listTasks'),
])
expected = swarming_pb2.ClientPermissions(
cancel_task=False,
cancel_tasks=True,
delete_bot=False,
delete_bots=True,
get_bootstrap_token=False,
list_tasks=['default', 'template'],
list_bots=['default', 'template'],
get_configs=False,
put_configs=False,
terminate_bot=False)
actual = swarming_pb2.ClientPermissions()
response = self.post_prpc(
'GetPermissions',
swarming_pb2.PermissionsRequest(tags=['pool:default']))
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_privileged_user_permissions(self):
self.set_as_privileged_user()
expected = swarming_pb2.ClientPermissions(cancel_task=True,
cancel_tasks=False,
delete_bot=False,
delete_bots=False,
get_bootstrap_token=False,
get_configs=False,
put_configs=False,
terminate_bot=True)
actual = swarming_pb2.ClientPermissions()
response = self.post_prpc('GetPermissions',
swarming_pb2.PermissionsRequest())
_decode(response.body, actual)
self.assertEqual(expected, actual)
def test_admin_permissions(self):
self.set_as_admin()
expected = swarming_pb2.ClientPermissions(cancel_task=True,
cancel_tasks=True,
delete_bot=True,
delete_bots=True,
get_bootstrap_token=True,
get_configs=False,
put_configs=False,
terminate_bot=True)
actual = swarming_pb2.ClientPermissions()
response = self.post_prpc('GetPermissions',
swarming_pb2.PermissionsRequest())
_decode(response.body, actual)
self.assertEqual(expected, actual)
if __name__ == '__main__':
if '-v' in sys.argv:
unittest.TestCase.maxDiff = None
logging.basicConfig(level=logging.DEBUG)
else:
logging.basicConfig(level=logging.FATAL)
unittest.main()