Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234"""Run all E2E unittest suites, one file, or one test by dotted identifier.
Tests share one concurrent daemon; each ``@exclusive`` test gets a daemon of itsown, discarded when it ends, and a few of them run at a time beside the sharedones. An ``@exclusive_class`` class runs in batches, each on one private daemonits tests take in turn. Both queues start with the tests that took longest lasttime, so the slowest never start last."""
import argparsefrom concurrent.futures import ThreadPoolExecutorimport fcntlimport ioimport jsonfrom pathlib import Pathimport resourceimport sysimport timeimport unittest
import harnessimport scratch
SUITE_DIR = Path(__file__).parentsys.path.insert(0, str(SUITE_DIR))sys.path.insert(0, str(SUITE_DIR.parent))from python_test_runner import suite_for # noqa: E402from resources import limits # noqa: E402
DURATIONS = scratch.ROOT / "e2e-durations.json"# A class's tests share a daemon in batches of about this many seconds, so a# big class still spreads over the exclusive lanes.BATCH_SECONDS = 40# What a test with no recorded duration is assumed to take.UNKNOWN_SECONDS = 5
def cases(suite): for index, item in enumerate(suite): if isinstance(item, unittest.TestSuite): yield from cases(item) else: yield item suite._removeTestAtIndex(index)
def marked(test, mark): method = getattr(test, test._testMethodName) return getattr(method, mark, False) or getattr(type(test), mark, False)
def execute(test, daemon): log = daemon.home / "daemon.log" offset = log.stat().st_size if log.exists() else 0 stream = io.StringIO() started = time.monotonic() with harness.on_daemon(daemon): result = unittest.TextTestRunner(stream=stream, verbosity=2).run(test) if not result.wasSuccessful(): # Capture before private teardown removes the daemon's home. setattr(result, "daemon_log", harness.log_tail(log, offset)) try: harness.finish_test() except Exception: result.addError(test, sys.exc_info()) return test.id(), result, time.monotonic() - started
def execute_alone(tests): [test] = tests return [execute(test, harness.Daemon("exclusive"))]
def execute_batch(tests): """Run tests in turn on one serial daemon, replacing it whenever a test spent it or failed.""" outcomes, daemon = [], None for test in tests: daemon = daemon or harness.Daemon("class", serial=True) outcome = execute(test, daemon) outcomes.append(outcome) if daemon.spent or not outcome[1].wasSuccessful(): daemon.discard() daemon = None if daemon: daemon.discard() return outcomes
def batches(tests, durations): """Split each class's tests, slowest first, into batches of about BATCH_SECONDS.""" classes = {} for test in tests: classes.setdefault(type(test), []).append(test) for members in classes.values(): batch, seconds = [], 0.0 for test in members: duration = durations.get(test.id(), UNKNOWN_SECONDS) if batch and seconds + duration > BATCH_SECONDS: yield batch batch, seconds = [], 0.0 batch.append(test) seconds += duration yield batch
def run_suite(suite, jobs, exclusive_jobs): if suite.countTestCases() == 0: raise ValueError("no E2E tests collected") try: durations = json.loads(DURATIONS.read_text()) except (OSError, ValueError): durations = {} selected = sorted(cases(suite), key=lambda test: -durations.get(test.id(), 0)) exclusive = [test for test in selected if marked(test, "_e2e_exclusive")] serial = [ test for test in selected if test not in exclusive and marked(test, "_e2e_class_daemon") ] shared = [test for test in selected if test not in exclusive and test not in serial] jobs_alone = sorted( [(execute_alone, [test]) for test in exclusive] + [(execute_batch, batch) for batch in batches(serial, durations)], key=lambda job: -sum(durations.get(t.id(), UNKNOWN_SECONDS) for t in job[1]), ) daemon = harness.Daemon("shared", concurrent=True) if shared else None if daemon: daemon.boot() daemon.install_shared_features() with ( ThreadPoolExecutor(max_workers=exclusive_jobs) as exclusive_pool, ThreadPoolExecutor(max_workers=jobs) as shared_pool, ): alone = [exclusive_pool.submit(job, tests) for job, tests in jobs_alone] together = [shared_pool.submit(execute, test, daemon) for test in shared] selected.clear() shared.clear() exclusive.clear() serial.clear() jobs_alone.clear() outcomes = [future.result() for future in together] outcomes += [outcome for future in alone for outcome in future.result()] report(outcomes) durations.update({identity: duration for identity, _, duration in outcomes}) DURATIONS.write_text(json.dumps(durations, indent=1, sort_keys=True)) return all(result.wasSuccessful() for _, result, _ in outcomes)
def report(outcomes): for test, result, duration in outcomes: status = "ok" if result.wasSuccessful() else "FAILED" print(f"{status:6} {test} ({duration:.1f}s)", file=sys.stderr) for test, result, _duration in outcomes: for _case, traceback in result.failures + result.errors: print(f"\n{test}:\n{traceback}", file=sys.stderr) if getattr(result, "daemon_log", ""): print(f"daemon log during {test}:\n{result.daemon_log}", file=sys.stderr) print("slowest tests:", file=sys.stderr) for test, _result, duration in sorted(outcomes, key=lambda value: -value[2])[:10]: print(f" {duration:.1f}s {test}", file=sys.stderr) failing = sum(not result.wasSuccessful() for _, result, _ in outcomes) print(f"Ran {len(outcomes)} tests; {failing} failing", file=sys.stderr)
def main(): # Concurrent tests hold hundreds of connections between them, past the 256 # open files a macOS shell starts with. Like Go, lift the soft limit, to at # most macOS's OPEN_MAX. Daemons boot on it unless a test pins their own. _, hard = resource.getrlimit(resource.RLIMIT_NOFILE) resource.setrlimit(resource.RLIMIT_NOFILE, (min(hard, 10240), hard)) workers = limits() parser = argparse.ArgumentParser() parser.add_argument( "targets", nargs="*", help="one or more file.py, file:TestCase.test_method, or area targets", ) parser.add_argument( "-j", "--jobs", type=int, default=workers["shared"], help="concurrent tests on the shared lane (default: %(default)s)", ) parser.add_argument( "-x", "--exclusive-jobs", type=int, default=workers["exclusive"], help="concurrent exclusive daemons (default: %(default)s)", ) args = parser.parse_args() if args.targets: suites = [] for target in args.targets: path = SUITE_DIR / target if path.is_file(): suites.append(suite_for(path)) else: file_part, _, test_name = target.partition(":") path = SUITE_DIR / ( file_part if file_part.endswith(".py") else file_part + "_test.py" ) suites.append(suite_for(path, test_name or None)) suite = unittest.TestSuite(suites) else: suite = unittest.TestSuite( suite_for(path) for path in sorted(SUITE_DIR.glob("*_test.py")) ) if suite.countTestCases() == 0: parser.error("no E2E tests collected") # Concurrent E2E runs compete for the daemons and kernels on this laptop. with open(scratch.ROOT / "e2e.lock", "w") as lock: fcntl.flock(lock, fcntl.LOCK_EX) started = time.monotonic() try: harness.snapshot_daemon() successful = run_suite( suite, max(1, args.jobs), max(1, args.exclusive_jobs) ) finally: harness.shutdown() print( f"daemon boots: {harness.daemon_boots}; boot time: {harness.boot_seconds:.1f}s; restart time: {harness.restart_seconds:.1f}s; wall time: {time.monotonic() - started:.1f}s", flush=True, ) return 0 if successful else 1
if __name__ == "__main__": sys.exit(main())