# mode: run
# tag: coverage,trace

"""
PYTHON -c "import shutil; shutil.copy('pkg/coverage_test_pyx.pyx', 'pkg/coverage_test_pyx.pxi')"
PYTHON setup.py build_ext -i
PYTHON -m coverage run coverage_test.py
PYTHON collect_coverage.py
"""

######## setup.py ########

from distutils.core import setup
from Cython.Build import cythonize

setup(ext_modules = cythonize([
    'coverage_test_*.py*',
    'pkg/coverage_test_*.py*'
]))


######## .coveragerc ########
[run]
plugins = Cython.Coverage


######## pkg/__init__.py ########

######## pkg/coverage_test_py.py ########
# cython: linetrace=True
# distutils: define_macros=CYTHON_TRACE=1
import cython


def func1(a, b):
    x = 1               #  7
    c = func2(a) + b    #  8
    return x + c        #  9


def func2(a):
    return a * 2        # 13


def func3(a):
    x = 1               # 17
    a *= 2              #     # pragma: no cover
    a += x              #
    return a * 42       # 20  # pragma: no cover


@cython.cclass
class A:
    def meth(self):
        return 1        # 26


######## pkg/coverage_test_pyx.pyx ########
# cython: linetrace=True
# distutils: define_macros=CYTHON_TRACE=1



def func1(int a, int b):
    cdef int x = 1      #  7
    c = func2(a) + b    #  8
    return x + c        #  9


def func2(int a):
    return a * 2        # 13


def func3(int a):
    cdef int x = 1      # 17
    a *= 2              #     # pragma: no cover
    a += x              #
    return a * 42       # 20  # pragma: no cover



cdef class A:
    def meth(self):
        return 1        # 26


######## coverage_test_include_pyx.pyx ########
# cython: linetrace=True
# distutils: define_macros=CYTHON_TRACE=1

cdef int x = 5                                   #  4

cdef int cfunc1(int x):                          #  6
    return x * 3                                 #  7

include "pkg/coverage_test_pyx.pxi"              #  9

def main_func(int x):                            # 11
    return cfunc1(x) + func1(x, 4) + func2(x)    # 12


######## coverage_test.py ########

import os.path
try:
    # io.StringIO in Py2.x cannot handle str ...
    from StringIO import StringIO
except ImportError:
    from io import StringIO

from pkg import coverage_test_py
from pkg import coverage_test_pyx
import coverage_test_include_pyx


for module in [coverage_test_py, coverage_test_pyx, coverage_test_include_pyx]:
    assert not any(module.__file__.endswith(ext) for ext in '.py .pyc .pyo .pyw .pyx .pxi'.split()), \
        module.__file__


def run_coverage(module):
    module_name = module.__name__
    module_path = module_name.replace('.', os.path.sep) + '.' + module_name.rsplit('_', 1)[-1]

    assert module.func1(1, 2) == (1 * 2) + 2 + 1
    assert module.func2(2) == 2 * 2
    if '_include_' in module_name:
        assert module.main_func(2) == (2 * 3) + ((2 * 2) + 4 + 1) + (2 * 2)
    assert module.A().meth() == 1




if __name__ == '__main__':
    run_coverage(coverage_test_py)
    run_coverage(coverage_test_pyx)
    run_coverage(coverage_test_include_pyx)


######## collect_coverage.py ########

import re
import sys
import os
import os.path
import subprocess
from glob import iglob


def run_coverage_command(*command):
    env = dict(os.environ, LANG='', LC_ALL='C')
    process = subprocess.Popen(
        [sys.executable, '-m', 'coverage'] + list(command),
        stdout=subprocess.PIPE, env=env)
    stdout, _ = process.communicate()
    return stdout


