cpython/Lib/test/libregrtest/main.py at 00c841d61a24042ca907a081051c814506a5f597 · miss-islington/cpython

Repository navigation

History

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

import datetime

import faulthandler

import locale

import os

import platform

import random

import re

import sys

import sysconfig

import tempfile

import time

import unittest

from test.libregrtest.cmdline import _parse_args

from test.libregrtest.runtest import (

findtests, runtest, get_abs_module,

STDTESTS, NOTTESTS, PASSED, FAILED, ENV_CHANGED, SKIPPED, RESOURCE_DENIED,

INTERRUPTED, CHILD_ERROR,

PROGRESS_MIN_TIME, format_test_result)

from test.libregrtest.setup import setup_tests

from test.libregrtest.utils import removepy, count, format_duration, printlist

from test import support

try:

import gc

except ImportError:

gc = None

# When tests are run from the Python build directory, it is best practice

# to keep the test files in a subfolder. This eases the cleanup of leftover

# files using the "make distclean" command.

if sysconfig.is_python_build():

TEMPDIR = sysconfig.get_config_var('abs_builddir')

if TEMPDIR is None:

# bpo-30284: On Windows, only srcdir is available. Using abs_builddir

# mostly matters on UNIX when building Python out of the source tree,

# especially when the source tree is read only.

TEMPDIR = sysconfig.get_config_var('srcdir')

TEMPDIR = os.path.join(TEMPDIR, 'build')

else:

TEMPDIR = tempfile.gettempdir()

TEMPDIR = os.path.abspath(TEMPDIR)

class Regrtest:

"""Execute a test suite.

This also parses command-line options and modifies its behavior

accordingly.

tests -- a list of strings containing test names (optional)

testdir -- the directory in which to look for tests (optional)

Users other than the Python test suite will certainly want to

specify testdir; if it's omitted, the directory containing the

Python test suite is searched for.

If the tests argument is omitted, the tests listed on the

command-line will be used. If that's empty, too, then all *.py

files beginning with test_ will be used.

The other default arguments (verbose, quiet, exclude,

single, randomize, findleaks, use_resources, trace, coverdir,

print_slow, and random_seed) allow programmers calling main()

directly to set the values that would normally be set by flags

on the command line.

"""

def __init__(self):

# Namespace of command line options

self.ns = None

# tests

self.tests = []

self.selected = []

# test results

self.good = []

self.bad = []

self.skipped = []

self.resource_denieds = []

self.environment_changed = []

self.rerun = []

self.first_result = None

self.interrupted = False

# used by --slow

self.test_times = []

# used by --coverage, trace.Trace instance

self.tracer = None

# used by --findleaks, store for gc.garbage

self.found_garbage = []

# used to display the progress bar "[ 3/100]"

self.start_time = time.monotonic()

self.test_count = ''

self.test_count_width = 1

# used by --single

self.next_single_test = None

self.next_single_filename = None

def accumulate_result(self, test, result):

ok, test_time = result

if ok not in (CHILD_ERROR, INTERRUPTED):

self.test_times.append((test_time, test))

if ok == PASSED:

self.good.append(test)

elif ok in (FAILED, CHILD_ERROR):

self.bad.append(test)

elif ok == ENV_CHANGED:

self.environment_changed.append(test)

elif ok == SKIPPED:

self.skipped.append(test)

elif ok == RESOURCE_DENIED:

self.skipped.append(test)

self.resource_denieds.append(test)

elif ok != INTERRUPTED:

raise ValueError("invalid test result: %r" % ok)

def display_progress(self, test_index, test):

if self.ns.quiet:

return

# "[ 51/405/1] test_tcl passed"

line = f"{test_index:{self.test_count_width}}{self.test_count}"

fails = len(self.bad) + len(self.environment_changed)

if fails and not self.ns.pgo:

line = f"{line}/{fails}"

line = f"[{line}] {test}"

# add the system load prefix: "load avg: 1.80 "

if hasattr(os, 'getloadavg'):

load_avg_1min = os.getloadavg()[0]

line = f"load avg: {load_avg_1min:.2f} {line}"

# add the timestamp prefix: "0:01:05 "

