.coveragerc

+[run]
+concurrency = multiprocessing
+include =
+  */buildgrid/*
+
+omit =
+  # Omit profiling helper module
+  # Omit generated code
+  */buildgrid/google/*
+  */.eggs/*
+
+[report]
+show_missing = True
+precision = 2
+
+[paths]
+source =
+    buildgrid/
+    */site-packages/buildgrid/
+    */buildgrid/buildgrid/

.gitlab-ci.yml

+image: buildstream/buildstream-fedora:master-81-06ae434
+
+variables:
+  BGD: bgd --verbose
+
+stages:
+  - test
+  - post
+
+before_script:
+  - export PATH=~/.local/bin:${PATH}
+  - pip3 install --user -e .
+
+tests-fedora:
+  stage: test
+  variables:
+    PYTEST_ADDOPTS: "--color=yes"
+  script:
+    - yum -y install clang libffi-devel openssl-devel python3-devel
+    - python3 setup.py test
+    - mkdir -p coverage/
+    - cp .coverage.* coverage/coverage."${CI_JOB_NAME}"
+  artifacts:
+    paths:
+    - coverage/
+
+
+tests-dummy-job-fedora:
+  stage: test
+  script:
+    - ${BGD} server start &
+    - sleep 1 # Allow server to boot
+    - ${BGD} bot --host=0.0.0.0 dummy &
+    - ${BGD} execute --host=0.0.0.0 request --wait-for-completion
+
+coverage:
+  stage: post
+  coverage: '/TOTAL +\d+ +\d+ +(\d+\.\d+)%/'
+  script:
+    - mkdir report
+    - cd report
+    - cp ../coverage/coverage.* .
+    - ls coverage.*
+    - coverage combine --rcfile=../.coveragerc -a coverage.*
+    - coverage report --rcfile=../.coveragerc -m
+  dependencies:
+  - tests-fedora

app/commands/cmd_bot.py

@@ -76,10 +76,6 @@ def dummy(context):
     except KeyboardInterrupt:
         pass
 
-    except Exception as e:
-        context.logger.error(e)
-        return
-
 @cli.command('buildbox', short_help='Create a bot session with busybox')
 @click.option('--fuse-dir', show_default = True, default=str(PurePath(Path.home(), 'fuse')))
 @click.option('--local-cas', show_default = True, default=str(PurePath(Path.home(), 'cas')))
