GNOME.org
 

[Notes] [Git][BuildGrid/buildgrid][master] 5 commits: Removed grpc generated files from coverage

Title: GitLab

finnball pushed to branch master at BuildGrid / buildgrid

Commits:

10 changed files:

Changes:

  • .coveragerc
    ... ... @@ -7,6 +7,8 @@ omit =
    7 7 
       # Omit profiling helper module
    8 8 
       # Omit generated code
    9 9 
       */buildgrid/google/*
    10 
    +  *_pb2.py
    11 
    +  *_pb2_grpc.py
    10 12 
       */.eggs/*
    11 13
    12 14 
     [report]

  • app/commands/cmd_bot.py
    ... ... @@ -33,7 +33,8 @@ import tempfile
    33 33
    34 34 
     from pathlib import Path, PurePath
    35 35
    36 
    -from buildgrid.bot import bot
    36 
    +from buildgrid.bot import bot, bot_interface
    37 
    +from buildgrid.bot.bot_session import BotSession, Device, Worker
    37 38 
     from buildgrid._exceptions import BotError
    38 39
    39 40 
     from ..cli import pass_context
    ... ... @@ -44,17 +45,23 @@ from google.protobuf import any_pb2
    44 45
    45 46 
     @click.group(short_help = 'Create a bot client')
    46 47 
     @click.option('--parent', default='bgd_test')
    47 
    -@click.option('--number-of-leases', default=1)
    48 48 
     @click.option('--port', default='50051')
    49 49 
     @click.option('--host', default='localhost')
    50 50 
     @pass_context
    51 
    -def cli(context, host, port, number_of_leases, parent):
    51 
    +def cli(context, host, port, parent):
    52 
    +    channel = grpc.insecure_channel('{}:{}'.format(host, port))
    53 
    +    interface = bot_interface.BotInterface(channel)
    54 
    +
    52 55 
         context.logger = logging.getLogger(__name__)
    53 56 
         context.logger.info("Starting on port {}".format(port))
    54 57
    55 
    -    context.channel = grpc.insecure_channel('{}:{}'.format(host, port))
    56 
    -    context.number_of_leases = number_of_leases
    57 
    -    context.parent = parent
    58 
    +    worker = Worker()
    59 
    +    worker.add_device(Device())
    60 
    +
    61 
    +    bot_session = BotSession(parent, interface)
    62 
    +    bot_session.add_worker(worker)
    63 
    +
    64 
    +    context.bot_session = bot_session
    58 65
    59 66 
     @cli.command('dummy', short_help='Create a dummy bot session')
    60 67 
     @pass_context
    ... ... @@ -63,15 +70,10 @@ def dummy(context):
    63 70 
         Simple dummy client. Creates a session, accepts leases, does fake work and
    64 71 
         updates the server.
    65 72 
         """
    66 
    -
    67 
    -    context.logger.info("Creating a bot session")
    68 
    -
    69 73 
         try:
    70 
    -        bot.Bot(work=_work_dummy,
    71 
    -                context=context,
    72 
    -                channel=context.channel,
    73 
    -                parent=context.parent,
    74 
    -                number_of_leases=context.number_of_leases)
    74 
    +        b = bot.Bot(context.bot_session)
    75 
    +        b.session(_work_dummy,
    76 
    +                  context)
    75 77
    76 78 
         except KeyboardInterrupt:
    77 79 
             pass
    ... ... @@ -85,7 +87,7 @@ def dummy(context):
    85 87 
     @click.option('--port', show_default = True, default=11001)
    86 88 
     @click.option('--remote', show_default = True, default='localhost')
    87 89 
     @pass_context
    88 
    -def _work_buildbox(context, remote, port, server_cert, client_key, client_cert, local_cas, fuse_dir):
    90 
    +def work_buildbox(context, remote, port, server_cert, client_key, client_cert, local_cas, fuse_dir):
    89 91 
         """
    90 92 
         Uses BuildBox to run commands.
    91 93 
         """
    ... ... @@ -101,11 +103,14 @@ def _work_buildbox(context, remote, port, server_cert, client_key, client_cert,
    101 103 
         context.fuse_dir = fuse_dir
    102 104
    103 105 
         try:
    104 
    -        bot.Bot(work=_work_buildbox,
    105 
    -                context=context,
    106 
    -                channel=context.channel,
    107 
    -                parent=context.parent,
    108 
    -                number_of_leases=context.number_of_leases)
    106 
    +        b = bot.Bot(work=_work_buildbox,
    107 
    +                    bot_session=context.bot_session,
    108 
    +                    channel=context.channel,
    109 
    +                    parent=context.parent)
    110 
    +
    111 
    +        b.session(context.parent,
    112 
    +                  _work_buildbox,
    113 
    +                  context)
    109 114
    110 115 
         except KeyboardInterrupt:
    111 116 
             pass

  • buildgrid/bot/bot.py
    ... ... @@ -23,163 +23,45 @@ Creates a bot session.
    23 23 
     """
    24 24
    25 25 
     import asyncio
    26 
    -import inspect
    26 
    +import collections
    27 27 
     import logging
    28 
    -import platform
    29 
    -import queue
    30 28 
     import time
    31 29
    32 
    -from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2, worker_pb2
    33 
    -
    34 
    -from . import bot_interface
    30 
    +from . import bot_interface, bot_session
    31 
    +from .bot_session import BotStatus, LeaseState
    35 32 
     from .._exceptions import BotError
    36 33
    37 
    -class Bot(object):
    34 
    +class Bot:
    38 35 
         """
    39 36 
         Creates a local BotSession.
    40 37 
         """
    41 38
    42 
    -    def __init__(self, work, context, channel, parent, number_of_leases):
    43 
    -        if not inspect.iscoroutinefunction(work):
    44 
    -            raise BotError("work function must be async")
    45 
    -
    46 
    -        self.interface = bot_interface.BotInterface(channel)
    39 
    +    def __init__(self, bot_session, update_period=1):
    47 40 
             self.logger = logging.getLogger(__name__)
    48 41
    49 
    -        self._create_session(parent, number_of_leases)
    50 
    -        self._work_queue = queue.Queue(maxsize = number_of_leases)
    51 
    -
    52 
    -        try:
    53 
    -            while True:
    54 
    -                ## TODO: Leases independently finish
    55 
    -                ## Allow leases to queue finished work independently instead
    56 
    -                ## of waiting for all to finish
    57 
    -                futures = [self._do_work(work, context, lease) for lease in self._get_work()]
    58 
    -                if futures:
    59 
    -                    loop = asyncio.new_event_loop()
    60 
    -                    leases_complete, _ = loop.run_until_complete(asyncio.wait(futures))
    61 
    -                    work_complete = [(lease.result().id, lease.result(),) for lease in leases_complete]
    62 
    -                    self._work_complete(work_complete)
    63 
    -                    loop.close()
    64 
    -                self._update_bot_session()
    65 
    -                time.sleep(2)
    66 
    -        except Exception as e:
    67 
    -            self.logger.error(e)
    68 
    -            raise BotError(e)
    69 
    -
    70 
    -    @property
    71 
    -    def bot_session(self):
    72 
    -        ## Read only, shouldn't have to set any of the variables in here
    73 
    -        return self._bot_session
    74 
    -
    75 
    -    def close_session(self):
    76 
    -        self.logger.warning("Session closing not yet implemented")
    77 
    -
    78 
    -    async def _do_work(self, work, context, lease):
    79 
    -        """ Work is done here, work function should be asynchronous
    80 
    -        """
    81 
    -        self.logger.info("Work found: {}".format(lease.id))
    82 
    -        lease = await work(context=context, lease=lease)
    83 
    -        lease.state = bots_pb2.LeaseState.Value('COMPLETED')
    84 
    -        self.logger.info("Work complete: {}".format(lease.id))
    85 
    -        return lease
    86 
    -
    87 
    -    def _update_bot_session(self):
    88 
    -        """ Should call the server periodically to inform the server the client
    89 
    -        has not died.
    90 
    -        """
    91 
    -        self.logger.debug("Updating bot session")
    92 
    -        self._bot_session = self.interface.update_bot_session(self._bot_session)
    93 
    -        leases_update = ([self._update_lease(lease) for lease in self._bot_session.leases])
    94 
    -        del self._bot_session.leases[:]
    95 
    -        self._bot_session.leases.extend(leases_update)
    96 
    -
    97 
    -    def _get_work(self):
    98 
    -        while not self._work_queue.empty():
    99 
    -            yield self._work_queue.get()
    100 
    -            self._work_queue.task_done()
    101 
    -
    102 
    -    def _work_complete(self, leases_complete):
    103 
    -        """ Bot updates itself with any completed work.
    104 
    -        """
    105 
    -        # Should really improve this...
    106 
    -        # Maybe add some call back function sentoff work...
    107 
    -        leases_active = list(filter(self._lease_active, self._bot_session.leases))
    108 
    -        leases_not_active = [lease for lease in self._bot_session.leases if not self._lease_active(lease)]
    109 
    -        del self._bot_session.leases[:]
    110 
    -        for lease in leases_active:
    111 
    -            for lease_tuple in leases_complete:
    112 
    -                if lease.id == lease_tuple[0]:
    113 
    -                    leases_not_active.extend([lease_tuple[1]])
    114 
    -        self._bot_session.leases.extend(leases_not_active)
    115 
    -
    116 
    -    def _update_lease(self, lease):
    117 
    -        """
    118 
    -        State machine for any recieved updates to the leases.
    119 
    -        """
    120 
    -        if self._lease_pending(lease):
    121 
    -            lease.state = bots_pb2.LeaseState.Value('ACTIVE')
    122 
    -            self._work_queue.put(lease)
    123 
    -            return lease
    124 
    -
    125 
    -        else:
    126 
    -            return lease
    127 
    -
    128 
    -    def _create_session(self, parent, number_of_leases):
    129 
    -        self.logger.debug("Creating bot session")
    130 
    -        worker = self._create_worker()
    42 
    +        self._bot_session = bot_session
    43 
    +        self._update_period = update_period
    131 44
    132 
    -        """ Unique bot ID within the farm used to identify this bot
    133 
    -        Needs to be human readable.
    134 
    -        All prior sessions with bot_id of same ID are invalidated.
    135 
    -        If a bot attempts to update an invalid session, it must be rejected and
    136 
    -        may be put in quarantine.
    137 
    -        """
    138 
    -        bot_id = '{}.{}'.format(parent, platform.node())
    45 
    +    def session(self, work, context):
    46 
    +        loop = asyncio.get_event_loop()
    139 47
    140 
    -        leases = [bots_pb2.Lease() for x in range(number_of_leases)]
    48 
    +        self._bot_session.create_bot_session(work, context)
    141 49
    142 
    -        bot_session = bots_pb2.BotSession(worker = worker,
    143 
    -                                          status = bots_pb2.BotStatus.Value('OK'),
    144 
    -                                          leases = leases,
    145 
    -                                          bot_id = bot_id)
    146 
    -        self._bot_session = self.interface.create_bot_session(parent, bot_session)
    147 
    -        self.logger.info("Name: {}, Id: {}".format(self._bot_session.name,
    148 
    -                                                      self._bot_session.bot_id))
    149 
    -
    150 
    -    def _create_worker(self):
    151 
    -        devices = self._create_devices()
    152 
    -
    153 
    -        # Contains a list of devices and the connections between them.
    154 
    -        worker = worker_pb2.Worker(devices = devices)
    155 
    -
    156 
    -        """ Keys supported:
    157 
    -        *pool
    158 
    -        """
    159 
    -        worker.Property.key = "pool"
    160 
    -        worker.Property.value = "all"
    161 
    -
    162 
    -        return worker
    163 
    -
    164 
    -    def _create_devices(self):
    165 
    -        """ Creates devices available to the worker
    166 
    -        The first device is know as the Primary Device - the revice which
    167 
    -        is running a bit and responsible to actually executing commands.
    168 
    -        All other devices are known as Attatched Devices and must be controlled
    169 
    -        by the Primary Device.
    170 
    -        """
    171 
    -
    172 
    -        devices = []
    173 
    -
    174 
    -        for i in range(0, 1): # Append one device for now
    175 
    -            dev = worker_pb2.Device()
    176 
    -
    177 
    -            devices.append(dev)
    178 
    -
    179 
    -        return devices
    180 
    -
    181 
    -    def _lease_pending(self, lease):
    182 
    -        return lease.state == bots_pb2.LeaseState.Value('PENDING')
    183 
    -
    184 
    -    def _lease_active(self, lease):
    185 
    -        return lease.state == bots_pb2.LeaseState.Value('ACTIVE')
    50 
    +        try:
    51 
    +            task = asyncio.ensure_future(self._update_bot_session())
    52 
    +            loop.run_forever()
    53 
    +
    54 
    +        except KeyboardInterrupt:
    55 
    +            pass
    56 
    +
    57 
    +        finally:
    58 
    +            task.cancel()
    59 
    +            loop.close()
    60 
    +
    61 
    +    async def _update_bot_session(self):
    62 
    +        while True:
    63 
    +            """ Calls the server periodically to inform the server the client
    64 
    +            has not died.
    65 
    +            """
    66 
    +            self._bot_session.update_bot_session()
    67 
    +            await asyncio.sleep(self._update_period)

  • buildgrid/bot/bot_interface.py
    ... ... @@ -29,7 +29,7 @@ from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2, bo
    29 29
    30 30 
     from .._exceptions import BotError
    31 31
    32 
    -class BotInterface(object):
    32 
    +class BotInterface:
    33 33 
         """ Interface handles calls to the server.
    34 34 
         """
    35 35
    ... ... @@ -39,22 +39,12 @@ class BotInterface(object):
    39 39 
             self._stub = bots_pb2_grpc.BotsStub(channel)
    40 40
    41 41 
         def create_bot_session(self, parent, bot_session):
    42 
    -        try:
    43 
    -            request = bots_pb2.CreateBotSessionRequest(parent = parent,
    44 
    -                                                       bot_session = bot_session)
    45 
    -            return self._stub.CreateBotSession(request)
    46 
    -
    47 
    -        except Exception as e:
    48 
    -            self.logger.error(e)
    49 
    -            raise BotError(e)
    42 
    +        request = bots_pb2.CreateBotSessionRequest(parent = parent,
    43 
    +                                                   bot_session = bot_session)
    44 
    +        return self._stub.CreateBotSession(request)
    50 45
    51 46 
         def update_bot_session(self, bot_session, update_mask = None):
    52 
    -        try:
    53 
    -            request = bots_pb2.UpdateBotSessionRequest(name = bot_session.name,
    54 
    -                                                       bot_session = bot_session,
    55 
    -                                                       update_mask = update_mask)
    56 
    -            return self._stub.UpdateBotSession(request)
    57 
    -
    58 
    -        except Exception as e:
    59 
    -            self.logger.error(e)
    60 
    -            raise BotError(e)
    47 
    +        request = bots_pb2.UpdateBotSessionRequest(name = bot_session.name,
    48 
    +                                                   bot_session = bot_session,
    49 
    +                                                   update_mask = update_mask)
    50 
    +        return self._stub.UpdateBotSession(request)

  • buildgrid/bot/bot_session.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 
    +"""
    16 
    +Bot Session
    17 
    +====
    18 
    +
    19 
    +Allows connections
    20 
    +"""
    21 
    +import asyncio
    22 
    +import logging
    23 
    +import platform
    24 
    +import uuid
    25 
    +
    26 
    +from enum import Enum
    27 
    +
    28 
    +from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2, worker_pb2
    29 
    +
    30 
    +class BotStatus(Enum):
    31 
    +    BOT_STATUS_UNSPECIFIED = bots_pb2.BotStatus.Value('BOT_STATUS_UNSPECIFIED')
    32 
    +    OK                     = bots_pb2.BotStatus.Value('OK')
    33 
    +    UNHEALTHY              = bots_pb2.BotStatus.Value('UNHEALTHY');
    34 
    +    HOST_REBOOTING         = bots_pb2.BotStatus.Value('HOST_REBOOTING')
    35 
    +    BOT_TERMINATING        = bots_pb2.BotStatus.Value('BOT_TERMINATING')
    36 
    +
    37 
    +class LeaseState(Enum):
    38 
    +    LEASE_STATE_UNSPECIFIED = bots_pb2.LeaseState.Value('LEASE_STATE_UNSPECIFIED')
    39 
    +    PENDING                 = bots_pb2.LeaseState.Value('PENDING')
    40 
    +    ACTIVE                  = bots_pb2.LeaseState.Value('ACTIVE')
    41 
    +    COMPLETED               = bots_pb2.LeaseState.Value('COMPLETED')
    42 
    +    CANCELLED               = bots_pb2.LeaseState.Value('CANCELLED')
    43 
    +
    44 
    +
    45 
    +class BotSession:
    46 
    +    def __init__(self, parent, interface):
    47 
    +        """ Unique bot ID within the farm used to identify this bot
    48 
    +        Needs to be human readable.
    49 
    +        All prior sessions with bot_id of same ID are invalidated.
    50 
    +        If a bot attempts to update an invalid session, it must be rejected and
    51 
    +        may be put in quarantine.
    52 
    +        """
    53 
    +
    54 
    +        self.logger = logging.getLogger(__name__)
    55 
    +
    56 
    +        self._bot_id = '{}.{}'.format(parent, platform.node())
    57 
    +        self._interface = interface
    58 
    +        self._leases = {}
    59 
    +        self._name = None
    60 
    +        self._parent = parent
    61 
    +        self._status = BotStatus.OK.value
    62 
    +        self._work = None
    63 
    +        self._worker = None
    64 
    +
    65 
    +    @property
    66 
    +    def bot_id(self):
    67 
    +        return self._bot_id
    68 
    +
    69 
    +    def add_worker(self, worker):
    70 
    +        self._worker = worker
    71 
    +
    72 
    +    def create_bot_session(self, work, context=None):
    73 
    +        self.logger.debug("Creating bot session")
    74 
    +        self._work = work
    75 
    +        self._context = context
    76 
    +
    77 
    +        session = self._interface.create_bot_session(self._parent, self.get_pb2())
    78 
    +        self._name = session.name
    79 
    +
    80 
    +        self.logger.info("Created bot session with name: {}".format(self._name))
    81 
    +
    82 
    +        for lease in session.leases:
    83 
    +            self._update_lease_from_server(lease)
    84 
    +
    85 
    +    def update_bot_session(self):
    86 
    +        session = self._interface.update_bot_session(self.get_pb2())
    87 
    +        for lease in session.leases:
    88 
    +            self._update_lease_from_server(lease)
    89 
    +
    90 
    +        for k, v in self._leases.items():
    91 
    +            if v.state == LeaseState.COMPLETED.value:
    92 
    +                del self._leases[k]
    93 
    +
    94 
    +    def get_pb2(self):
    95 
    +        leases = list(self._leases.values())
    96 
    +        if not leases:
    97 
    +            leases = None
    98 
    +
    99 
    +        return bots_pb2.BotSession(worker=self._worker.get_pb2(),
    100 
    +                                   status=self._status,
    101 
    +                                   leases=leases,
    102 
    +                                   bot_id=self._bot_id,
    103 
    +                                   name = self._name)
    104 
    +
    105 
    +    def lease_completed(self, lease):
    106 
    +        lease.state = LeaseState.COMPLETED.value
    107 
    +        self._leases[lease.id] = lease
    108 
    +
    109 
    +    def _update_lease_from_server(self, lease):
    110 
    +        """
    111 
    +        State machine for any recieved updates to the leases.
    112 
    +        """
    113 
    +        ## TODO: Compare with previous state of lease
    114 
    +        lease_bot = self._leases.get(lease.id)
    115 
    +        if lease.state == LeaseState.PENDING.value:
    116 
    +            lease.state = LeaseState.ACTIVE.value
    117 
    +            asyncio.ensure_future(self.create_work(lease))
    118 
    +            self._leases[lease.id] = lease
    119 
    +
    120 
    +    async def create_work(self, lease):
    121 
    +        self.logger.debug("Work created: {}".format(lease.id))
    122 
    +        lease = await self._work(self._context, lease)
    123 
    +        self.logger.debug("Work complete: {}".format(lease.id))
    124 
    +        self.lease_completed(lease)
    125 
    +
    126 
    +class Worker:
    127 
    +    def __init__(self, properties=None, configs=None):
    128 
    +        self.properties = {}
    129 
    +        self._configs = {}
    130 
    +        self._devices = []
    131 
    +
    132 
    +        if properties:
    133 
    +            for k, v in properties.items():
    134 
    +                if k == 'pool':
    135 
    +                    self.properties[k] = v
    136 
    +                else:
    137 
    +                    raise KeyError('Key not supported: {}'.format(k))
    138 
    +
    139 
    +        if configs:
    140 
    +            for k, v in configs.items():
    141 
    +                if k == 'DockerImage':
    142 
    +                    self.configs[k] = v
    143 
    +                else:
    144 
    +                    raise KeyError('Key not supported: {}'.format(k))
    145 
    +
    146 
    +    @property
    147 
    +    def configs(self):
    148 
    +        return self._configs
    149 
    +
    150 
    +    def add_device(self, device):
    151 
    +        self._devices.append(device)
    152 
    +
    153 
    +    def get_pb2(self):
    154 
    +        devices = [device.get_pb2() for device in self._devices]
    155 
    +        worker = worker_pb2.Worker(devices=devices)
    156 
    +        property_message = worker_pb2.Worker.Property()
    157 
    +        for k, v in self.properties.items():
    158 
    +            property_message.key = k
    159 
    +            property_message.value = v
    160 
    +            worker.properties.extend([property_message])
    161 
    +
    162 
    +        config_message = worker_pb2.Worker.Config()
    163 
    +        for k, v in self.properties.items():
    164 
    +            property_message.key = k
    165 
    +            property_message.value = v
    166 
    +            worker.configs.extend([config_message])
    167 
    +
    168 
    +        return worker
    169 
    +
    170 
    +class Device:
    171 
    +    def __init__(self, properties=None):
    172 
    +        """ Creates devices available to the worker
    173 
    +        The first device is know as the Primary Device - the revice which
    174 
    +        is running a bit and responsible to actually executing commands.
    175 
    +        All other devices are known as Attatched Devices and must be controlled
    176 
    +        by the Primary Device.
    177 
    +        """
    178 
    +
    179 
    +        self._name = str(uuid.uuid4())
    180 
    +        self._properties = {}
    181 
    +
    182 
    +        if properties:
    183 
    +            for k, v in properties.items():
    184 
    +                if k == 'os':
    185 
    +                    self._properties[k] = v
    186 
    +
    187 
    +                elif k == 'docker':
    188 
    +                    if v not in ('True', 'False'):
    189 
    +                        raise ValueError('Value not supported: {}'.format(v))
    190 
    +                    self._properties[k] = v
    191 
    +
    192 
    +                else:
    193 
    +                    raise KeyError('Key not supported: {}'.format(k))
    194 
    +
    195 
    +    @property
    196 
    +    def name(self):
    197 
    +        return self._name
    198 
    +
    199 
    +    @property
    200 
    +    def properties(self):
    201 
    +        return self._properties
    202 
    +
    203 
    +
    204 
    +    def get_pb2(self):
    205 
    +        device = worker_pb2.Device(handle=self._name)
    206 
    +        property_message = worker_pb2.Device.Property()
    207 
    +        for k, v in self._properties.items():
    208 
    +            property_message.key = k
    209 
    +            property_message.value = v
    210 
    +            device.properties.extend([property_message])
    211 
    +        return device

  • buildgrid/server/scheduler.py
    ... ... @@ -50,24 +50,17 @@ class Scheduler():
    50 50 
             self.queue.append(job)
    51 51
    52 52 
         def retry_job(self, name):
    53 
    -        job = self.jobs[name]
    54 
    -
    55 
    -        if job.n_tries >= self.MAX_N_TRIES:
    56 
    -            # TODO: Decide what to do with these jobs
    57 
    -            job.update_execute_stage(ExecuteStage.COMPLETED)
    58 
    -        else:
    59 
    -            job.update_execute_stage(ExecuteStage.QUEUED)
    60 
    -            job.n_tries += 1
    61 
    -            self.queue.appendleft(job)
    62 
    -
    63 
    -        self.jobs[name] = job
    53 
    +        if job in self.jobs[name]:
    54 
    +            if job.n_tries >= self.MAX_N_TRIES:
    55 
    +                # TODO: Decide what to do with these jobs
    56 
    +                job.update_execute_stage(ExecuteStage.COMPLETED)
    57 
    +                # TODO: Mark these jobs as done
    58 
    +            else:
    59 
    +                job.update_execute_stage(ExecuteStage.QUEUED)
    60 
    +                job.n_tries += 1
    61 
    +                self.queue.appendleft(job)
    64 62
    65 
    -    def create_job(self):
    66 
    -        if len(self.queue) > 0:
    67 
    -            job = self.queue.popleft()
    68 
    -            job.update_execute_stage(ExecuteStage.EXECUTING)
    69 
    -            self.jobs[job.name] = job
    70 
    -            return job
    63 
    +            self.jobs[name] = job
    71 64
    72 65 
         def job_complete(self, name, result):
    73 66 
             job = self.jobs[name]
    ... ... @@ -81,48 +74,13 @@ class Scheduler():
    81 74 
                 response.operations.extend([v.get_operation()])
    82 75 
             return response
    83 76
    84 
    -    def update_lease(self, lease):
    85 
    -        name = lease.id
    77 
    +    def update_job_lease_state(self, name, state):
    86 78 
             job = self.jobs.get(name)
    87 
    -        state = lease.state
    88 
    -
    89 
    -        if state   == LeaseState.LEASE_STATE_UNSPECIFIED.value:
    90 
    -            create_job = self.create_job()
    91 
    -            if create_job is None:
    92 
    -                # No job? Return lease.
    93 
    -                return lease
    94 
    -            else:
    95 
    -                job = create_job
    96 
    -                job.create_lease()
    97 
    -
    98 
    -        elif state == LeaseState.PENDING.value:
    99 
    -            job.lease = lease
    100 
    -
    101 
    -        elif state == LeaseState.ACTIVE.value:
    102 
    -            job.lease = lease
    103 
    -
    104 
    -        elif state == LeaseState.COMPLETED.value:
    105 
    -            self.job_complete(job.name, lease.result)
    106 
    -
    107 
    -            create_job = self.create_job()
    108 
    -            if create_job is None:
    109 
    -                # Docs say not to use this state though if job has
    110 
    -                # completed and no more jobs, then use this state to stop
    111 
    -                # job being processed again
    112 
    -                job.lease = lease
    113 
    -                job.lease.state = LeaseState.LEASE_STATE_UNSPECIFIED.value
    114 
    -            else:
    115 
    -                job = create_job
    116 
    -                job.lease = job.create_lease()
    117 
    -
    118 
    -        elif state == LeaseState.CANCELLED.value:
    119 
    -            job.lease = lease
    120 
    -
    121 
    -        else:
    122 
    -            raise Exception("Unknown state: {}".format(state))
    123 
    -
    79 
    +        job.lease.state = state
    124 80 
             self.jobs[name] = job
    125 
    -        return job.lease
    81 
    +
    82 
    +    def get_job_lease(self, name):
    83 
    +        return self.jobs[name].lease
    126 84
    127 85 
         def cancel_session(self, name):
    128 86 
             job = self.jobs[name]
    ... ... @@ -130,3 +88,12 @@ class Scheduler():
    130 88 
             if state == LeaseState.PENDING.value or \
    131 89 
                state == LeaseState.ACTIVE.value:
    132 90 
                 self.retry_job(name)
    91 
    +
    92 
    +    def create_leases(self):
    93 
    +        while len(self.queue) > 0:
    94 
    +            job = self.queue.popleft()
    95 
    +            job.update_execute_stage(ExecuteStage.EXECUTING)
    96 
    +            job.lease = job.create_lease()
    97 
    +            job.lease.state = LeaseState.PENDING.value
    98 
    +            self.jobs[job.name] = job
    99 
    +            yield job.lease

  • buildgrid/server/worker/bots_interface.py
    ... ... @@ -35,6 +35,7 @@ class BotsInterface():
    35 35 
             self.logger = logging.getLogger(__name__)
    36 36
    37 37 
             self._bot_ids = {}
    38 
    +        self._bot_sessions = {}
    38 39 
             self._scheduler = scheduler
    39 40
    40 41 
         def create_bot_session(self, parent, bot_session):
    ... ... @@ -59,7 +60,12 @@ class BotsInterface():
    59 60 
             bot_session.name = name
    60 61
    61 62 
             self._bot_ids[name] = bot_id
    63 
    +        self._bot_sessions[name] = bot_session
    62 64 
             self.logger.info("Created bot session name={} with bot_id={}".format(name, bot_id))
    65 
    +
    66 
    +        for lease in self._scheduler.create_leases():
    67 
    +            bot_session.leases.extend([lease])
    68 
    +
    63 69 
             return bot_session
    64 70
    65 71 
         def update_bot_session(self, name, bot_session):
    ... ... @@ -69,13 +75,66 @@ class BotsInterface():
    69 75 
             self.logger.debug("Updating bot session name={}".format(name))
    70 76 
             self._check_bot_ids(bot_session.bot_id, name)
    71 77
    72 
    -        leases = [self._scheduler.update_lease(lease) for lease in bot_session.leases]
    78 
    +        server_session = self._bot_sessions[name]
    79 
    +
    80 
    +        leases = filter(None, [self.check_states(lease) for lease in bot_session.leases])
    73 81
    74 82 
             del bot_session.leases[:]
    75 83 
             bot_session.leases.extend(leases)
    76 84
    85 
    +        for lease in self._scheduler.create_leases():
    86 
    +            bot_session.leases.extend([lease])
    87 
    +
    88 
    +        self._bot_sessions[name] = bot_session
    77 89 
             return bot_session
    78 90
    91 
    +    def check_states(self, client_lease):
    92 
    +        """ Edge detector for states
    93 
    +        """
    94 
    +        ## TODO: Handle cancelled states
    95 
    +        try:
    96 
    +            server_lease = self._scheduler.get_job_lease(client_lease.id)
    97 
    +        except KeyError:
    98 
    +            raise InvalidArgumentError("Lease not found on server: {}".format(client_lease))
    99 
    +
    100 
    +        server_state = LeaseState(server_lease.state)
    101 
    +        client_state = LeaseState(client_lease.state)
    102 
    +
    103 
    +        if server_state == LeaseState.PENDING:
    104 
    +
    105 
    +            if client_state == LeaseState.ACTIVE:
    106 
    +                self._scheduler.update_job_lease_state(client_lease.id, client_lease.state)
    107 
    +            elif client_state == LeaseState.COMPLETED:
    108 
    +                # TODO: Lease was rejected
    109 
    +                raise NotImplementedError("'Not Accepted' is unsupported")
    110 
    +            else:
    111 
    +                raise OutofSyncError("Server lease: {}. Client lease: {}".format(server_lease, client_lease))
    112 
    +
    113 
    +        elif server_state == LeaseState.ACTIVE:
    114 
    +
    115 
    +            if client_state == LeaseState.ACTIVE:
    116 
    +                pass
    117 
    +
    118 
    +            elif client_state == LeaseState.COMPLETED:
    119 
    +                self._scheduler.job_complete(client_lease.id, client_lease.result)
    120 
    +                self._scheduler.update_job_lease_state(client_lease.id, client_lease.state)
    121 
    +                return None
    122 
    +
    123 
    +            else:
    124 
    +                raise OutofSyncError("Server lease: {}. Client lease: {}".format(server_lease, client_lease))
    125 
    +
    126 
    +        elif server_state == LeaseState.COMPLETED:
    127 
    +            raise OutofSyncError("Server lease: {}. Client lease: {}".format(server_lease, client_lease))
    128 
    +
    129 
    +        elif server_state == LeaseState.CANCELLED:
    130 
    +            raise NotImplementedError("Cancelled states not supported yet")
    131 
    +
    132 
    +        else:
    133 
    +            # Sould never get here
    134 
    +            raise OutofSyncError("State now allowed: {}".format(server_state))
    135 
    +
    136 
    +        return client_lease
    137 
    +
    79 138 
         def _check_bot_ids(self, bot_id, name = None):
    80 139 
             """ Checks the ID and the name of the bot.
    81 140 
             """
    ... ... @@ -103,7 +162,11 @@ class BotsInterface():
    103 162 
                 raise InvalidArgumentError("Bot id does not exist: {}".format(name))
    104 163
    105 164 
             self.logger.debug("Attempting to close {} with name: {}".format(bot_id, name))
    106 
    -        self._scheduler.retry_job(name)
    165 
    +        for lease in self._bot_sessions[name].leases:
    166 
    +            if lease.state != LeaseState.COMPLETED.value:
    167 
    +                # TODO: Be wary here, may need to handle rejected leases in future
    168 
    +                self._scheduler.retry_job(lease.id)
    169 
    +
    107 170 
             self.logger.debug("Closing bot session: {}".format(name))
    108 171 
             self._bot_ids.pop(name)
    109 172 
             self.logger.info("Closed bot {} with name: {}".format(bot_id, name))

  • buildgrid/server/worker/bots_service.py
    ... ... @@ -43,7 +43,6 @@ class BotsService(bots_pb2_grpc.BotsServicer):
    43 43 
                 self.logger.error(e)
    44 44 
                 context.set_details(str(e))
    45 45 
                 context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
    46 
    -            return bots_pb2.BotSession()
    47 46
    48 47 
         def UpdateBotSession(self, request, context):
    49 48 
             try:
    ... ... @@ -53,13 +52,16 @@ class BotsService(bots_pb2_grpc.BotsServicer):
    53 52 
                 self.logger.error(e)
    54 53 
                 context.set_details(str(e))
    55 54 
                 context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
    56 
    -            return bots_pb2.BotSession()
    57 55
    58 56 
             except OutofSyncError as e:
    59 57 
                 self.logger.error(e)
    60 58 
                 context.set_details(str(e))
    61 59 
                 context.set_code(grpc.StatusCode.DATA_LOSS)
    62 
    -            return bots_pb2.BotSession()
    60 
    +
    61 
    +        except NotImplementedError as e:
    62 
    +            self.logger.error(e)
    63 
    +            context.set_details(str(e))
    64 
    +            context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    63 65
    64 66 
         def PostBotEventTemp(self, request, context):
    65 67 
             context.set_code(grpc.StatusCode.UNIMPLEMENTED)

  • tests/integration/bot_session.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 
    +import grpc
    16 
    +import pytest
    17 
    +import uuid
    18 
    +
    19 
    +from unittest import mock
    20 
    +
    21 
    +from buildgrid.bot import bot_session, bot_interface
    22 
    +
    23 
    +@pytest.mark.parametrize("docker_value", ["True", "False"])
    24 
    +@pytest.mark.parametrize("os_value", ["nexus7", "nexus8"])
    25 
    +def test_create_device(docker_value, os_value):
    26 
    +    properties = {'docker' : docker_value, 'os' : os_value}
    27 
    +    device = bot_session.Device(properties)
    28 
    +
    29 
    +    assert uuid.UUID(device.name, version=4)
    30 
    +    assert properties == device.properties
    31 
    +
    32 
    +def test_create_device_key_fail():
    33 
    +    properties = {'voight' : 'kampff'}
    34 
    +
    35 
    +    with pytest.raises(KeyError):
    36 
    +        device = bot_session.Device(properties)
    37 
    +
    38 
    +def test_create_device_value_fail():
    39 
    +    properties = {'docker' :  True}
    40 
    +
    41 
    +    with pytest.raises(ValueError):
    42 
    +        device = bot_session.Device(properties)
    43 
    +
    44 
    +def test_create_worker():
    45 
    +    properties = {'pool' : 'swim'}
    46 
    +    configs = {'DockerImage' : 'Windows'}
    47 
    +    worker = bot_session.Worker(properties, configs)
    48 
    +
    49 
    +    assert properties == worker.properties
    50 
    +    assert configs == worker.configs
    51 
    +
    52 
    +    device = bot_session.Device()
    53 
    +    worker.add_device(device)
    54 
    +
    55 
    +    assert worker._devices[0] == device
    56 
    +
    57 
    +def test_create_worker_key_fail():
    58 
    +    properties = {'voight' : 'kampff'}
    59 
    +    configs = {'voight' : 'kampff'}
    60 
    +
    61 
    +    with pytest.raises(KeyError):
    62 
    +        bot_session.Worker(properties)
    63 
    +    with pytest.raises(KeyError):
    64 
    +        bot_session.Worker(configs)

  • tests/integration/bots_service.py
    ... ... @@ -36,6 +36,12 @@ from buildgrid.server.worker import bots_interface, bots_service
    36 36 
     def context():
    37 37 
         yield mock.MagicMock(spec = _Context)
    38 38
    39 
    +@pytest.fixture
    40 
    +def action_job():
    41 
    +    action_digest = remote_execution_pb2.Digest()
    42 
    +    j = job.Job(action_digest, None)
    43 
    +    yield j
    44 
    +
    39 45 
     @pytest.fixture
    40 46 
     def bot_session():
    41 47 
         bot = bots_pb2.BotSession()
    ... ... @@ -101,7 +107,6 @@ def test_update_bot_session_zombie(bot_session, context, instance):
    101 107
    102 108 
         response = instance.UpdateBotSession(request, context)
    103 109
    104 
    -    assert isinstance(response, bots_pb2.BotSession)
    105 110 
         context.set_code.assert_called_once_with(grpc.StatusCode.INVALID_ARGUMENT)
    106 111
    107 112 
     def test_update_bot_session_bot_id_fail(bot_session, context, instance):
    ... ... @@ -113,62 +118,121 @@ def test_update_bot_session_bot_id_fail(bot_session, context, instance):
    113 118
    114 119 
         context.set_code.assert_called_once_with(grpc.StatusCode.INVALID_ARGUMENT)
    115 120
    116 
    -@pytest.mark.parametrize("number_of_leases", [1, 3, 500])
    117 
    -def test_update_leases(number_of_leases, bot_session, context, instance):
    118 
    -    leases = [bots_pb2.Lease() for x in range(number_of_leases)]
    119 
    -    bot_session.leases.extend(leases)
    121 
    +@pytest.mark.parametrize("number_of_jobs", [0, 1, 3, 500])
    122 
    +def test_number_of_leases(number_of_jobs, bot_session, context, instance):
    120 123 
         request = bots_pb2.CreateBotSessionRequest(parent='',
    121 124 
                                                    bot_session=bot_session)
    122 
    -    # Simulated the severed binding between client and server
    123 
    -    bot = copy.deepcopy(instance.CreateBotSession(request, context))
    124 
    -
    125 
    -    request = bots_pb2.UpdateBotSessionRequest(name=bot.name,
    126 
    -                                               bot_session=bot)
    125 
    +    # Inject work
    126 
    +    for n in range(0, number_of_jobs):
    127 
    +        action_digest = remote_execution_pb2.Digest()
    128 
    +        instance._instance._scheduler.append_job(job.Job(action_digest))
    127 129
    128 
    -    response = instance.UpdateBotSession(request, context)
    130 
    +    response = instance.CreateBotSession(request, context)
    129 131
    132 
    +    assert len(response.leases) == number_of_jobs
    130 133 
         assert isinstance(response, bots_pb2.BotSession)
    131 
    -    assert len(response.leases) == len(bot.leases)
    132 
    -    assert bot == response
    133 134
    134 135 
     def test_update_leases_with_work(bot_session, context, instance):
    135 
    -    leases = [bots_pb2.Lease() for x in range(2)]
    136 
    -    bot_session.leases.extend(leases)
    136 
    +    request = bots_pb2.CreateBotSessionRequest(parent='',
    137 
    +                                               bot_session=bot_session)
    138 
    +    # Inject work
    139 
    +    action_digest = remote_execution_pb2.Digest(hash = 'gaff')
    140 
    +    instance._instance._scheduler.append_job(job.Job(action_digest))
    141 
    +
    142 
    +    response = instance.CreateBotSession(request, context)
    143 
    +
    144 
    +    assert len(response.leases) == 1
    145 
    +    response_action = remote_execution_pb2.Digest()
    146 
    +    response.leases[0].payload.Unpack(response_action)
    147 
    +
    148 
    +    assert isinstance(response, bots_pb2.BotSession)
    149 
    +    assert response.leases[0].state == LeaseState.PENDING.value
    150 
    +    assert uuid.UUID(response.leases[0].id, version=4)
    151 
    +    assert response_action == action_digest
    152 
    +
    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 
    +    instance._instance._scheduler.append_job(job.Job(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 
    +
    178 
    +    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    179 
    +                                               bot_session=response)
    180 
    +    response = copy.deepcopy(instance.UpdateBotSession(request, context))
    181 
    +
    182 
    +    assert len(response.leases) == 0
    183 
    +
    184 
    +def test_work_rejected_by_bot(bot_session, context, instance):
    185 
    +    request = bots_pb2.CreateBotSessionRequest(parent='',
    186 
    +                                               bot_session=bot_session)
    187 
    +    # Inject work
    188 
    +    action_digest = remote_execution_pb2.Digest(hash = 'gaff')
    189 
    +    instance._instance._scheduler.append_job(job.Job(action_digest))
    190 
    +
    191 
    +    # Simulated the severed binding between client and server
    192 
    +    response = copy.deepcopy(instance.CreateBotSession(request, context))
    193 
    +
    194 
    +    # Reject work
    195 
    +    assert response.leases[0].state == LeaseState.PENDING.value
    196 
    +    response.leases[0].state = LeaseState.COMPLETED.value
    197 
    +    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    198 
    +                                               bot_session=response)
    199 
    +
    200 
    +    response = instance.UpdateBotSession(request, context)
    201 
    +
    202 
    +    context.set_code.assert_called_once_with(grpc.StatusCode.UNIMPLEMENTED)
    137 203
    138 
    -    # Inject some work to be done
    139 
    -    action = remote_execution_pb2.Action()
    140 
    -    action.command_digest.hash = 'rick'
    141 
    -    instance._instance._scheduler.append_job(job.Job(action))
    142 204
    205 
    +@pytest.mark.parametrize("state", [ LeaseState.LEASE_STATE_UNSPECIFIED, LeaseState.PENDING])
    206 
    +def test_work_out_of_sync_from_pending(state, bot_session, context, instance):
    143 207 
         request = bots_pb2.CreateBotSessionRequest(parent='',
    144 208 
                                                    bot_session=bot_session)
    209 
    +    # Inject work
    210 
    +    action_digest = remote_execution_pb2.Digest(hash = 'gaff')
    211 
    +    instance._instance._scheduler.append_job(job.Job(action_digest))
    145 212 
         # Simulated the severed binding between client and server
    146 213 
         bot = copy.deepcopy(instance.CreateBotSession(request, context))
    147 214
    148 215 
         request = bots_pb2.UpdateBotSessionRequest(name=bot.name,
    149 216 
                                                    bot_session=bot)
    150 217
    151 
    -    response = instance.UpdateBotSession(request, context)
    152 
    -    response_action = remote_execution_pb2.Action()
    153 
    -    _unpack_any(response.leases[0].payload, response_action)
    218 
    +    response = copy.deepcopy(instance.UpdateBotSession(request, context))
    154 219
    155 
    -    assert isinstance(response, bots_pb2.BotSession)
    156 
    -    assert response.leases[0].state == LeaseState.PENDING.value
    157 
    -    assert response.leases[1].state == LeaseState.LEASE_STATE_UNSPECIFIED.value
    158 
    -    assert uuid.UUID(response.leases[0].id, version=4)
    159 
    -    assert response_action == action
    220 
    +    response.leases[0].state = state.value
    160 221
    161 
    -def test_update_leases_work_complete(bot_session, context, instance):
    162 
    -    leases = [bots_pb2.Lease() for x in range(2)]
    163 
    -    bot_session.leases.extend(leases)
    222 
    +    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    223 
    +                                               bot_session=response)
    224 
    +
    225 
    +    response = instance.UpdateBotSession(request, context)
    164 226
    165 
    -    # Inject some work to be done
    166 
    -    action = remote_execution_pb2.Action()
    167 
    -    action.command_digest.hash = 'rick'
    168 
    -    instance._instance._scheduler.append_job(job.Job(action))
    227 
    +    context.set_code.assert_called_once_with(grpc.StatusCode.DATA_LOSS)
    169 228
    229 
    +@pytest.mark.parametrize("state", [ LeaseState.LEASE_STATE_UNSPECIFIED, LeaseState.PENDING])
    230 
    +def test_work_out_of_sync_from_active(state, bot_session, context, instance):
    170 231 
         request = bots_pb2.CreateBotSessionRequest(parent='',
    171 232 
                                                    bot_session=bot_session)
    233 
    +    # Inject work
    234 
    +    action_digest = remote_execution_pb2.Digest(hash = 'gaff')
    235 
    +    instance._instance._scheduler.append_job(job.Job(action_digest))
    172 236 
         # Simulated the severed binding between client and server
    173 237 
         bot = copy.deepcopy(instance.CreateBotSession(request, context))
    174 238
    ... ... @@ -177,26 +241,75 @@ def test_update_leases_work_complete(bot_session, context, instance):
    177 241
    178 242 
         response = copy.deepcopy(instance.UpdateBotSession(request, context))
    179 243
    180 
    -    operation_name = response.leases[0].id
    244 
    +    response = instance.UpdateBotSession(request, context)
    181 245
    182 
    -    assert response.leases[0].state == LeaseState.PENDING.value
    183 
    -    response.leases[0].state = LeaseState.COMPLETED.value
    246 
    +    context.set_code.assert_called_once_with(grpc.StatusCode.UNIMPLEMENTED)
    247 
    +
    248 
    +@pytest.mark.parametrize("state", [ LeaseState.LEASE_STATE_UNSPECIFIED, LeaseState.PENDING])
    249 
    +def test_work_out_of_sync_from_pending(state, bot_session, context, instance):
    250 
    +    request = bots_pb2.CreateBotSessionRequest(parent='',
    251 
    +                                               bot_session=bot_session)
    252 
    +    # Inject work
    253 
    +    action_digest = remote_execution_pb2.Digest(hash = 'gaff')
    254 
    +    instance._instance._scheduler.append_job(job.Job(action_digest))
    255 
    +    # Simulated the severed binding between client and server
    256 
    +    response = copy.deepcopy(instance.CreateBotSession(request, context))
    257 
    +
    258 
    +    response.leases[0].state = state.value
    259 
    +
    260 
    +    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    261 
    +                                               bot_session=response)
    262 
    +
    263 
    +    response = instance.UpdateBotSession(request, context)
    264 
    +
    265 
    +    context.set_code.assert_called_once_with(grpc.StatusCode.DATA_LOSS)
    266 
    +
    267 
    +@pytest.mark.parametrize("state", [ LeaseState.LEASE_STATE_UNSPECIFIED, LeaseState.PENDING])
    268 
    +def test_work_out_of_sync_from_active(state, bot_session, context, instance):
    269 
    +    request = bots_pb2.CreateBotSessionRequest(parent='',
    270 
    +                                               bot_session=bot_session)
    271 
    +    # Inject work
    272 
    +    action_digest = remote_execution_pb2.Digest(hash = 'gaff')
    273 
    +    instance._instance._scheduler.append_job(job.Job(action_digest))
    274 
    +    # Simulated the severed binding between client and server
    275 
    +    response = copy.deepcopy(instance.CreateBotSession(request, context))
    276 
    +
    277 
    +    response.leases[0].state = LeaseState.ACTIVE.value
    278 
    +
    279 
    +    request = copy.deepcopy(bots_pb2.UpdateBotSessionRequest(name=response.name,
    280 
    +                                                             bot_session=response))
    281 
    +
    282 
    +    response = instance.UpdateBotSession(request, context)
    283 
    +
    284 
    +    response.leases[0].state = state.value
    184 285
    185 286 
         request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    186 287 
                                                    bot_session=response)
    288 
    +
    289 
    +    response = instance.UpdateBotSession(request, context)
    290 
    +
    291 
    +    context.set_code.assert_called_once_with(grpc.StatusCode.DATA_LOSS)
    292 
    +
    293 
    +def test_work_active_to_active(bot_session, context, instance):
    294 
    +    request = bots_pb2.CreateBotSessionRequest(parent='',
    295 
    +                                               bot_session=bot_session)
    296 
    +    # Inject work
    297 
    +    action_digest = remote_execution_pb2.Digest(hash = 'gaff')
    298 
    +    instance._instance._scheduler.append_job(job.Job(action_digest))
    187 299 
         # Simulated the severed binding between client and server
    188 
    -    response = copy.deepcopy(instance.UpdateBotSession(request, context))
    189 
    -    assert isinstance(response, bots_pb2.BotSession)
    190 
    -    assert instance._instance._scheduler.jobs[operation_name]._execute_stage == ExecuteStage.COMPLETED
    300 
    +    response = copy.deepcopy(instance.CreateBotSession(request, context))
    301 
    +
    302 
    +    response.leases[0].state = LeaseState.ACTIVE.value
    303 
    +
    304 
    +    request = bots_pb2.UpdateBotSessionRequest(name=response.name,
    305 
    +                                               bot_session=response)
    306 
    +
    307 
    +    response = instance.UpdateBotSession(request, context)
    308 
    +
    309 
    +    assert response.leases[0].state == LeaseState.ACTIVE.value
    191 310
    192 311 
     def test_post_bot_event_temp(context, instance):
    193 312 
         request = bots_pb2.PostBotEventTempRequest()
    194 313 
         instance.PostBotEventTemp(request, context)
    195 314
    196 315 
         context.set_code.assert_called_once_with(grpc.StatusCode.UNIMPLEMENTED)
    197 
    -
    198 
    -def _unpack_any(unpack_from, to):
    199 
    -    any = any_pb2.Any()
    200 
    -    any.CopyFrom(unpack_from)
    201 
    -    any.Unpack(to)
    202 
    -    return to

    • —
    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]