test_time = time.monotonic() - self.start_time

test_time = datetime.timedelta(seconds=int(test_time))

line = f"{test_time} {line}"

print(line, flush=True)

def parse_args(self, kwargs):

ns = _parse_args(sys.argv[1:], **kwargs)

if ns.timeout and not hasattr(faulthandler, 'dump_traceback_later'):

print("Warning: The timeout option requires "

"faulthandler.dump_traceback_later", file=sys.stderr)

ns.timeout = None

if ns.threshold is not None and gc is None:

print('No GC available, ignore --threshold.', file=sys.stderr)

ns.threshold = None

if ns.findleaks:

if gc is not None:

# Uncomment the line below to report garbage that is not

# freeable by reference counting alone. By default only

# garbage that is not collectable by the GC is reported.

pass

#gc.set_debug(gc.DEBUG_SAVEALL)

else:

print('No GC available, disabling --findleaks',

file=sys.stderr)

ns.findleaks = False

# Strip .py extensions.

removepy(ns.args)

return ns

def find_tests(self, tests):

self.tests = tests

if self.ns.single:

self.next_single_filename = os.path.join(TEMPDIR, 'pynexttest')

try:

with open(self.next_single_filename, 'r') as fp:

next_test = fp.read().strip()

self.tests = [next_test]

except OSError:

pass

if self.ns.fromfile:

self.tests = []

# regex to match 'test_builtin' in line:

# '0:00:00 [ 4/400] test_builtin -- test_dict took 1 sec'

regex = re.compile(r'\btest_[a-zA-Z0-9_]+\b')

with open(os.path.join(support.SAVEDCWD, self.ns.fromfile)) as fp:

for line in fp:

line = line.split('#', 1)[0]

line = line.strip()

match = regex.search(line)

if match is not None:

self.tests.append(match.group())

removepy(self.tests)

stdtests = STDTESTS[:]

nottests = NOTTESTS.copy()

if self.ns.exclude:

for arg in self.ns.args:

if arg in stdtests:

stdtests.remove(arg)

nottests.add(arg)

self.ns.args = []

# if testdir is set, then we are not running the python tests suite, so

# don't add default tests to be executed or skipped (pass empty values)

if self.ns.testdir:

alltests = findtests(self.ns.testdir, list(), set())

else:

alltests = findtests(self.ns.testdir, stdtests, nottests)

if not self.ns.fromfile:

self.selected = self.tests or self.ns.args or alltests

else:

self.selected = self.tests

if self.ns.single:

self.selected = self.selected[:1]

try:

pos = alltests.index(self.selected[0])

self.next_single_test = alltests[pos + 1]

except IndexError:

pass

# Remove all the selected tests that precede start if it's set.

if self.ns.start:

try:

del self.selected[:self.selected.index(self.ns.start)]

except ValueError:

print("Couldn't find starting test (%s), using all tests"

% self.ns.start, file=sys.stderr)

if self.ns.randomize:

if self.ns.random_seed is None:

self.ns.random_seed = random.randrange(10000000)

random.seed(self.ns.random_seed)

random.shuffle(self.selected)

def list_tests(self):

for name in self.selected:

print(name)

def _list_cases(self, suite):

for test in suite:

if isinstance(test, unittest.loader._FailedTest):

continue

if isinstance(test, unittest.TestSuite):

self._list_cases(test)

elif isinstance(test, unittest.TestCase):

if support.match_test(test):

print(test.id())

def list_cases(self):

support.verbose = False

support.set_match_tests(self.ns.match_tests)

for test in self.selected:

abstest = get_abs_module(self.ns, test)

try:

suite = unittest.defaultTestLoader.loadTestsFromName(abstest)

self._list_cases(suite)

except unittest.SkipTest:

self.skipped.append(test)

if self.skipped:

print(file=sys.stderr)

print(count(len(self.skipped), "test"), "skipped:", file=sys.stderr)

printlist(self.skipped, file=sys.stderr)

def rerun_failed_tests(self):

self.ns.verbose = True

self.ns.failfast = False

self.ns.verbose3 = False

self.first_result = self.get_tests_result()

