grim/local-pipelines

Pulled Step's out to their own class/file and added unit testing for them

2016-07-26, Gary Kramlich
c1318ad7ebfc
Parents 09b27a025683
Children 0d24f0adff6c
Pulled Step's out to their own class/file and added unit testing for them
6 files changed, 142 insertions(+), 18 deletions(-)
  • +2 -0
    .hgignore
  • +3 -0
    bitbucket-pipelines.yml
  • +2 -0
    dev-requirements.txt
  • +5 -18
    pipelines/pipeline.py
  • +58 -0
    pipelines/step.py
  • +72 -0
    tests/test_step.py
    • --- a/.hgignore Tue Jul 26 19:58:49 2016 -0500
    +++ b/.hgignore Tue Jul 26 21:26:55 2016 -0500
    @@ -1,4 +1,5 @@
    syntax: regexp
    +^\.cache/
    ^build/
    ^dist/
    ^venv/
    @@ -7,4 +8,5 @@
    *.pyc
    .*.swp
    *.egg-info
    +.coverage
    --- a/bitbucket-pipelines.yml Tue Jul 26 19:58:49 2016 -0500
    +++ b/bitbucket-pipelines.yml Tue Jul 26 21:26:55 2016 -0500
    @@ -4,5 +4,8 @@
    - step:
    script:
    - set -ex
    + - find . -type f -iname "*.pyc" -exec rm -f {} \; || true
    + - find . -type d -iname __pycache__ -exec rm -rf {} \; || true
    - pip install -r dev-requirements.txt
    - flake8
    + - PYTHONPATH=$(pwd) py.test --color=auto --cov=pipelines --cov-report=term-missing tests
    --- a/dev-requirements.txt Tue Jul 26 19:58:49 2016 -0500
    +++ b/dev-requirements.txt Tue Jul 26 21:26:55 2016 -0500
    @@ -1,3 +1,5 @@
    -f requirements.txt
    flake8
    +pytest
    +pytest-cover
    --- a/pipelines/pipeline.py Tue Jul 26 19:58:49 2016 -0500
    +++ b/pipelines/pipeline.py Tue Jul 26 21:26:55 2016 -0500
    @@ -21,7 +21,8 @@
    import fnmatch
    import os
    import subprocess
    -import tempfile
    +
    +from pipelines.step import Step
    class Pipeline(object):
    @@ -44,21 +45,7 @@
    return self.config["pipelines"]["default"]
    def _determine_image(self, step):
    - return step.get("image", self.config.get("image"))
    -
    - def _script(self, step):
    - fd, filename = tempfile.mkstemp(prefix='pipeline', suffix='.sh')
    - filename = os.path.realpath(filename)
    -
    - with os.fdopen(fd, "w") as ofp:
    - ofp.write("#!/bin/bash\n")
    -
    - content = " && \\\n".join(step["script"])
    - ofp.write(content)
    -
    - ofp.write("\n")
    -
    - return filename
    + return step.image or self.config.get("image")
    def _get_command(self, step, script):
    workdir = "/opt/atlassian/bitbucketci/agent/build"
    @@ -93,13 +80,13 @@
    return " ".join(cmd)
    def _steps(self):
    - return [s["step"] for s in self.steps]
    + return [Step.from_dict(s["step"]) for s in self.steps]
    def run(self):
    return_code = 0
    for n, step in enumerate(self._steps()):
    - script = self._script(step)
    + script = step.script_file()
    command = self._get_command(step, script)
    try:
    --- /dev/null Thu Jan 01 00:00:00 1970 +0000
    +++ b/pipelines/step.py Tue Jul 26 21:26:55 2016 -0500
    @@ -0,0 +1,58 @@
    +# vi:et:ts=4 sw=4 sts=4
    +#
    +# local-pipelines : run Bitbucket pipelines locally
    +# Copyright (C) 2016 Gary Kramlich <grim@reaperworld.com>
    +# Copyright (C) 2016 Sean Farley <sean@farley.io>
    +#
    +# This program is free software; you can redistribute it and/or modify
    +# it under the terms of the GNU General Public License as published by
    +# the Free Software Foundation; either version 3 of the License, or
    +# (at your option) any later version.
    +#
    +# This program is distributed in the hope that it will be useful,
    +# but WITHOUT ANY WARRANTY; without even the implied warranty of
    +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
    +# GNU General Public License for more details.
    +#
    +# You should have received a copy of the GNU General Public License
    +# along with this program; if not, write to the Free Software Foundation,
    +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
    +
    +import os
    +import tempfile
    +
    +
    +class Step(object):
    + def __init__(self, image, script):
    + self._image = image
    +
    + self._script = ["#!/bin/bash", "set -ex"]
    + if script is not None:
    + self._script.extend(script)
    +
    + @classmethod
    + def from_dict(cls, dict):
    + image = dict.get("image", None)
    + script = dict.get("script", [])
    +
    + return cls(image, script)
    +
    + @property
    + def image(self):
    + return self._image
    +
    + @property
    + def script(self):
    + return self._script
    +
    + def __str__(self):
    + return "\n".join(self._script)
    +
    + def script_file(self):
    + fd, filename = tempfile.mkstemp(prefix="pipeline-", suffix=".sh")
    + filename = os.path.realpath(filename)
    +
    + with os.fdopen(fd, "w") as ofp:
    + ofp.write(str(self))
    +
    + return filename
    --- /dev/null Thu Jan 01 00:00:00 1970 +0000
    +++ b/tests/test_step.py Tue Jul 26 21:26:55 2016 -0500
    @@ -0,0 +1,72 @@
    +# vi:et:ts=4 sw=4 sts=4
    +#
    +# local-pipelines : run Bitbucket pipelines locally
    +# Copyright (C) 2016 Gary Kramlich <grim@reaperworld.com>
    +# Copyright (C) 2016 Sean Farley <sean@farley.io>
    +#
    +# This program is free software; you can redistribute it and/or modify
    +# it under the terms of the GNU General Public License as published by
    +# the Free Software Foundation; either version 3 of the License, or
    +# (at your option) any later version.
    +#
    +# This program is distributed in the hope that it will be useful,
    +# but WITHOUT ANY WARRANTY; without even the implied warranty of
    +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
    +# GNU General Public License for more details.
    +#
    +# You should have received a copy of the GNU General Public License
    +# along with this program; if not, write to the Free Software Foundation,
    +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
    +
    +from __future__ import print_function
    +
    +from pipelines.step import Step
    +
    +
    +class TestStep(object):
    + def _test(self, step, image, script):
    + assert step.image == image
    +
    + real_script = ["#!/bin/bash", "set -ex"]
    + if script is not None:
    + real_script.extend(script)
    +
    + assert step.script == real_script
    + assert str(step) == "\n".join(real_script)
    +
    + def test_manual(self):
    + tests = {
    + None: None,
    + "none": None,
    + "empty": [],
    + "single": ["one"],
    + "double": ["one", "two"],
    + }
    +
    + for image, script in tests.items():
    + self._test(Step(image, script), image, script)
    +
    + def test_from_dict(self):
    + tests = [
    + {
    + "image": None,
    + "script": None,
    + }, {
    + "image": "none",
    + "script": None,
    + }, {
    + "image": "empty",
    + "script": [],
    + }, {
    + "image": "single",
    + "script": ["one"],
    + }, {
    + "image": "double",
    + "script": ["one", "two"],
    + },
    + ]
    +
    + for test in tests:
    + step = Step.from_dict(test)
    +
    + self._test(step, test["image"], test["script"])