def run_report():
    stdout = run_coverage_command('report', '--show-missing')
    stdout = stdout.decode('iso8859-1')  # 'safe' decoding
    lines = stdout.splitlines()
    print(stdout)

    # FIXME:  'coverage_test_pyx.pxi' may not be found if coverage.py requests it before the .pyx file
    for module_path in ('coverage_test_py.py', 'coverage_test_pyx.pyx', 'coverage_test_include_pyx.pyx'):
        assert any(module_path in line for line in lines), "'%s' not found in coverage report:\n\n%s" % (
            module_path, stdout)

    files = {}
    line_iter = iter(lines)
    for line in line_iter:
        if line.startswith('---'):
            break
    extend = [''] * 2
    for line in line_iter:
        if not line or line.startswith('---'):
            continue
        name, statements, missed, covered, _missing = (line.split(None, 4) + extend)[:5]
        missing = []
        for start, end in re.findall('([0-9]+)(?:-([0-9]+))?', _missing):
            if end:
                missing.extend(range(int(start), int(end)+1))
            else:
                missing.append(int(start))
        files[os.path.basename(name)] = (statements, missed, covered, missing)

    report = files['coverage_test_pyx.pyx']
    assert  7 not in report[-1], report
    assert 12 not in report[-1], report


def run_xml_report():
    stdout = run_coverage_command('xml', '-o', '-')
    print(stdout)

    import xml.etree.ElementTree as etree
    data = etree.fromstring(stdout)

    files = {}
    for module in data.iterfind('.//class'):
        files[module.get('filename').replace('\\', '/')] = dict(
            (int(line.get('number')), int(line.get('hits')))
            for line in module.findall('lines/line')
        )

    report = files['pkg/coverage_test_pyx.pyx']
    assert report[7] > 0, report
    assert report[8] > 0, report
    assert report[9] > 0, report
    assert report[26] > 0, report


def run_json_report():
    import coverage
    if coverage.version_info < (5, 0):
        # JSON output comes in coverage 5.0
        return

    stdout = run_coverage_command('json', '-o', '-')

    import json
    files = json.loads(stdout.decode("ascii"))['files']

    for filename in [
        'pkg/coverage_test_py.py',
        'pkg/coverage_test_pyx.pyx',
    ]:
        report = files[filename.replace('/', os.sep)]
        summary = report['summary']
        assert summary['missing_lines'] == 2, summary
        assert summary['excluded_lines'] == 2, summary
        assert report['missing_lines'] == [17, 19], report
        assert report['excluded_lines'] == [18, 20], report

        assert not frozenset(
            report['missing_lines'] + report['excluded_lines']
        ).intersection(report['executed_lines'])


def run_html_report():
    from collections import defaultdict

    stdout = run_coverage_command('html', '-d', 'html')
    # coverage 6.1+ changed the order of the attributes => need to parse them separately
    _parse_id = re.compile(r'id=["\'][^0-9"\']*(?P<id>[0-9]+)[^0-9"\']*["\']').search
    _parse_state = re.compile(r'class=["\'][^"\']*(?P<state>mis|run|exc)[^"\']*["\']').search

    files = {}
    for file_path in iglob('html/*.html'):
        with open(file_path) as f:
            page = f.read()
        report = defaultdict(set)
        for line in re.split(r'id=["\']source["\']', page)[-1].splitlines():
            lineno = _parse_id(line)
            state = _parse_state(line)
            if not lineno or not state:
                continue
            report[state.group('state')].add(int(lineno.group('id')))
        files[file_path] = report

    for filename, report in files.items():
        if "coverage_test_pyx" not in filename:
            continue
        executed = report["run"]
        missing = report["mis"]
        excluded = report["exc"]
        assert executed, (filename, report)
        assert 7 in executed, executed
        assert 8 in executed, executed
        assert 9 in executed, executed
        assert 26 in executed, executed
        assert 17 in missing, missing
        assert 18 in excluded, excluded
        assert 19 in missing, missing
        assert 20 in excluded, excluded


if __name__ == '__main__':
    run_report()
    run_xml_report()
    run_json_report()
    run_html_report()