print()

print("Re-running failed tests in verbose mode")

self.rerun = self.bad[:]

for test in self.rerun:

print("Re-running test %r in verbose mode" % test, flush=True)

try:

self.ns.verbose = True

ok = runtest(self.ns, test)

except KeyboardInterrupt:

self.interrupted = True

# print a newline separate from the ^C

print()

break

else:

if ok[0] in {PASSED, ENV_CHANGED, SKIPPED, RESOURCE_DENIED}:

self.bad.remove(test)

else:

if self.bad:

print(count(len(self.bad), 'test'), "failed again:")

printlist(self.bad)

self.display_result()

def display_result(self):

# If running the test suite for PGO then no one cares about results.

if self.ns.pgo:

return

print()

print("== Tests result: %s ==" % self.get_tests_result())

if self.interrupted:

print()

# print a newline after ^C

print("Test suite interrupted by signal SIGINT.")

executed = set(self.good) | set(self.bad) | set(self.skipped)

omitted = set(self.selected) - executed

print(count(len(omitted), "test"), "omitted:")

printlist(omitted)

if self.good and not self.ns.quiet:

print()

if (not self.bad

and not self.skipped

and not self.interrupted

and len(self.good) > 1):

print("All", end=' ')

print(count(len(self.good), "test"), "OK.")

if self.ns.print_slow:

self.test_times.sort(reverse=True)

print()

print("10 slowest tests:")

for time, test in self.test_times[:10]:

print("- %s: %s" % (test, format_duration(time)))

if self.bad:

print()

print(count(len(self.bad), "test"), "failed:")

printlist(self.bad)

if self.environment_changed:

print()

print("{} altered the execution environment:".format(

count(len(self.environment_changed), "test")))

printlist(self.environment_changed)

if self.skipped and not self.ns.quiet:

print()

print(count(len(self.skipped), "test"), "skipped:")

printlist(self.skipped)

if self.rerun:

print()

print("%s:" % count(len(self.rerun), "re-run test"))

printlist(self.rerun)

def run_tests_sequential(self):

if self.ns.trace:

import trace

self.tracer = trace.Trace(trace=False, count=True)

save_modules = sys.modules.keys()

print("Run tests sequentially")

previous_test = None

for test_index, test in enumerate(self.tests, 1):

start_time = time.monotonic()

text = test

if previous_test:

text = '%s -- %s' % (text, previous_test)

self.display_progress(test_index, text)

if self.tracer:

# If we're tracing code coverage, then we don't exit with status

# if on a false return value from main.

cmd = ('result = runtest(self.ns, test); '

'self.accumulate_result(test, result)')

ns = dict(locals())

self.tracer.runctx(cmd, globals=globals(), locals=ns)

result = ns['result']

else:

try:

result = runtest(self.ns, test)

except KeyboardInterrupt:

self.interrupted = True

self.accumulate_result(test, (INTERRUPTED, None))

break

else:

self.accumulate_result(test, result)

previous_test = format_test_result(test, result[0])

test_time = time.monotonic() - start_time

if test_time >= PROGRESS_MIN_TIME:

previous_test = "%s in %s" % (previous_test, format_duration(test_time))

elif result[0] == PASSED:

# be quiet: say nothing if the test passed shortly

previous_test = None

if self.ns.findleaks:

gc.collect()

if gc.garbage:

print("Warning: test created", len(gc.garbage), end=' ')

print("uncollectable object(s).")

# move the uncollectable objects somewhere so we don't see

# them again

self.found_garbage.extend(gc.garbage)

del gc.garbage[:]

# Unload the newly imported modules (best effort finalization)

for module in sys.modules.keys():

if module not in save_modules and module.startswith("test."):

support.unload(module)

if previous_test:

print(previous_test)

def _test_forever(self, tests):

while True:

for test in tests:

yield test

if self.bad:

return

if self.ns.fail_env_changed and self.environment_changed:

return

def display_header(self):

# Print basic platform information

print("==", platform.python_implementation(), *sys.version.split())

print("==", platform.platform(aliased=True),

"%s-endian" % sys.byteorder)

print("== cwd:", os.getcwd())

