import os
import shutil
import tempfile
from typing import Tuple, Any
from ghtest import scan, suggest, make_test, write_module
from ghtest.tests_writer import TestArtifact
from ghtest.tests_creator import _run_tests
[docs]
def create_tests(
cassette_dir: str,
test_dir: str,
src_dir: str,
clean_up: bool = True,
unsafe: bool = True,
history: bool = False,
vb: int = 0,
) -> Tuple[Any, Any, Any, Any]:
"""
Run test suite, return values from component functions.
Args:
cassette_dir (str): folder with vcr cassettes
test_dir (str): folder with test modules
src_dir (str): folder with src under test
clean_up (bool): remove existing cassettes and tests
unsafe (bool): run potentially destructive functions without requesting permission
history (bool): disregard parameter suggestions from history
vb (int): verbosity
Returns:
tuple: (scs, sps, gts, trs) (scans, suggested params, generated tests, test results)
Side Effects:
deletes cassette_dir, test_dir
writes cassette_dir, test_dir
executes functions in src, potentially including destructive functions
"""
if clean_up:
if os.path.exists(cassette_dir):
shutil.rmtree(cassette_dir)
if os.path.exists(test_dir):
shutil.rmtree(test_dir)
os.makedirs(cassette_dir, exist_ok=True)
os.makedirs(test_dir, exist_ok=True)
# Set unsafe mode
if unsafe:
os.environ["GHTEST_ASSUME_SAFE"] = "1"
if not history:
os.environ["GHTEST_PARAM_HISTORY"] = "" # Disable history for reproducibility
if vb > 1:
print(f"Scanning {src_dir}...") # noqa: E701
scs = scan(src_dir)
scs = [
f
for f in scs
if not f.qualname.startswith("_")
and not f.qualname.startswith("test_")
and not f.qualname.startswith("scenario_")
and f.qualname != "main"
]
if vb > 1:
print(f"Found {len(scs)} functions.") # noqa: E701
for f in scs:
if vb > 1:
print(
f" {f.qualname}: role={f.crud_role}, resource={f.crud_resource}"
) # noqa: E701
if vb > 1:
print("Suggesting params...") # noqa: E701
sps = []
total_params = 0
for func in scs:
sp = suggest(func, literal_only=False)
sps.append(sp)
total_params += len(sp.param_sets)
if vb > 1:
print(f" {func.qualname}: {len(sp.param_sets)} param sets") # noqa: E701
if func.qualname == "list_repos":
if vb > 1:
print(f" list_repos params: {sp.param_sets}") # noqa: E701
if func.qualname == "create_repo" and sp.scenario:
if vb > 1:
print(
f" create_repo scenario steps: {[s.params for s in sp.scenario.steps]}"
) # noqa: E701
if vb:
print(f"Total param sets: {total_params}") # noqa: E701
if vb > 1:
print("Generating tests and recording cassettes...") # noqa: E701
gts = []
for sp in sps:
if vb > 1:
print(f" Processing {sp.qualname}...") # noqa: E701
try:
gt = make_test(suggestion=sp, cassette_dir=cassette_dir)
gts.append(gt)
except Exception as e:
if vb:
print(f" ERROR creating test for {sp.qualname}: {e}") # noqa: E701
# Coverage Feedback Loop
try:
import coverage
if vb:
print("Running tests with coverage for feedback...") # noqa: E701
with tempfile.TemporaryDirectory() as tmp_cassette_dir:
# Regenerate tests for coverage run using temp cassette dir
gts_cov = []
for sp in sps:
try:
gt = make_test(suggestion=sp, cassette_dir=tmp_cassette_dir)
gts_cov.append(gt)
except Exception:
pass
cov = coverage.Coverage(source=[src_dir])
cov.start()
_run_tests(gts_cov, interactive=False, vb=vb)
cov.stop()
cov.save()
if vb:
print("Analyzing coverage for targeted suggestions...") # noqa: E701
sps_targeted = []
for func in scs:
# Pass coverage data to suggest
sp = suggest(func, literal_only=False, coverage_data=cov)
sps_targeted.append(sp)
# Regenerate tests with targeted suggestions (using real cassette_dir)
gts = []
for sp in sps_targeted:
try:
gt = make_test(suggestion=sp, cassette_dir=cassette_dir)
gts.append(gt)
except Exception as e:
if vb:
print(
f" ERROR creating targeted test for {sp.qualname}: {e}"
) # noqa: E701
except ImportError:
if vb:
print("Coverage library not found, skipping feedback loop.") # noqa: E701
except Exception as e:
if vb:
print(f"Error during coverage feedback: {e}") # noqa: E701
if vb:
print(f"Generated {len(gts)} test objects.") # noqa: E701
if vb > 1:
print("Running tests to record cassettes (final pass)...") # noqa: E701
trs = _run_tests(gts, interactive=not unsafe, vb=vb)
if vb > 1:
print("Writing test modules...") # noqa: E701
artifacts = []
for sp, tr in zip(sps_targeted if "sps_targeted" in locals() else sps, trs):
if tr is not None:
artifacts.append(TestArtifact(suggestion=sp, run=tr)) # noqa: E701
write_module(
artifacts=artifacts,
output_dir=test_dir,
exception_assertion="type",
include_return_summary=True,
)
return scs, sps, gts, trs