Skip to content

Commit

Permalink
Reload after coverage initialization (#270)
Browse files Browse the repository at this point in the history 
This commit reloads modules after initializing coverage racing,
 to include package imports into "tested" code results.
  • Loading branch information
@deathaxe
deathaxe authored Jul 6, 2024
1 parent a2a776a commit 412f377
Show file tree
Hide file tree
Showing 2 changed files with 105 additions and 79 deletions.
39 changes: 29 additions & 10 deletions unittesting/reloader.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import sublime
import sublime_plugin
import sys
from time import sleep


def path_contains(a, b):
Expand Down Expand Up @@ -55,6 +56,33 @@ def reload_package(pkg_name, on_done=None):
sublime.set_timeout(on_done)
return

dummy_name, dummy_py = _reload_package(pkg_name)

def check_loaded():
if dummy_name not in sys.modules:
sublime.set_timeout(check_loaded, 100)
return

os.remove(dummy_py)
if on_done:
sublime.set_timeout(on_done, 200)

sublime.set_timeout(check_loaded, 100)


def async_reload_package(pkg_name):
if pkg_name not in sys.modules:
return

dummy_name, dummy_py = _reload_package(pkg_name)

while dummy_name not in sys.modules:
sleep(0.1)

os.remove(dummy_py)


def _reload_package(pkg_name):
all_modules = {
module_name: module
for module_name, module in get_package_modules(pkg_name).items()
Expand Down Expand Up @@ -92,13 +120,4 @@ def reload_package(pkg_name, on_done=None):

open(dummy_py, "a").close()

def check_loaded():
if dummy_name not in sys.modules:
sublime.set_timeout(check_loaded, 100)
return

os.remove(dummy_py)
if on_done:
sublime.set_timeout(on_done, 200)

sublime.set_timeout(check_loaded, 100)
return dummy_name, dummy_py
145 changes: 76 additions & 69 deletions unittesting/unit.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
from .core import DeferrableTestCase
from .core import DeferrableTestLoader
from .core import DeferringTextTestRunner
from .reloader import reload_package
from .reloader import async_reload_package, reload_package

try:
import coverage
Expand Down Expand Up @@ -97,91 +97,98 @@ def run(self, package=None, **kwargs):

stream = self.load_stream(package, settings)

def run_tests():
if settings["async"]:
threading.Thread(
target=self.run_coverage, args=(package, stream, settings)
).start()
else:
self.run_coverage(package, stream, settings)

if settings["reload_package_on_testing"]:
reload_package(package, on_done=run_tests)
if settings["async"]:
threading.Thread(
target=self.run_coverage, args=(package, stream, settings)
).start()
else:
run_tests()
self.run_coverage(package, stream, settings)

def run_coverage(self, package, stream, settings):
if not coverage or not settings["coverage"]:
if settings["coverage"]:
stream.write("Warning: coverage cannot be loaded.\n\n")
# prepare coverage
cleanup_hooks = []
if settings["coverage"] and not coverage:
stream.write("Warning: coverage cannot be loaded.\n\n")

elif settings["coverage"]:
packages_path = sublime.packages_path()
package_path = os.path.join(packages_path, package)
data_file_dir = os.path.join(packages_path, "User", "UnitTesting", package)
os.makedirs(data_file_dir, exist_ok=True)
data_file = os.path.join(data_file_dir, "coverage")
if os.path.exists(data_file):
os.unlink(data_file)
config_file = os.path.join(package_path, ".coveragerc")
include = "{}/*".format(package_path)
omit = "{}/{}/*".format(package_path, settings["tests_dir"])
if os.path.exists(config_file):
with open(config_file, "r") as f:
txt = f.read()
if re.search("^include", txt, re.M):
include = None
if re.search("^omit", txt, re.M):
omit = None
else:
config_file = False

self.run_tests(stream, package, settings, [])
return
cov = coverage.Coverage(
data_file=data_file, config_file=config_file, include=include, omit=omit
)

packages_path = sublime.packages_path()
package_path = os.path.join(packages_path, package)
data_file_dir = os.path.join(packages_path, "User", "UnitTesting", package)
os.makedirs(data_file_dir, exist_ok=True)
data_file = os.path.join(data_file_dir, "coverage")
if os.path.exists(data_file):
os.unlink(data_file)
config_file = os.path.join(package_path, ".coveragerc")
include = "{}/*".format(package_path)
omit = "{}/{}/*".format(package_path, settings["tests_dir"])
if os.path.exists(config_file):
with open(config_file, "r") as f:
txt = f.read()
if re.search("^include", txt, re.M):
include = None
if re.search("^omit", txt, re.M):
omit = None
else:
config_file = False
cov.start()

cov = coverage.Coverage(
data_file=data_file, config_file=config_file, include=include, omit=omit
)
if settings["coverage_on_worker_thread"]:
original_set_timeout_async = sublime.set_timeout_async

cov.start()
def set_timeout_async(callback, *args, **kwargs):
def _():
sys.settrace(threading._trace_hook)
callback()

if settings["coverage_on_worker_thread"]:
original_set_timeout_async = sublime.set_timeout_async
original_set_timeout_async(_, *args, **kwargs)

def set_timeout_async(callback, *args, **kwargs):
def _():
sys.settrace(threading._trace_hook)
callback()
sublime.set_timeout_async = set_timeout_async

original_set_timeout_async(_, *args, **kwargs)
def cleanup():
if settings["coverage_on_worker_thread"]:
sublime.set_timeout_async = original_set_timeout_async

stream.write("\n")
cov.stop()
coverage.files.RELATIVE_DIR = os.path.normcase(package_path + os.sep)
ignore_errors = cov.get_option("report:ignore_errors")
show_missing = cov.get_option("report:show_missing")
cov.report(
file=stream, ignore_errors=ignore_errors, show_missing=show_missing
)

if settings["generate_xml_report"]:
xml_report_file = os.path.join(package_path, "coverage.xml")
cov.xml_report(outfile=xml_report_file, ignore_errors=ignore_errors)

if settings["generate_html_report"]:
html_output_dir = os.path.join(package_path, "htmlcov")
cov.html_report(
directory=html_output_dir, ignore_errors=ignore_errors
)

sublime.set_timeout_async = set_timeout_async
cov.save()

def cleanup():
if settings["coverage_on_worker_thread"]:
sublime.set_timeout_async = original_set_timeout_async

stream.write("\n")
cov.stop()
coverage.files.RELATIVE_DIR = os.path.normcase(package_path + os.sep)
ignore_errors = cov.get_option("report:ignore_errors")
show_missing = cov.get_option("report:show_missing")
cov.report(
file=stream, ignore_errors=ignore_errors, show_missing=show_missing
)
cleanup_hooks = [cleanup]

if settings["generate_xml_report"]:
xml_report_file = os.path.join(package_path, "coverage.xml")
cov.xml_report(outfile=xml_report_file, ignore_errors=ignore_errors)
def run_tests():
self.run_tests(stream, package, settings, cleanup_hooks)

if settings["generate_html_report"]:
html_output_dir = os.path.join(package_path, "htmlcov")
cov.html_report(directory=html_output_dir, ignore_errors=ignore_errors)
if settings["reload_package_on_testing"]:
if not settings["async"]:
reload_package(package, on_done=run_tests)
return

cov.save()
async_reload_package(package)

self.run_tests(stream, package, settings, [cleanup])
run_tests()

def run_tests(self, stream, package, settings, cleanup_hooks=[]):
def run_tests(self, stream, package, settings, cleanup_hooks):
if settings["capture_console"]:
stdout = sys.stdout
stderr = sys.stderr
Expand Down

0 comments on commit 412f377

Please sign in to comment.