cpu_count = os.cpu_count()

if cpu_count:

print("== CPU count:", cpu_count)

print("== encodings: locale=%s, FS=%s"

% (locale.getpreferredencoding(False),

sys.getfilesystemencoding()))

def get_tests_result(self):

result = []

if self.bad:

result.append("FAILURE")

elif self.ns.fail_env_changed and self.environment_changed:

result.append("ENV CHANGED")

if self.interrupted:

result.append("INTERRUPTED")

if not result:

result.append("SUCCESS")

result = ', '.join(result)

if self.first_result:

result = '%s then %s' % (self.first_result, result)

return result

def run_tests(self):

# For a partial run, we do not need to clutter the output.

if (self.ns.header

or not(self.ns.pgo or self.ns.quiet or self.ns.single

or self.tests or self.ns.args)):

self.display_header()

if self.ns.randomize:

print("Using random seed", self.ns.random_seed)

if self.ns.forever:

self.tests = self._test_forever(list(self.selected))

self.test_count = ''

self.test_count_width = 3

else:

self.tests = iter(self.selected)

self.test_count = '/{}'.format(len(self.selected))

self.test_count_width = len(self.test_count) - 1

if self.ns.use_mp:

from test.libregrtest.runtest_mp import run_tests_multiprocess

run_tests_multiprocess(self)

else:

self.run_tests_sequential()

def finalize(self):

if self.next_single_filename:

if self.next_single_test:

with open(self.next_single_filename, 'w') as fp:

fp.write(self.next_single_test + '\n')

else:

os.unlink(self.next_single_filename)

if self.tracer:

r = self.tracer.results()

r.write_results(show_missing=True, summary=True,

coverdir=self.ns.coverdir)

print()

duration = time.monotonic() - self.start_time

print("Total duration: %s" % format_duration(duration))

print("Tests result: %s" % self.get_tests_result())

if self.ns.runleaks:

os.system("leaks %d" % os.getpid())

def main(self, tests=None, **kwargs):

global TEMPDIR

if sysconfig.is_python_build():

try:

os.mkdir(TEMPDIR)

except FileExistsError:

pass

# Define a writable temp dir that will be used as cwd while running

# the tests. The name of the dir includes the pid to allow parallel

# testing (see the -j option).

test_cwd = 'test_python_{}'.format(os.getpid())

test_cwd = os.path.join(TEMPDIR, test_cwd)

# Run the tests in a context manager that temporarily changes the CWD to a

# temporary and writable directory. If it's not possible to create or

# change the CWD, the original CWD will be used. The original CWD is

# available from support.SAVEDCWD.

with support.temp_cwd(test_cwd, quiet=True):

self._main(tests, kwargs)

def _main(self, tests, kwargs):

self.ns = self.parse_args(kwargs)

if self.ns.huntrleaks:

warmup, repetitions, _ = self.ns.huntrleaks

if warmup < 1 or repetitions < 1:

msg = ("Invalid values for the --huntrleaks/-R parameters. The "

"number of warmups and repetitions must be at least 1 "

"each (1:1).")

print(msg, file=sys.stderr, flush=True)

sys.exit(2)

if self.ns.slaveargs is not None:

from test.libregrtest.runtest_mp import run_tests_slave

run_tests_slave(self.ns.slaveargs)

if self.ns.wait:

input("Press any key to continue...")

support.PGO = self.ns.pgo

setup_tests(self.ns)

self.find_tests(tests)

if self.ns.list_tests:

self.list_tests()

sys.exit(0)

if self.ns.list_cases:

self.list_cases()

sys.exit(0)

self.run_tests()

self.display_result()

if self.ns.verbose2 and self.bad:

self.rerun_failed_tests()

self.finalize()

if self.bad:

sys.exit(2)

if self.interrupted:

sys.exit(130)

if self.ns.fail_env_changed and self.environment_changed:

sys.exit(3)

sys.exit(0)

def main(tests=None, **kwargs):

"""Run the Python suite."""

Regrtest().main(tests=tests, **kwargs)

FilesExpand file tree

main.py

Latest commit

History

main.py

File metadata and controls