PK!N&&coveragespace/__init__.pyfrom pkg_resources import DistributionNotFound, get_distribution try: __version__ = get_distribution('coveragespace').version except DistributionNotFound: __version__ = '(local)' CLI = 'coveragespace' API = 'https://api.coverage.space' VERSION = "{0} v{1}".format(CLI, __version__) PK!Qϒcoveragespace/cache.pyimport logging import os import pickle log = logging.getLogger(__name__) class Cache(object): PATH = os.path.join('.cache', 'coveragespace') def __init__(self): self._data = {} self._load() def _load(self): try: with open(self.PATH, 'rb') as fin: text = fin.read() except IOError: text = None try: data = pickle.loads(text) except (TypeError, KeyError, IndexError): data = None if isinstance(data, dict): self._data = data def _store(self): directory = os.path.dirname(self.PATH) if not os.path.exists(directory): os.makedirs(directory) text = pickle.dumps(self._data) with open(self.PATH, 'wb') as fout: fout.write(text) def set(self, key, value): try: url, data = key except ValueError: log.debug("Setting cache for %s", key) else: log.debug("Setting cache for %s: %s", url, data) key = self._slugify(*key) self._data[key] = value log.debug("Cached value: %s", value) self._store() def get(self, key, default=None): try: url, data = key except ValueError: log.debug("Getting cache for %s", key) else: log.debug("Getting cache for %s: %s", url, data) key = self._slugify(*key) value = self._data.get(key, default) log.debug("Cached value: %s", value) return value @staticmethod def _slugify(url, data): return (url, hash(frozenset(data.items()))) PK!.KN N coveragespace/cli.py"""Update project metrics on The Coverage Space. Usage: coveragespace [] [--verbose] [--exit-code] coveragespace --reset [--verbose] coveragespace (-h | --help) coveragespace (-V | --version) Options: -h --help Show this help screen. -V --version Show the program version. -v --verbose Always display the coverage metrics. -x --exit-code Return non-zero exit code on failures. """ from __future__ import unicode_literals import json import logging import sys import colorama import six from backports.shutil_get_terminal_size import \ get_terminal_size # pylint: disable=relative-import from docopt import DocoptExit, docopt from . import API, VERSION, client, services from .plugins import get_coverage, launch_report log = logging.getLogger(__name__) def main(): """Parse command-line arguments, configure logging, and run the program.""" colorama.init(autoreset=True) arguments = docopt(__doc__, version=VERSION) slug = arguments[''] metric = arguments[''] reset = arguments['--reset'] value = arguments[''] verbose = arguments['--verbose'] hardfail = arguments['--exit-code'] logging.basicConfig( level=logging.DEBUG if verbose else logging.WARNING, format="%(levelname)s: %(name)s: %(message)s", ) if '/' not in slug: raise DocoptExit(" slug must contain a slash" + '\n') success = run(slug, metric, value, reset, verbose, hardfail) if not success and hardfail: sys.exit(1) def run(*args, **kwargs): """Run the program.""" if services.detected(): log.info("Coverage check skipped when running on CI service") return True return call(*args, **kwargs) def call(slug, metric, value, reset=False, verbose=False, hardfail=False): """Call the API and display errors.""" url = "{}/{}".format(API, slug) if reset: data = {metric: None} response = client.delete(url, data) else: data = {metric: value or get_coverage()} response = client.get(url, data) if response.status_code == 200: if verbose: display("coverage increased", response.json(), colorama.Fore.GREEN) return True elif response.status_code == 202: display("coverage reset", response.json(), colorama.Fore.BLUE) return True elif response.status_code == 422: color = colorama.Fore.RED if hardfail else colorama.Fore.YELLOW data = response.json() data['help'] = \ "To reset metrics, run: coveragespace {} --reset".format(slug) display("coverage decreased", data, color) launch_report() return False else: try: data = response.json() display("coverage unknown", data, colorama.Fore.RED) except (TypeError, ValueError) as exc: data = response.data.decode('utf-8') log.error("%s\n\nwhen decoding response:\n\n%s\n", exc, data) return False def display(title, data, color=""): """Write colored text to the console.""" color += colorama.Style.BRIGHT width, _ = get_terminal_size() six.print_(color + "{0:=^{1}}".format(' ' + title + ' ', width)) six.print_(color + json.dumps(data, indent=4)) six.print_(color + '=' * width) PK!:_coveragespace/client.py"""API client functions.""" import logging import time import requests from .cache import Cache log = logging.getLogger(__name__) cache = Cache() def get(url, data): log.info("Getting %s: %s", url, data) response = cache.get((url, data)) if response is None: for i in range(3): response = requests.put(url, data=data) if response.status_code == 500: time.sleep(i + 1) continue else: break cache.set((url, data), response) log.info("Response: %s", response) return response def delete(url, data): log.info("Deleting %s: %s", url, data) for i in range(3): response = requests.delete(url, data=data) if response.status_code == 500: time.sleep(i + 1) continue else: break log.info("Response: %s", response) return response PK!B  coveragespace/plugins.py"""Plugins to extract coverage data from various formats.""" import logging import os import time import webbrowser from abc import ABCMeta, abstractmethod import coverage from six import with_metaclass from .cache import Cache log = logging.getLogger(__name__) cache = Cache() class BasePlugin(with_metaclass(ABCMeta)): # pylint: disable=no-init """Base class for coverage plugins.""" @abstractmethod def matches(self, cwd): """Determine if the current directory contains coverage data. :return bool: Indicates that the current directory should be processed. """ raise NotImplementedError @abstractmethod def get_coverage(self, cwd): """Extract the coverage data from the current directory. :return float: Percentage of lines covered. """ raise NotImplementedError @abstractmethod def get_report(self, cwd): """Get the path to the coverage report. :return str: Path to coverage report or `None` if not available. """ raise NotImplementedError def get_coverage(cwd=None): """Extract the current coverage data.""" cwd = cwd or os.getcwd() plugin = _find_plugin(cwd) percentage = plugin.get_coverage(cwd) return round(percentage, 1) def launch_report(cwd=None): """Open the generated coverage report in a web browser.""" cwd = cwd or os.getcwd() plugin = _find_plugin(cwd, allow_missing=True) if plugin: path = plugin.get_report(cwd) if path and not _launched_recently(path): log.info("Launching report: %s", path) webbrowser.open("file://" + path, new=2, autoraise=True) def _find_plugin(cwd, allow_missing=False): """Find an return a matching coverage plugin.""" for cls in BasePlugin.__subclasses__(): # pylint: disable=no-member plugin = cls() if plugin.matches(cwd): return plugin msg = "No coverage data found: {}".format(cwd) log.info(msg) if allow_missing: return None raise RuntimeError(msg) def _launched_recently(path): now = time.time() then = cache.get(path, default=0) elapsed = now - then log.debug("Last launched %s seconds ago", elapsed) cache.set(path, now) return elapsed < 60 * 60 # 1 hour class CoveragePy(BasePlugin): # pylint: disable=no-init """Coverage extractor for the coverage.py format.""" def matches(self, cwd): return any(( '.coverage' in os.listdir(cwd), '.coveragerc' in os.listdir(cwd), )) def get_coverage(self, cwd): os.chdir(cwd) cov = coverage.Coverage() cov.load() with open(os.devnull, 'w') as ignore: total = cov.report(file=ignore) return total def get_report(self, cwd): path = os.path.join(cwd, 'htmlcov', 'index.html') if os.path.isfile(path): log.info("Found coverage report: %s", path) return path log.info("No coverage report found: %s", cwd) return None PK!Kcoveragespace/services.py"""Utilities to detect when this program is running on external services.""" import os CONTINUOUS_INTEGRATION = [ # General 'CI', 'CONTINUOUS_INTEGRATION', 'DISABLE_COVERAGE', # Travis CI 'TRAVIS', # Appveyor 'APPVEYOR', # CircleCI 'CIRCLECI', # Drone 'DRONE', ] def detected(): return any(name in CONTINUOUS_INTEGRATION for name in os.environ) PK!<ͤ22coveragespace/tests/__init__.py"""Unit tests for the `coveragespace` package.""" PK!~Gcoveragespace/tests/conftest.py"""Unit tests configuration file.""" import logging def pytest_configure(config): """Disable verbose output when running tests.""" logging.basicConfig( level=logging.DEBUG, format="%(levelname)s: %(name)s: %(message)s", ) terminal = config.pluginmanager.getplugin('terminal') base = terminal.TerminalReporter class QuietReporter(base): """A py.test reporting that only shows dots when running tests.""" def __init__(self, *args, **kwargs): base.__init__(self, *args, **kwargs) self.verbosity = 0 self.showlongtestinfo = False self.showfspath = False terminal.TerminalReporter = QuietReporter PK!9tt!coveragespace/tests/test_cache.py# pylint: disable=missing-docstring,unused-variable,unused-argument,expression-not-assigned,singleton-comparison import os import pytest from expecter import expect from coveragespace.cache import Cache def describe_cache(): @pytest.fixture def cache(): return Cache() @pytest.fixture def cache_empty(cache): # pylint: disable=protected-access cache._data.clear() return cache @pytest.fixture def cache_missing(cache_empty): try: os.remove(Cache.PATH) except OSError: pass return cache_empty @pytest.fixture def cache_corrupt(cache): # pylint: disable=protected-access cache._data = "corrupt" cache._store() return cache def describe_init(): def it_loads_previous_results(cache_empty): cache_empty.set(("url", {}), "previous") cache = Cache() expect(cache.get(("url", {}))) == "previous" def it_handles_missing_cache_files(cache_missing): expect(Cache().get(("url", {}))) == None def it_handles_corrupt_cache_files(cache_corrupt): expect(Cache().get(("url", {}))) == None def describe_get(): def it_hits_with_existing_data(cache_empty): cache = cache_empty cache.set(("url", {}), "existing") expect(cache.get(("url", {}))) == "existing" def it_misses_with_no_data(cache_empty): expect(cache_empty.get(("url", {}))) == None def it_returns_the_default_on_miss(cache_empty): expect(cache_empty.get("foo", 42)) == 42 PK!icoveragespace/tests/test_cli.py# pylint: disable=unused-variable,expression-not-assigned,singleton-comparison from expecter import expect from mock import Mock, patch from coveragespace import cli def describe_call(): @patch('coveragespace.cache.Cache.get', Mock()) def it_handles_invalid_response(): expect(cli.call('slug', 'metric', 42)) == False @patch('coveragespace.cache.Cache.get', Mock(return_value=None)) @patch('coveragespace.cache.Cache.set', Mock(return_value=None)) @patch('time.sleep', Mock()) @patch('requests.put') def it_retries_500s(requests_put): requests_put.return_value = Mock(status_code=500) cli.call('slug', 'metric', 42) expect(requests_put.call_count) == 3 PK!?#coveragespace/tests/test_plugins.py# pylint: disable=missing-docstring,unused-variable,unused-argument,expression-not-assigned,singleton-comparison import os import time import pytest from expecter import expect from mock import Mock, patch from coveragespace.plugins import _launched_recently, cache, get_coverage class MockCoverage(Mock): @staticmethod def report(*args, **kwargs): return 42.456 def describe_get_coverage(): @pytest.fixture def coveragepy_data(tmpdir): cwd = tmpdir.chdir() with open("foobar.py", 'w') as stream: pass with open(".coverage", 'w') as stream: stream.write(""" !coverage.py: This is a private format, don\'t read it directly! {"arcs":{"foobar.py": [[-1, 2]]}} """.strip()) @pytest.fixture def coveragepy_data_custom(tmpdir): cwd = tmpdir.chdir() with open("foobar.py", 'w') as stream: pass with open(".coveragerc", 'w') as stream: stream.write(""" [run] data_file = .cache/coverage """.strip()) os.makedirs('.cache') with open(".cache/coverage", 'w') as stream: stream.write(""" !coverage.py: This is a private format, don\'t read it directly! {"arcs":{"foobar.py": [[-1, 3]]}} """.strip()) @patch('coverage.Coverage', MockCoverage) def it_supports_coveragepy(coveragepy_data): expect(get_coverage()) == 42.5 @patch('coverage.Coverage', MockCoverage) def it_supports_coveragepy_with_custom_location(coveragepy_data_custom): expect(get_coverage()) == 42.5 def describe_launched_recently(): def when_never_launched(): cache.set('mock/path', 0) expect(_launched_recently('mock/path')) == False def when_just_launched(): cache.set('mock/path', time.time()) expect(_launched_recently('mock/path')) == True def when_launched_59_minutes_ago(): cache.set('mock/path', time.time() - 60 * 59) expect(_launched_recently('mock/path')) == True def when_launched_61_minutes_ago(): cache.set('mock/path', time.time() - 60 * 61) expect(_launched_recently('mock/path')) == False PK!3$$coveragespace/tests/test_services.py# pylint: disable=missing-docstring,unused-variable,unused-argument,expression-not-assigned,singleton-comparison from expecter import expect from coveragespace import services def describe_detected(): def when_off_ci(monkeypatch): monkeypatch.delenv('APPVEYOR', raising=False) monkeypatch.delenv('CI', raising=False) monkeypatch.delenv('CONTINUOUS_INTEGRATION', raising=False) monkeypatch.delenv('TRAVIS', raising=False) monkeypatch.delenv('DISABLE_COVERAGE', raising=False) expect(services.detected()) == False def when_on_ci(monkeypatch): monkeypatch.setenv('TRAVIS', 'true') expect(services.detected()) == True PK!HV/8,coveragespace-2.1.dist-info/entry_points.txtN+I/N.,()J/K-JLO-.HLNE%dZ&fqqPK!t@@&coveragespace-2.1.dist-info/LICENSE.md**The MIT License (MIT)** Copyright © 2016, Jace Browning Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!H9VWX!coveragespace-2.1.dist-info/WHEEL A н#f."jm)!fb҅~ܴA,mTD}E n0H饹*|D[¬c i=0(q3PK!H@ Ǚ $coveragespace-2.1.dist-info/METADATAWS6~_7b3@uJ %>0A7R%9!뻒iͶZ}BrfYj#dA7ً3r(1juY2=TA`%Xd2Dž9h&~%Ɗ>e0V,MIy!it)8VpWwo8J%EcE5pI<<.DimRg#IiEk%E_im,4rߵhi'd'<NtV0cP < RD@!pe^sKyKiaP`s5ZVގ de Ee1:ƲE<:IC\/T)M;q2Bd5+UڝUX a-5c.ПeArT_e[pHtCv7hCM[nlqcv-NH*0^salR%5VmlQYd&/͹,$e9n&"Ёt\rl9va;?t(_1sYjb!.nQI#ԳHq=pZJB_c'&9r1cg?dEx݃{O(FzRXQ(ILLY a'f2qE"hmSTDx+ͷ$RIrL܉ϸ soOB3Y {w>t*P5>E,ڳ{AoKZ*f>ƏDfvse5S"!3IK_ޤet78M=CE\Ӈiݍp(:A`(5<8ԲʜE: 8}-h̩yӠqJ_TRy! U0@a) ((z||16@׊QDs"m1%X dhGDAR:6WȄzݰwrE1(nBʁZd5#afח ־G,NiI3ygV;CC8Vn uE!YшO|k #߈p,F'jDtE\cfC熦 OIJWiH"VJ *(nx{O&zgPK!Hw|"coveragespace-2.1.dist-info/RECORDɒJ}? TC&^(L 2!G!Eouܨۮd;'#?jǤd(ES fl@3?gESUr$LIcYffFA'O-nvShL$s^?nQp| F2x ,%A.7x7W݅ZT/x{ R'4;A7Qے^2G^A5Fh u=+;NBIO[a.MXZKؐ0H^~,Y]䲽S=>Tfpw9 ,{9A {=˿Qo]fZuIJ)]>d=$͠8nF0r&߮N!N.}M{&ǚG(L-{I]IQeVcx:c#`R;k{Ml Y>l W-׿v)oNqW쇅XvCtlfZp@2!7E^7 PK!N&&coveragespace/__init__.pyPK!Qϒ]coveragespace/cache.pyPK!.KN N #coveragespace/cli.pyPK!:_coveragespace/client.pyPK!B  zcoveragespace/plugins.pyPK!K%coveragespace/services.pyPK!<ͤ22'coveragespace/tests/__init__.pyPK!~G(coveragespace/tests/conftest.pyPK!9tt!+coveragespace/tests/test_cache.pyPK!i1coveragespace/tests/test_cli.pyPK!?#4coveragespace/tests/test_plugins.pyPK!3$$=coveragespace/tests/test_services.pyPK!HV/8,@coveragespace-2.1.dist-info/entry_points.txtPK!t@@&FAcoveragespace-2.1.dist-info/LICENSE.mdPK!H9VWX!Ecoveragespace-2.1.dist-info/WHEELPK!H@ Ǚ $`Fcoveragespace-2.1.dist-info/METADATAPK!Hw|"OKcoveragespace-2.1.dist-info/RECORDPKN