GNOME.org
 

[Notes] [Git][BuildGrid/buildgrid][finn/48-cancellation-leases] 6 commits: Updated imports for function `run_in_subprocess`

Title: GitLab

finn pushed to branch finn/48-cancellation-leases at BuildGrid / buildgrid

Commits:

7 changed files:

Changes:

  • tests/cas/test_client.py
    ... ... @@ -26,7 +26,8 @@ from buildgrid.client.cas import download, upload
    26 26 
     from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
    27 27 
     from buildgrid.utils import create_digest
    28 28
    29 
    -from ..utils.cas import serve_cas, run_in_subprocess
    29 
    +from ..utils.cas import serve_cas
    30 
    +from ..utils.utils import run_in_subprocess
    30 31
    31 32
    32 33 
     INTANCES = ['', 'instance']

  • tests/cas/test_storage.py
    ... ... @@ -32,7 +32,8 @@ from buildgrid.server.cas.storage.s3 import S3Storage
    32 32 
     from buildgrid.server.cas.storage.with_cache import WithCacheStorage
    33 33 
     from buildgrid.settings import HASH
    34 34
    35 
    -from ..utils.cas import serve_cas, run_in_subprocess
    35 
    +from ..utils.cas import serve_cas
    36 
    +from ..utils.utils import run_in_subprocess
    36 37
    37 38
    38 39 
     BLOBS = [(b'abc', b'defg', b'hijk', b'')]

  • tests/integration/bot_session.py
    ... ... @@ -14,56 +14,174 @@
    14 14
    15 15 
     # pylint: disable=redefined-outer-name
    16 16
    17 
    -import uuid
    17 
    +import asyncio
    18 18
    19 
    +import grpc
    19 20 
     import pytest
    20 21
    21 
    -from buildgrid.bot import bot_session
    22 
    +from buildgrid._enums import LeaseState
    23 
    +from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
    24 
    +from buildgrid.bot.hardware.worker import Worker
    25 
    +from buildgrid.bot.hardware.interface import HardwareInterface
    26 
    +from buildgrid.bot.session import BotSession
    27 
    +from buildgrid.bot.interface import BotInterface
    22 28
    29 
    +from ..utils.utils import run_in_subprocess
    30 
    +from ..utils.bots_interface import serve_bots_interface
    23 31
    24 
    -@pytest.mark.parametrize("docker_value", ["True", "False"])
    25 
    -@pytest.mark.parametrize("os_value", ["nexus7", "nexus8"])
    26 
    -def test_create_device(docker_value, os_value):
    27 
    -    properties = {'docker': docker_value, 'os': os_value}
    28 
    -    device = bot_session.Device(properties)
    29 32
    30 
    -    assert uuid.UUID(device.name, version=4)
    31 
    -    assert properties == device.properties
    33 
    +INSTANCES = ['', 'instance']
    32 34
    33 35
    34 
    -def test_create_device_key_fail():
    35 
    -    properties = {'voight': 'kampff'}
    36 
    +# Use subprocess to avoid creation of gRPC threads in main process
    37 
    +# See https://github.com/grpc/grpc/blob/master/doc/fork_support.md
    38 
    +# Multiprocessing uses pickle which protobufs don't work with
    39 
    +# Workaround wrapper to send messages as strings
    40 
    +class ServerInterface:
    36 41
    37 
    -    with pytest.raises(KeyError):
    38 
    -        bot_session.Device(properties)
    42 
    +    def __init__(self, remote):
    43 
    +        self.__remote = remote
    39 44
    45 
    +    def create_bot_session(self, parent, bot_session):
    40 46
    41 
    -def test_create_device_value_fail():
    42 
    -    properties = {'docker': True}
    47 
    +        def __create_bot_session(queue, remote, parent, string_bot_session):
    48 
    +            bot_session = bots_pb2.BotSession()
    49 
    +            bot_session.ParseFromString(string_bot_session)
    43 50
    44 
    -    with pytest.raises(ValueError):
    45 
    -        bot_session.Device(properties)
    51 
    +            interface = BotInterface(grpc.insecure_channel(remote))
    46 52
    53 
    +            result = interface.create_bot_session(parent, bot_session)
    54 
    +            queue.put(result.SerializeToString())
    47 55
    48 
    -def test_create_worker():
    49 
    -    properties = {'pool': 'swim'}
    50 
    -    configs = {'DockerImage': 'Windows'}
    51 
    -    worker = bot_session.Worker(properties, configs)
    56 
    +        string_bot_session = bot_session.SerializeToString()
    57 
    +        result = run_in_subprocess(__create_bot_session,
    58 
    +                                   self.__remote, parent, string_bot_session)
    52 59
    53 
    -    assert properties == worker.properties
    54 
    -    assert configs == worker.configs
    60 
    +        bot_session = bots_pb2.BotSession()
    61 
    +        bot_session.ParseFromString(result)
    62 
    +        return bot_session
    55 63
    56 
    -    device = bot_session.Device()
    57 
    -    worker.add_device(device)
    64 
    +    def update_bot_session(self, bot_session, update_mask=None):
    58 65
    59 
    -    assert worker._devices[0] == device
    66 
    +        def __update_bot_session(queue, remote, string_bot_session, update_mask):
    67 
    +            bot_session = bots_pb2.BotSession()
    68 
    +            bot_session.ParseFromString(string_bot_session)
    60 69
    70 
    +            interface = BotInterface(grpc.insecure_channel(remote))
    61 71
    62 
    -def test_create_worker_key_fail():
    63 
    -    properties = {'voight': 'kampff'}
    64 
    -    configs = {'voight': 'kampff'}
    72 
    +            result = interface.update_bot_session(bot_session, update_mask)
    73 
    +            queue.put(result.SerializeToString())
    65 74
    66 
    -    with pytest.raises(KeyError):
    67 
    -        bot_session.Worker(properties)
    68 
    -    with pytest.raises(KeyError):
    69 
    -        bot_session.Worker(configs)
    75 
    +        string_bot_session = bot_session.SerializeToString()
    76 
    +        result = run_in_subprocess(__update_bot_session,
    77 
    +                                   self.__remote, string_bot_session, update_mask)
    78 
    +
    79 
    +        bot_session = bots_pb2.BotSession()
    80 
    +        bot_session.ParseFromString(result)
    81 
    +        return bot_session
    82 
    +
    83 
    +
    84 
    +@pytest.mark.parametrize('instance', INSTANCES)
    85 
    +def test_create_bot_session(instance):
    86 
    +
    87 
    +    with serve_bots_interface([instance]) as server:
    88 
    +        interface = ServerInterface(server.remote)
    89 
    +        hardware_interface = HardwareInterface(Worker())
    90 
    +        session = BotSession(instance, interface, hardware_interface, None)
    91 
    +        session.create_bot_session()
    92 
    +        assert session.get_pb2() == server.get_bot_session()
    93 
    +
    94 
    +
    95 
    +@pytest.mark.parametrize('instance', INSTANCES)
    96 
    +def test_update_bot_session(instance):
    97 
    +
    98 
    +    with serve_bots_interface([instance]) as server:
    99 
    +        interface = ServerInterface(server.remote)
    100 
    +        hardware_interface = HardwareInterface(Worker())
    101 
    +        session = BotSession(instance, interface, hardware_interface, None)
    102 
    +        session.create_bot_session()
    103 
    +        assert session.get_pb2() == server.get_bot_session()
    104 
    +        session.update_bot_session()
    105 
    +        assert session.get_pb2() == server.get_bot_session()
    106 
    +
    107 
    +
    108 
    +@pytest.mark.parametrize('instance', INSTANCES)
    109 
    +def test_create_bot_session_with_work(instance):
    110 
    +
    111 
    +    def __work(lease, context, event):
    112 
    +        return lease
    113 
    +
    114 
    +    with serve_bots_interface([instance]) as server:
    115 
    +        interface = ServerInterface(server.remote)
    116 
    +        hardware_interface = HardwareInterface(Worker())
    117 
    +        session = BotSession(instance, interface, hardware_interface, __work)
    118 
    +        server.inject_work()
    119 
    +        session.create_bot_session()
    120 
    +
    121 
    +        assert len(session.get_pb2().leases) == 1
    122 
    +
    123 
    +        loop = asyncio.get_event_loop()
    124 
    +        for task in asyncio.Task.all_tasks():
    125 
    +            loop.run_until_complete(task)
    126 
    +
    127 
    +        assert session.get_pb2().leases[0].state == LeaseState.COMPLETED.value
    128 
    +
    129 
    +
    130 
    +@pytest.mark.parametrize('instance', INSTANCES)
    131 
    +def test_update_bot_session_with_work(instance):
    132 
    +
    133 
    +    def __work(lease, context, event):
    134 
    +        return lease
    135 
    +
    136 
    +    with serve_bots_interface([instance]) as server:
    137 
    +        interface = ServerInterface(server.remote)
    138 
    +        hardware_interface = HardwareInterface(Worker())
    139 
    +        session = BotSession(instance, interface, hardware_interface, __work)
    140 
    +        session.create_bot_session()
    141 
    +        server.inject_work()
    142 
    +        session.update_bot_session()
    143 
    +
    144 
    +        assert len(session.get_pb2().leases) == 1
    145 
    +
    146 
    +        loop = asyncio.get_event_loop()
    147 
    +        for task in asyncio.Task.all_tasks():
    148 
    +            loop.run_until_complete(task)
    149 
    +
    150 
    +        assert session.get_pb2().leases[0].state == LeaseState.COMPLETED.value
    151 
    +
    152 
    +
    153 
    +@pytest.mark.parametrize('instance', INSTANCES)
    154 
    +def test_cancel_leases(instance):
    155 
    +
    156 
    +    def __work(lease, context, cancel_event):
    157 
    +        # while not cancel_event.is_set():
    158 
    +
    159 
    +        return lease
    160 
    +
    161 
    +    with serve_bots_interface([instance]) as server:
    162 
    +        interface = ServerInterface(server.remote)
    163 
    +        hardware_interface = HardwareInterface(Worker())
    164 
    +        session = BotSession(instance, interface, hardware_interface, __work)
    165 
    +
    166 
    +        lease = bots_pb2.Lease()
    167 
    +        lease.state = LeaseState.PENDING.value
    168 
    +        lease.id = 'foo'
    169 
    +        server.inject_work(lease)
    170 
    +        session.create_bot_session()
    171 
    +
    172 
    +        leases_pb2 = session.get_pb2().leases
    173 
    +        assert len(leases_pb2) == 1
    174 
    +        assert leases_pb2[0].state == LeaseState.ACTIVE.value
    175 
    +
    176 
    +        server.cancel_lease(leases_pb2[0].id)
    177 
    +        session.update_bot_session()
    178 
    +        assert len(session.get_pb2().leases) == 1
    179 
    +
    180 
    +        loop = asyncio.get_event_loop()
    181 
    +        for task in asyncio.Task.all_tasks():
    182 
    +            try:
    183 
    +                loop.run_until_complete(task)
    184 
    +            except asyncio.CancelledError:
    185 
    +                pass
    186 
    +
    187 
    +        assert session.get_pb2().leases[0].state == LeaseState.CANCELLED.value

  • tests/integration/bots_service.py
    ... ... @@ -17,7 +17,6 @@
    17 17
    18 18 
     # pylint: disable=redefined-outer-name
    19 19
    20 
    -import copy
    21 20 
     from unittest import mock
    22 21
    23 22 
     import grpc
    ... ... @@ -150,129 +149,6 @@ def test_update_leases_with_work(bot_session, context, instance):
    150 149 
         assert response_action == action_digest
    151 150
    152 151
    153 
    -def test_update_leases_work_complete(bot_session, context, instance):
    154 
    -    request = bots_pb2.CreateBotSessionRequest(parent='',
    155 
    -                                               bot_session=bot_session)
    156 
    -    # Create bot session
    157 
    -    # Simulated the severed binding between client and server
    158 
    -    response = copy.deepcopy(instance.CreateBotSession(request, context))
    159 
    -
    160 
    -    # Inject work
    161 
    -    action_digest = remote_execution_pb2.Digest(hash='gaff')
    162 
    -    _inject_work(instance._instances[""]._scheduler, action_digest=action_digest)
    163 
    -
    164 
    -    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    165 
    -                                               bot_session=response)
    166 
    -    response = copy.deepcopy(instance.UpdateBotSession(request, context))
    167 
    -
    168 
    -    assert response.leases[0].state == LeaseState.PENDING.value
    169 
    -    response.leases[0].state = LeaseState.ACTIVE.value
    170 
    -
    171 
    -    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    172 
    -                                               bot_session=response)
    173 
    -
    174 
    -    response = copy.deepcopy(instance.UpdateBotSession(request, context))
    175 
    -
    176 
    -    response.leases[0].state = LeaseState.COMPLETED.value
    177 
    -    response.leases[0].result.Pack(remote_execution_pb2.ActionResult())
    178 
    -
    179 
    -    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    180 
    -                                               bot_session=response)
    181 
    -    response = copy.deepcopy(instance.UpdateBotSession(request, context))
    182 
    -
    183 
    -    assert len(response.leases) is 0
    184 
    -
    185 
    -
    186 
    -def test_work_rejected_by_bot(bot_session, context, instance):
    187 
    -    request = bots_pb2.CreateBotSessionRequest(parent='',
    188 
    -                                               bot_session=bot_session)
    189 
    -    # Inject work
    190 
    -    action_digest = remote_execution_pb2.Digest(hash='gaff')
    191 
    -    _inject_work(instance._instances[""]._scheduler, action_digest=action_digest)
    192 
    -
    193 
    -    # Simulated the severed binding between client and server
    194 
    -    response = copy.deepcopy(instance.CreateBotSession(request, context))
    195 
    -
    196 
    -    # Reject work
    197 
    -    assert response.leases[0].state == LeaseState.PENDING.value
    198 
    -    response.leases[0].state = LeaseState.COMPLETED.value
    199 
    -    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    200 
    -                                               bot_session=response)
    201 
    -
    202 
    -    response = instance.UpdateBotSession(request, context)
    203 
    -
    204 
    -    context.set_code.assert_called_once_with(grpc.StatusCode.UNIMPLEMENTED)
    205 
    -
    206 
    -
    207 
    -@pytest.mark.parametrize("state", [LeaseState.LEASE_STATE_UNSPECIFIED, LeaseState.PENDING])
    208 
    -def test_work_out_of_sync_from_pending(state, bot_session, context, instance):
    209 
    -    request = bots_pb2.CreateBotSessionRequest(parent='',
    210 
    -                                               bot_session=bot_session)
    211 
    -    # Inject work
    212 
    -    action_digest = remote_execution_pb2.Digest(hash='gaff')
    213 
    -    _inject_work(instance._instances[""]._scheduler, action_digest=action_digest)
    214 
    -
    215 
    -    # Simulated the severed binding between client and server
    216 
    -    response = copy.deepcopy(instance.CreateBotSession(request, context))
    217 
    -
    218 
    -    response.leases[0].state = state.value
    219 
    -
    220 
    -    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    221 
    -                                               bot_session=response)
    222 
    -
    223 
    -    response = instance.UpdateBotSession(request, context)
    224 
    -
    225 
    -    context.set_code.assert_called_once_with(grpc.StatusCode.DATA_LOSS)
    226 
    -
    227 
    -
    228 
    -@pytest.mark.parametrize("state", [LeaseState.LEASE_STATE_UNSPECIFIED, LeaseState.PENDING])
    229 
    -def test_work_out_of_sync_from_active(state, bot_session, context, instance):
    230 
    -    request = bots_pb2.CreateBotSessionRequest(parent='',
    231 
    -                                               bot_session=bot_session)
    232 
    -    # Inject work
    233 
    -    action_digest = remote_execution_pb2.Digest(hash='gaff')
    234 
    -    _inject_work(instance._instances[""]._scheduler, action_digest=action_digest)
    235 
    -
    236 
    -    # Simulated the severed binding between client and server
    237 
    -    response = copy.deepcopy(instance.CreateBotSession(request, context))
    238 
    -
    239 
    -    response.leases[0].state = LeaseState.ACTIVE.value
    240 
    -
    241 
    -    request = copy.deepcopy(bots_pb2.UpdateBotSessionRequest(name=response.name,
    242 
    -                                                             bot_session=response))
    243 
    -
    244 
    -    response = instance.UpdateBotSession(request, context)
    245 
    -
    246 
    -    response.leases[0].state = state.value
    247 
    -
    248 
    -    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    249 
    -                                               bot_session=response)
    250 
    -
    251 
    -    response = instance.UpdateBotSession(request, context)
    252 
    -
    253 
    -    context.set_code.assert_called_once_with(grpc.StatusCode.DATA_LOSS)
    254 
    -
    255 
    -
    256 
    -def test_work_active_to_active(bot_session, context, instance):
    257 
    -    request = bots_pb2.CreateBotSessionRequest(parent='',
    258 
    -                                               bot_session=bot_session)
    259 
    -    # Inject work
    260 
    -    action_digest = remote_execution_pb2.Digest(hash='gaff')
    261 
    -    _inject_work(instance._instances[""]._scheduler, action_digest=action_digest)
    262 
    -
    263 
    -    # Simulated the severed binding between client and server
    264 
    -    response = copy.deepcopy(instance.CreateBotSession(request, context))
    265 
    -
    266 
    -    response.leases[0].state = LeaseState.ACTIVE.value
    267 
    -
    268 
    -    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    269 
    -                                               bot_session=response)
    270 
    -
    271 
    -    response = instance.UpdateBotSession(request, context)
    272 
    -
    273 
    -    assert response.leases[0].state == LeaseState.ACTIVE.value
    274 
    -
    275 
    -
    276 152 
     def test_post_bot_event_temp(context, instance):
    277 153 
         request = bots_pb2.PostBotEventTempRequest()
    278 154 
         instance.PostBotEventTemp(request, context)

  • tests/integration/hardware_interface.py
    1 
    +# Copyright (C) 2018 Bloomberg LP
    2 
    +#
    3 
    +# Licensed under the Apache License, Version 2.0 (the "License");
    4 
    +# you may not use this file except in compliance with the License.
    5 
    +# You may obtain a copy of the License at
    6 
    +#
    7 
    +#  <http://www.apache.org/licenses/LICENSE-2.0>
    8 
    +#
    9 
    +# Unless required by applicable law or agreed to in writing, software
    10 
    +# distributed under the License is distributed on an "AS IS" BASIS,
    11 
    +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12 
    +# See the License for the specific language governing permissions and
    13 
    +# limitations under the License.
    14 
    +
    15 
    +# pylint: disable=redefined-outer-name
    16 
    +
    17 
    +
    18 
    +import pytest
    19 
    +
    20 
    +from buildgrid._exceptions import FailedPreconditionError
    21 
    +from buildgrid.bot.hardware.interface import HardwareInterface
    22 
    +from buildgrid.bot.hardware.device import Device
    23 
    +from buildgrid.bot.hardware.worker import Worker
    24 
    +
    25 
    +
    26 
    +CONFIGURATIONS_WORKER = [{'DockerImage': ['Blade']}, {'DockerImage': ['Sheep', 'Snake']}]
    27 
    +PROPERTIES_WORKER = [{'pool': ['Blade']}, {'pool': ['Sheep', 'Snake']}]
    28 
    +PROPERTIES_DEVICE = [{'os': ['Blade']}, {'has-docker': ['Sheep', 'Snake']}]
    29 
    +
    30 
    +
    31 
    +def test_create_hardware():
    32 
    +    worker = Worker()
    33 
    +    interface = HardwareInterface(worker)
    34 
    +
    35 
    +    device0 = Device()
    36 
    +    worker.add_device(device0)
    37 
    +    protobuf_worker = interface.get_worker_pb2()
    38 
    +    assert len(protobuf_worker.devices) == 1
    39 
    +
    40 
    +    worker.add_device(Device())
    41 
    +    protobuf_worker = interface.get_worker_pb2()
    42 
    +    assert len(protobuf_worker.devices) == 2
    43 
    +
    44 
    +    assert protobuf_worker.devices[0].handle != protobuf_worker.devices[1].handle
    45 
    +    assert device0.name == protobuf_worker.devices[0].handle
    46 
    +
    47 
    +
    48 
    +@pytest.mark.parametrize('config', CONFIGURATIONS_WORKER)
    49 
    +def test_worker_config(config):
    50 
    +    worker = Worker(configs=config)
    51 
    +    protobuf_worker = worker.get_pb2()
    52 
    +
    53 
    +    proto_cfg = {}
    54 
    +    for cfg in protobuf_worker.configs:
    55 
    +        k = cfg.key
    56 
    +        v = cfg.value
    57 
    +
    58 
    +        proto_cfg_values = proto_cfg.get(k)
    59 
    +        if not proto_cfg_values:
    60 
    +            proto_cfg[k] = [v]
    61 
    +        else:
    62 
    +            proto_cfg_values.append(v)
    63 
    +
    64 
    +    assert config == proto_cfg
    65 
    +    assert worker.configs == config
    66 
    +
    67 
    +
    68 
    +@pytest.mark.parametrize('prop', PROPERTIES_WORKER)
    69 
    +def test_worker_property(prop):
    70 
    +    worker = Worker(properties=prop)
    71 
    +    protobuf_worker = worker.get_pb2()
    72 
    +
    73 
    +    proto_prop = {}
    74 
    +    for p in protobuf_worker.properties:
    75 
    +        k = p.key
    76 
    +        v = p.value
    77 
    +
    78 
    +        proto_prop_values = proto_prop.get(k)
    79 
    +        if not proto_prop_values:
    80 
    +            proto_prop[k] = [v]
    81 
    +        else:
    82 
    +            proto_prop_values.append(v)
    83 
    +
    84 
    +    assert prop == proto_prop
    85 
    +    assert worker.properties == prop
    86 
    +
    87 
    +
    88 
    +@pytest.mark.parametrize('prop', PROPERTIES_DEVICE)
    89 
    +def test_device_property(prop):
    90 
    +    device = Device(properties=prop)
    91 
    +    protobuf_device = device.get_pb2()
    92 
    +
    93 
    +    proto_prop = {}
    94 
    +    for p in protobuf_device.properties:
    95 
    +        k = p.key
    96 
    +        v = p.value
    97 
    +
    98 
    +        proto_prop_values = proto_prop.get(k)
    99 
    +        if not proto_prop_values:
    100 
    +            proto_prop[k] = [v]
    101 
    +        else:
    102 
    +            proto_prop_values.append(v)
    103 
    +
    104 
    +    assert prop == proto_prop
    105 
    +    assert device.properties == prop
    106 
    +
    107 
    +
    108 
    +@pytest.mark.parametrize('config', [{'piano': ['Blade']}])
    109 
    +def test_worker_config_fail(config):
    110 
    +    with pytest.raises(KeyError):
    111 
    +        Worker(configs=config)
    112 
    +
    113 
    +
    114 
    +@pytest.mark.parametrize('prop', [{'piano': ['Blade']}])
    115 
    +def test_worker_property_fail(prop):
    116 
    +    with pytest.raises(KeyError):
    117 
    +        Worker(properties=prop)
    118 
    +
    119 
    +
    120 
    +@pytest.mark.parametrize('prop', [{'piano': ['Blade']}])
    121 
    +def test_device_property_fail(prop):
    122 
    +    with pytest.raises(KeyError):
    123 
    +        Device(properties=prop)
    124 
    +
    125 
    +
    126 
    +@pytest.mark.parametrize('requirements', CONFIGURATIONS_WORKER)
    127 
    +def test_configure_hardware(requirements):
    128 
    +    hardware_interface = HardwareInterface(Worker(configs=requirements))
    129 
    +
    130 
    +    worker_requirements = Worker(configs=requirements)
    131 
    +
    132 
    +    hardware_interface.configure_hardware(worker_requirements.get_pb2())
    133 
    +
    134 
    +
    135 
    +@pytest.mark.parametrize('requirements', CONFIGURATIONS_WORKER)
    136 
    +def test_configure_hardware_fail(requirements):
    137 
    +    hardware_interface = HardwareInterface(Worker())
    138 
    +
    139 
    +    worker_requirements = Worker(configs=requirements)
    140 
    +
    141 
    +    with pytest.raises(FailedPreconditionError):
    142 
    +        hardware_interface.configure_hardware(worker_requirements.get_pb2())

  • tests/integration/operations_service.py
    ... ... @@ -28,10 +28,8 @@ from buildgrid._enums import OperationStage
    28 28 
     from buildgrid._exceptions import InvalidArgumentError
    29 29 
     from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
    30 30 
     from buildgrid._protos.google.longrunning import operations_pb2
    31 
    -from buildgrid._protos.google.rpc import status_pb2
    32 31 
     from buildgrid.server.cas.storage import lru_memory_cache
    33 32 
     from buildgrid.server.controller import ExecutionController
    34 
    -from buildgrid.server.job import LeaseState
    35 33 
     from buildgrid.server.operations import service
    36 34 
     from buildgrid.server.operations.service import OperationsService
    37 35 
     from buildgrid.utils import create_digest
    ... ... @@ -166,31 +164,6 @@ def test_list_operations_instance_fail(instance, controller, execute_request, co
    166 164 
         context.set_code.assert_called_once_with(grpc.StatusCode.INVALID_ARGUMENT)
    167 165
    168 166
    169 
    -def test_list_operations_with_result(instance, controller, execute_request, context):
    170 
    -    response_execute = controller.execution_instance.execute(execute_request.action_digest,
    171 
    -                                                             execute_request.skip_cache_lookup)
    172 
    -
    173 
    -    action_result = remote_execution_pb2.ActionResult()
    174 
    -    output_file = remote_execution_pb2.OutputFile(path='unicorn')
    175 
    -    action_result.output_files.extend([output_file])
    176 
    -
    177 
    -    controller.operations_instance._scheduler.jobs[response_execute.name].create_lease()
    178 
    -    controller.operations_instance._scheduler.update_job_lease_state(response_execute.name,
    179 
    -                                                                     LeaseState.COMPLETED,
    180 
    -                                                                     lease_status=status_pb2.Status(),
    181 
    -                                                                     lease_result=_pack_any(action_result))
    182 
    -
    183 
    -    request = operations_pb2.ListOperationsRequest(name=instance_name)
    184 
    -    response = instance.ListOperations(request, context)
    185 
    -
    186 
    -    assert response.operations[0].name.split('/')[-1] == response_execute.name
    187 
    -
    188 
    -    execute_response = remote_execution_pb2.ExecuteResponse()
    189 
    -    response.operations[0].response.Unpack(execute_response)
    190 
    -
    191 
    -    assert execute_response.result.output_files == action_result.output_files
    192 
    -
    193 
    -
    194 167 
     def test_list_operations_empty(instance, context):
    195 168 
         request = operations_pb2.ListOperationsRequest(name=instance_name)
    196 169

  • tests/utils/bots_interface.py
    1 
    +# Copyright (C) 2018 Bloomberg LP
    2 
    +#
    3 
    +# Licensed under the Apache License, Version 2.0 (the "License");
    4 
    +# you may not use this file except in compliance with the License.
    5 
    +# You may obtain a copy of the License at
    6 
    +#
    7 
    +#  <http://www.apache.org/licenses/LICENSE-2.0>
    8 
    +#
    9 
    +# Unless required by applicable law or agreed to in writing, software
    10 
    +# distributed under the License is distributed on an "AS IS" BASIS,
    11 
    +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12 
    +# See the License for the specific language governing permissions and
    13 
    +# limitations under the License.
    14 
    +
    15 
    +from concurrent import futures
    16 
    +from contextlib import contextmanager
    17 
    +import multiprocessing
    18 
    +import os
    19 
    +import signal
    20 
    +import uuid
    21 
    +
    22 
    +import grpc
    23 
    +import pytest_cov
    24 
    +
    25 
    +from buildgrid._enums import LeaseState
    26 
    +from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
    27 
    +from buildgrid.server.bots import service
    28 
    +
    29 
    +
    30 
    +@contextmanager
    31 
    +def serve_bots_interface(instances):
    32 
    +    server = Server(instances)
    33 
    +    try:
    34 
    +        yield server
    35 
    +    finally:
    36 
    +        server.quit()
    37 
    +
    38 
    +
    39 
    +# Use subprocess to avoid creation of gRPC threads in main process
    40 
    +# See https://github.com/grpc/grpc/blob/master/doc/fork_support.md
    41 
    +class Server:
    42 
    +
    43 
    +    def __init__(self, instances):
    44 
    +        self.instances = instances
    45 
    +
    46 
    +        self.__queue = multiprocessing.Queue()
    47 
    +        # Queue purely for bot session updates
    48 
    +        self.__bot_session_queue = multiprocessing.Queue()
    49 
    +        # Queue to send messages to subprocess
    50 
    +        self.__message_queue = multiprocessing.Queue()
    51 
    +        self.__process = multiprocessing.Process(
    52 
    +            target=Server.serve,
    53 
    +            args=(self.__queue, self.instances,
    54 
    +                  self.__bot_session_queue, self.__message_queue))
    55 
    +        self.__process.start()
    56 
    +
    57 
    +        self.port = self.__queue.get()
    58 
    +        self.remote = 'localhost:{}'.format(self.port)
    59 
    +
    60 
    +    @staticmethod
    61 
    +    def serve(queue, instances, bot_session_queue, message_queue):
    62 
    +        pytest_cov.embed.cleanup_on_sigterm()
    63 
    +
    64 
    +        # Use max_workers default from Python 3.5+
    65 
    +        max_workers = (os.cpu_count() or 1) * 5
    66 
    +        server = grpc.server(futures.ThreadPoolExecutor(max_workers))
    67 
    +        port = server.add_insecure_port('localhost:0')
    68 
    +
    69 
    +        bots_service = service.BotsService(server)
    70 
    +        for name in instances:
    71 
    +            bots_interface = BotsInterface(bot_session_queue, message_queue)
    72 
    +            bots_service.add_instance(name, bots_interface)
    73 
    +
    74 
    +        server.start()
    75 
    +        queue.put(port)
    76 
    +        signal.pause()
    77 
    +
    78 
    +    def get_bot_session(self, timeout=1):
    79 
    +        bot_session = bots_pb2.BotSession()
    80 
    +        bot_session.ParseFromString(self.__bot_session_queue.get(timeout=timeout))
    81 
    +        return bot_session
    82 
    +
    83 
    +    # Injects leases
    84 
    +    def inject_work(self, lease=None, timeout=1):
    85 
    +        if not lease:
    86 
    +            lease = bots_pb2.Lease()
    87 
    +            lease.state = LeaseState.PENDING.value
    88 
    +
    89 
    +        lease_string = lease.SerializeToString()
    90 
    +        self.__message_queue.put(('INJECT_WORK', lease_string))
    91 
    +
    92 
    +    # Triggers a cancellation of a lease from server
    93 
    +    def cancel_lease(self, lease_id):
    94 
    +        self.__message_queue.put(('CANCEL_LEASE', lease_id))
    95 
    +
    96 
    +    def quit(self):
    97 
    +        if self.__process:
    98 
    +            self.__process.terminate()
    99 
    +            self.__process.join()
    100 
    +
    101 
    +
    102 
    +class BotsInterface:
    103 
    +
    104 
    +    def __init__(self, bot_session_queue, message_queue):
    105 
    +        self.__bot_session_queue = bot_session_queue
    106 
    +        self.__message_queue = message_queue
    107 
    +
    108 
    +    def register_instance_with_server(self, instance_name, server):
    109 
    +        server.add_bots_interface(self, instance_name)
    110 
    +
    111 
    +    def create_bot_session(self, parent, bot_session):
    112 
    +        name = "{}/{}".format(parent, str(uuid.uuid4()))
    113 
    +        bot_session.name = name
    114 
    +
    115 
    +        while not self.__message_queue.empty():
    116 
    +            message = self.__message_queue.get()
    117 
    +            if message[0] == 'INJECT_WORK':
    118 
    +                lease_string = message[1]
    119 
    +                lease = bots_pb2.Lease()
    120 
    +                lease.ParseFromString(lease_string)
    121 
    +                bot_session.leases.extend([lease])
    122 
    +
    123 
    +        self.__bot_session_queue.put(bot_session.SerializeToString())
    124 
    +        return bot_session
    125 
    +
    126 
    +    def update_bot_session(self, name, bot_session):
    127 
    +        for lease in bot_session.leases:
    128 
    +            state = LeaseState(lease.state)
    129 
    +            if state == LeaseState.COMPLETED:
    130 
    +                lease.Clear()
    131 
    +
    132 
    +            elif state == LeaseState.CANCELLED:
    133 
    +                lease.Clear()
    134 
    +
    135 
    +        while not self.__message_queue.empty():
    136 
    +            message = self.__message_queue.get()
    137 
    +
    138 
    +            if message[0] == 'INJECT_WORK':
    139 
    +                lease_string = message[1]
    140 
    +                lease = bots_pb2.Lease()
    141 
    +                lease.ParseFromString(lease_string)
    142 
    +                bot_session.leases.extend([lease])
    143 
    +
    144 
    +            elif message[0] == 'CANCEL_LEASE':
    145 
    +                lease_id = message[1]
    146 
    +                for lease in bot_session.leases:
    147 
    +                    if lease.id == lease_id:
    148 
    +                        lease.state = LeaseState.CANCELLED.value
    149 
    +
    150 
    +        self.__bot_session_queue.put(bot_session.SerializeToString())
    151 
    +        return bot_session

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.

    [Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]