@@ -114,10 +110,6 @@ def _work_buildbox(context, remote, port, server_cert, client_key, client_cert,
     except KeyboardInterrupt:
         pass
 
-    except Exception as e:
-        context.logger.error(e)
-        return
-
 async def _work_dummy(context, lease):
     await asyncio.sleep(random.randint(1,5))
     return lease
@@ -142,51 +134,46 @@ async def _work_buildbox(context, lease):
 
     stub = bytestream_pb2_grpc.ByteStreamStub(channel)
 
-    try:
-        remote_command = _fetch_command(context.local_cas, stub, action.command_digest)
-        environment = dict((x.name, x.value) for x in remote_command.environment_variables)
-        logger.debug("command hash: {}".format(action.command_digest.hash))
-        logger.debug("vdir hash: {}".format(action.input_root_digest.hash))
-        logger.debug("\n{}".format(' '.join(remote_command.arguments)))
+    remote_command = _fetch_command(context.local_cas, stub, action.command_digest)
+    environment = dict((x.name, x.value) for x in remote_command.environment_variables)
+    logger.debug("command hash: {}".format(action.command_digest.hash))
+    logger.debug("vdir hash: {}".format(action.input_root_digest.hash))
+    logger.debug("\n{}".format(' '.join(remote_command.arguments)))
 
-        command = ['buildbox',
-                   '--remote={}'.format('https://{}:{}'.format(context.remote, context.port)),
-                   '--server-cert={}'.format(context.server_cert),
-                   '--client-key={}'.format(context.client_key),
-                   '--client-cert={}'.format(context.client_cert),
-                   '--local={}'.format(context.local_cas),
-                   '--chdir={}'.format(environment['PWD']),
-                   context.fuse_dir,
-                   ]
+    command = ['buildbox',
+               '--remote={}'.format('https://{}:{}'.format(context.remote, context.port)),
+               '--server-cert={}'.format(context.server_cert),
+               '--client-key={}'.format(context.client_key),
+               '--client-cert={}'.format(context.client_cert),
+               '--local={}'.format(context.local_cas),
+               '--chdir={}'.format(environment['PWD']),
+               context.fuse_dir]
 
-        command.extend(remote_command.arguments)
+    command.extend(remote_command.arguments)
 
-        logger.debug(' '.join(command))
-        logger.debug("Input root digest:\n{}".format(action.input_root_digest))
-        logger.info("Launching process")
+    logger.debug(' '.join(command))
+    logger.debug("Input root digest:\n{}".format(action.input_root_digest))
+    logger.info("Launching process")
 
-        proc = subprocess.Popen(command,
-                                stdin=subprocess.PIPE,
-                                stdout=subprocess.PIPE)
-        std_send = action.input_root_digest.SerializeToString()
-        std_out, std_error = proc.communicate(std_send)
+    proc = subprocess.Popen(command,
+                            stdin=subprocess.PIPE,
+                            stdout=subprocess.PIPE)
+    std_send = action.input_root_digest.SerializeToString()
+    std_out, std_error = proc.communicate(std_send)
 
-        output_root_digest = remote_execution_pb2.Digest()
-        output_root_digest.ParseFromString(std_out)
-        logger.debug("Output root digest: {}".format(output_root_digest))
+    output_root_digest = remote_execution_pb2.Digest()
+    output_root_digest.ParseFromString(std_out)
+    logger.debug("Output root digest: {}".format(output_root_digest))
 
-        output_file = remote_execution_pb2.OutputDirectory(tree_digest = output_root_digest)
+    output_file = remote_execution_pb2.OutputDirectory(tree_digest = output_root_digest)
 
-        action_result = remote_execution_pb2.ActionResult()
-        action_result.output_directories.extend([output_file])
+    action_result = remote_execution_pb2.ActionResult()
+    action_result.output_directories.extend([output_file])
 
-        action_result_any = any_pb2.Any()
-        action_result_any.Pack(action_result)
+    action_result_any = any_pb2.Any()
+    action_result_any.Pack(action_result)
 
-        lease.inline_assignment.CopyFrom(action_result_any)
-
-    except Exception as e:
-        raise Exception(e)
+    lease.inline_assignment.CopyFrom(action_result_any)
 
     return lease
 
@@ -210,8 +197,5 @@ def _fetch_command(casdir, remote, digest):
         return remote_command
 
 def _file_read(file_path):
-    try:
-        with open(file_path, 'rb') as f:
-            return f.read()
-    except Exception as e:
-        raise Exception("Error reading: {}. Error: {}".format(file_path, e))
+    with open(file_path, 'rb') as f:
+        return f.read()

app/commands/cmd_execute.py

@@ -25,6 +25,8 @@ Request work to be executed and monitor status of jobs.
 import click
 import grpc
 import logging
+import sys
+import time
 
 from ..cli import pass_context
 
@@ -35,19 +37,21 @@ from google.protobuf import any_pb2
 
 @click.group(short_help = "Simple execute client")
 @click.option('--port', default='50051')
+@click.option('--host', default='localhost')
 @pass_context
-def cli(context, port):
+def cli(context, host, port):
     context.logger = logging.getLogger(__name__)
     context.logger.info("Starting on port {}".format(port))
 
-    context.channel = grpc.insecure_channel('localhost:{}'.format(port))
+    context.channel = grpc.insecure_channel('{}:{}'.format(host, port))
     context.port = port
 
 @cli.command('request', short_help='Send a dummy action')
 @click.option('--number', default=1)
 @click.option('--instance-name', default='testing')
+@click.option('--wait-for-completion', is_flag=True)
 @pass_context
-def request(context, number, instance_name):
+def request(context, number, instance_name, wait_for_completion):
     context.logger.info("Sending execution request...\n")
     stub = remote_execution_pb2_grpc.ExecutionStub(context.channel)
 
@@ -64,14 +68,23 @@ def request(context, number, instance_name):
     request = remote_execution_pb2.ExecuteRequest(instance_name = instance_name,
                                                   action = action,
                                                   skip_cache_lookup = True)
-    try:
-        for i in range(0, number):
-            response = stub.Execute(request)
-            context.logger.info("Response name: {}".format(response.name))
+    for i in range(0, number):
+        response = stub.Execute(request)
+        context.logger.info("Response name: {}".format(response.name))
 
-    except Exception as e:
-        context.logger.error(e)
-        return
+    try:
+        while wait_for_completion:
+            request = operations_pb2.ListOperationsRequest()
+            context.logger.debug('Querying to see if jobs are complete.')
+            stub = operations_pb2_grpc.OperationsStub(context.channel)
+            response = stub.ListOperations(request)
+            if all(operation.done for operation in response.operations):
+                context.logger.info('Jobs complete')
+                break
+            time.sleep(1)
+
+    except KeyboardInterrupt:
+        pass
 
 @cli.command('status', short_help='Get the status of an operation')
 @click.argument('operation-name')
@@ -82,13 +95,8 @@ def operation_status(context, operation_name):
 
     request = operations_pb2.GetOperationRequest(name=operation_name)
 
-    try:
-        response = stub.GetOperation(request)
-        _log_operation(context, response)
-
-    except Exception as e:
-        context.logger.error(e)
-        return
+    response = stub.GetOperation(request)
+    _log_operation(context, response)
 
 @cli.command('list', short_help='List operations')
 @pass_context
@@ -98,12 +106,7 @@ def list_operations(context):
 
     request = operations_pb2.ListOperationsRequest()
 
-    try:
-        response = stub.ListOperations(request)
-
-    except Exception as e:
-        context.logger.error(e)
-        return
+    response = stub.ListOperations(request)
 
     if len(response.operations) < 1:
         context.logger.warning("No operations to list")

setup.cfg

+[aliases]
+test=pytest
+
+[tool:pytest]
+#addopts = --pep8 --pylint
+addopts = --verbose --cov=buildgrid --cov-config=.coveragerc
+python_files = tests/*.py
+pep8ignore =
+    * E129
+    * E125
+    */lib/python3* ALL
+    */bin/* ALL
+    .eggs/* ALL
+    *_pb2.py ALL
+    *_pb2_grpc.py ALL
+pep8maxlinelength = 119
\ No newline at end of file

setup.py

@@ -43,18 +43,22 @@ setup(
         'protobuf',
         'grpcio',
         'Click',
-        ],
-    extras_require={
-        'cas-s3': ['boto3', 'botocore'],
-        ],
-    tests_require=[
-        'pytest',
         'boto3',
         'botocore',
-        'moto',
-        ],
+    ],
     entry_points='''
     [console_scripts]
     bgd=app:cli
     ''',
+    setup_requires=['pytest-runner'],
+    tests_require=['pep8',
+                   'boto3',
+                   'botocore',
+                   'moto',
+                   'coverage == 4.4.0',
+                   'pytest-cov >= 2.5.0',
+                   'pytest-pep8',
+                   'pytest-pylint',
+                   'pytest >= 3.1.0',
+                   'pylint >= 1.8 , < 2'],
 )

test/cas/test_services.py → tests/cas/test_services.py

@@ -70,7 +70,7 @@ def raise_mock_exception(*args, **kwargs):
     raise MockException()
 
 
-test_strings = [b"", b"hij", b"testing!" * 1000000]
+test_strings = [b"", b"hij"]
 instances = ["", "test_inst"]
 
 
@@ -91,6 +91,24 @@ def test_bytestream_read(data_to_read, instance):
     assert data == data_to_read
 
 
+@pytest.mark.parametrize("instance", instances)
+def test_bytestream_read_many(instance):
+    data_to_read = b"testing" * 10000
+
+    storage = SimpleStorage([b"abc", b"defg", data_to_read])
+    servicer = ByteStreamService(storage)
+
+    request = bytestream_pb2.ReadRequest()
+    if instance != "":
+        request.resource_name = instance + "/"
+    request.resource_name += f"blobs/{HASH(data_to_read).hexdigest()}/{len(data_to_read)}"
+
+    data = b""
+    for response in servicer.Read(request, None):
+        data += response.data
+    assert data == data_to_read
+
+
 @pytest.mark.parametrize("instance", instances)
 @pytest.mark.parametrize("extra_data", ["", "/", "/extra/data"])
 def test_bytestream_write(instance, extra_data):