Coverage for local/lib/python2.7/site-packages/sage/doctest/control.py : 64%

""" 

Classes involved in doctesting 

This module controls the various classes involved in doctesting. 

AUTHORS: 

- David Roe (2012-03-27) -- initial version, based on Robert Bradshaw's code. 

""" 

#***************************************************************************** 

# Copyright (C) 2012 David Roe <roed.math@gmail.com> 

# Robert Bradshaw <robertwb@gmail.com> 

# William Stein <wstein@gmail.com> 

# Copyright (C) 2016 Jeroen Demeyer <jdemeyer@cage.ugent.be> 

# 

# This program is free software: you can redistribute it and/or modify 

# it under the terms of the GNU General Public License as published by 

# the Free Software Foundation, either version 2 of the License, or 

# (at your option) any later version. 

# http://www.gnu.org/licenses/ 

#***************************************************************************** 

from __future__ import absolute_import, division, print_function 

import random, os, sys, time, json, re, types 

import atexit 

import shutil 

import socket 

import six 

import sage.misc.flatten 

from sage.structure.sage_object import SageObject 

from sage.env import DOT_SAGE, SAGE_LIB, SAGE_SRC 

from sage.misc.temporary_file import tmp_dir 

from cysignals.signals import AlarmInterrupt, init_cysignals 

from .sources import FileDocTestSource, DictAsObject 

from .forker import DocTestDispatcher 

from .reporting import DocTestReporter 

from .util import Timer, count_noun, dict_difference 

from .external import external_software, available_software 

nodoctest_regex = re.compile(r'\s*(#+|%+|r"+|"+|\.\.)\s*nodoctest') 

optionaltag_regex = re.compile(r'^\w+$') 

# Optional tags which are always automatically added 

auto_optional_tags = set(['py2' if six.PY2 else 'py3']) 

class DocTestDefaults(SageObject): 

""" 

This class is used for doctesting the Sage doctest module. 

It fills in attributes to be the same as the defaults defined in 

``SAGE_LOCAL/bin/sage-runtests``, expect for a few places, 

which is mostly to make doctesting more predictable. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults 

sage: D = DocTestDefaults() 

sage: D 

DocTestDefaults() 

sage: D.timeout 

-1 

Keyword arguments become attributes:: 

sage: D = DocTestDefaults(timeout=100) 

sage: D 

DocTestDefaults(timeout=100) 

sage: D.timeout 

100 

""" 

def __init__(self, **kwds): 

""" 

Edit these parameters after creating an instance. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults 

sage: D = DocTestDefaults() 

sage: 'sage' in D.optional 

True 

""" 

self.nthreads = 1 

self.serial = False 

self.timeout = -1 

self.all = False 

self.logfile = None 

self.sagenb = False 

self.long = False 

self.warn_long = None 

self.optional = set(['sage']) | auto_optional_tags 

self.randorder = None 

self.global_iterations = 1 # sage-runtests default is 0 

self.file_iterations = 1 # sage-runtests default is 0 

self.initial = False 

self.exitfirst = False 

self.force_lib = False 

self.abspath = True # sage-runtests default is False 

self.verbose = False 

self.debug = False 

self.only_errors = False 

self.gdb = False 

self.valgrind = False 

self.massif = False 

self.cachegrind = False 

self.omega = False 

self.failed = False 

self.new = False 

self.show_skipped = False 

# We don't want to use the real stats file by default so that 

# we don't overwrite timings for the actual running doctests. 

self.stats_path = os.path.join(DOT_SAGE, "timings_dt_test.json") 

self.coverage = False 

self.__dict__.update(kwds) 

def _repr_(self): 

""" 

Return the print representation. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults 

sage: DocTestDefaults(timeout=100, foobar="hello") 

DocTestDefaults(foobar='hello', timeout=100) 

""" 

s = "DocTestDefaults(" 

for k in sorted(dict_difference(self.__dict__, DocTestDefaults().__dict__).keys()): 

if s[-1] != "(": 

s += ", " 

s += str(k) + "=" + repr(getattr(self,k)) 

s += ")" 

return s 

def __eq__(self, other): 

""" 

Comparison by __dict__. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults 

sage: DD1 = DocTestDefaults(long=True) 

sage: DD2 = DocTestDefaults(long=True) 

sage: DD1 == DD2 

True 

""" 

if not isinstance(other, DocTestDefaults): 

return False 

return self.__dict__ == other.__dict__ 

def __ne__(self, other): 

""" 

Test for unequality. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults 

sage: DD1 = DocTestDefaults(long=True) 

sage: DD2 = DocTestDefaults(long=True) 

sage: DD1 != DD2 

False 

""" 

return not (self == other) 

def skipdir(dirname): 

""" 

Return True if and only if the directory ``dirname`` should not be 

doctested. 

EXAMPLES:: 

sage: from sage.doctest.control import skipdir 

sage: skipdir(sage.env.SAGE_SRC) 

False 

sage: skipdir(os.path.join(sage.env.SAGE_SRC, "sage", "doctest", "tests")) 

True 

""" 

if os.path.exists(os.path.join(dirname, "nodoctest.py")): 

return True 

return False 

def skipfile(filename): 

""" 

Return True if and only if the file ``filename`` should not be 

doctested. 

EXAMPLES:: 

sage: from sage.doctest.control import skipfile 

sage: skipfile("skipme.c") 

True 

sage: filename = tmp_filename(ext=".pyx") 

sage: skipfile(filename) 

False 

sage: with open(filename, "w") as f: 

....: _ = f.write("# nodoctest") 

sage: skipfile(filename) 

True 

""" 

base, ext = os.path.splitext(filename) 

if ext not in ('.py', '.pyx', '.pxd', '.pxi', '.sage', '.spyx', '.rst', '.tex'): 

return True 

with open(filename) as F: 

line_count = 0 

for line in F: 

if nodoctest_regex.match(line): 

return True 

line_count += 1 

if line_count >= 10: 

break 

return False 

class Logger(object): 

r""" 

File-like object which implements writing to multiple files at 

once. 

EXAMPLES:: 

sage: from sage.doctest.control import Logger 

sage: with open(tmp_filename(), "w+") as t: 

....: L = Logger(sys.stdout, t) 

....: _ = L.write("hello world\n") 

....: _ = t.seek(0) 

....: t.read() 

hello world 

'hello world\n' 

""" 

def __init__(self, *files): 

""" 

Initialize the logger for writing to all files in ``files``. 

TESTS:: 

sage: from sage.doctest.control import Logger 

sage: Logger().write("hello world\n") # no-op 

""" 

self.files = list(files) 

def write(self, x): 

r""" 

Write ``x`` to all files. 

TESTS:: 

sage: from sage.doctest.control import Logger 

sage: Logger(sys.stdout).write("hello world\n") 

hello world 

""" 

for f in self.files: 

f.write(x) 

def flush(self): 

""" 

Flush all files. 

TESTS:: 

sage: from sage.doctest.control import Logger 

sage: Logger(sys.stdout).flush() 

""" 

for f in self.files: 

f.flush() 

class DocTestController(SageObject): 

""" 

This class controls doctesting of files. 

After creating it with appropriate options, call the :meth:`run` method to run the doctests. 

""" 

def __init__(self, options, args): 

""" 

Initialization. 

INPUT: 

- options -- either options generated from the command line by SAGE_LOCAL/bin/sage-runtests 

or a DocTestDefaults object (possibly with some entries modified) 

- args -- a list of filenames to doctest 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DC = DocTestController(DocTestDefaults(), []) 

sage: DC 

DocTest Controller 

""" 

# First we modify options to take environment variables into 

# account and check compatibility of the user's specified 

# options. 

if options.timeout < 0: 

if options.gdb or options.debug: 

# Interactive debuggers: "infinite" timeout 

options.timeout = 0 

elif options.valgrind or options.massif or options.cachegrind or options.omega: 

# Non-interactive debuggers: 48 hours 

options.timeout = int(os.getenv('SAGE_TIMEOUT_VALGRIND', 48 * 60 * 60)) 

elif options.long: 

options.timeout = int(os.getenv('SAGE_TIMEOUT_LONG', 30 * 60)) 

else: 

options.timeout = int(os.getenv('SAGE_TIMEOUT', 5 * 60)) 

if options.coverage: 

# Coverage testing makes things extra slow, so just double 

# whatever the timeout would be otherwise 

options.timeout *= 2 

if options.nthreads == 0: 

options.nthreads = int(os.getenv('SAGE_NUM_THREADS_PARALLEL',1)) 

if options.failed and not (args or options.new or options.sagenb): 

# If the user doesn't specify any files then we rerun all failed files. 

options.all = True 

if options.global_iterations == 0: 

options.global_iterations = int(os.environ.get('SAGE_TEST_GLOBAL_ITER', 1)) 

if options.file_iterations == 0: 

options.file_iterations = int(os.environ.get('SAGE_TEST_ITER', 1)) 

if options.debug: 

if options.nthreads > 1: 

print("Debugging requires single-threaded operation, setting number of threads to 1.") 

if options.logfile: 

print("Debugging is not compatible with logging, disabling logfile.") 

options.serial = True 

options.logfile = None 

if options.serial: 

options.nthreads = 1 

if options.verbose: 

options.show_skipped = True 

if isinstance(options.optional, six.string_types): 

s = options.optional.lower() 

options.optional = set(s.split(',')) 

if "all" in options.optional: 

# Special case to run all optional tests 

options.optional = True 

else: 

# We replace the 'optional' tag by all optional 

# packages for which the installed version matches the 

# latest available version (this implies in particular 

# that the package is actually installed). 

if 'optional' in options.optional: 

options.optional.discard('optional') 

from sage.misc.package import list_packages 

for pkg in list_packages('optional', local=True).values(): 

if pkg['installed_version'] == pkg['remote_version']: 

options.optional.add(pkg['name']) 

# Check that all tags are valid 

for o in options.optional: 

if not optionaltag_regex.search(o): 

raise ValueError('invalid optional tag {!r}'.format(o)) 

options.optional |= auto_optional_tags 

self.options = options 

self.files = args 

if options.logfile: 

try: 

self.logfile = open(options.logfile, 'a') 

except IOError: 

print("Unable to open logfile {!r}\nProceeding without logging.".format(options.logfile)) 

self.logfile = None 

else: 

self.logfile = None 

# Flush any diagnostic messages we just printed 

sys.stdout.flush() 

sys.stderr.flush() 

# In serial mode, we run just one process. Then the doctests 

# will interfere with the output logging (both use stdout). 

# To solve this, we create real_stdout which will always 

# write to the actual standard output, regardless of 

# redirections. 

if options.serial: 

self._real_stdout = os.fdopen(os.dup(sys.stdout.fileno()), "w") 

self._close_stdout = True 

else: 

# Parallel mode: no special tricks needed 

self._real_stdout = sys.stdout 

self._close_stdout = False 

if self.logfile is None: 

self.logger = self._real_stdout 

else: 

self.logger = Logger(self._real_stdout, self.logfile) 

self.stats = {} 

self.load_stats(options.stats_path) 

self._init_warn_long() 

self._init_coverage() 

def __del__(self): 

if getattr(self, 'logfile', None) is not None: 

self.logfile.close() 

if getattr(self, '_close_stdout', False): 

self._real_stdout.close() 

def _init_warn_long(self): 

""" 

Pick a suitable default for the ``--warn-long`` option if not specified. 

It is desirable to have all tests (even ``# long`` ones) 

finish in less than about 5 seconds. Longer tests typically 

don't add coverage, they just make testing slow. 

The default used here is 60 seconds on a modern computer. It 

should eventually be lowered to 5 seconds, but its best to 

boil the frog slowly. 

The stored timings are used to adjust this limit according to 

the machine running the tests. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DC = DocTestController(DocTestDefaults(), []) 

sage: DC.options.warn_long = 5.0 

sage: DC._init_warn_long() 

sage: DC.options.warn_long # existing command-line options are not changed 

5.00000000000000 

""" 

if self.options.warn_long is not None: # Specified on the command line 

return 

try: 

self.options.warn_long = 60.0 * self.second_on_modern_computer() 

except RuntimeError as err: 

if not sage.doctest.DOCTEST_MODE: 

print(err) # No usable timing information 

def _init_coverage(self): 

# TODO: Determine what should be passed to source depending on 

# which packages are being tested 

self.coverage_dir = None 

self.coverage = None 

if self.options.coverage: 

import coverage 

from sage.env import SAGE_SRC, SAGE_LIB 

if not self.options.serial: 

cov_dirname = '.coverage.{}.{}'.format( 

socket.gethostname(), os.getpid()) 

cov_dir = os.path.join(os.getcwd(), cov_dirname) 

if os.path.exists(cov_dir): 

shutil.rmtree(cov_dir) 

os.mkdir(cov_dir) 

data_file = os.path.join(cov_dir, 'coverage') 

self.coverage_dir = cov_dir 

def cleanup_coverage_dir(): 

try: 

if os.path.exists(cov_dir): 

shutil.rmtree(cov_dir) 

except Exception: 

pass 

#atexit.register(cleanup_coverage_dir) 

else: 

data_file = os.path.join(os.getcwd(), '.coverage') 

self.coverage = coverage.Coverage( 

data_file=data_file, 

config_file=os.path.join(SAGE_SRC, '.coveragerc'), 

source=[os.path.join(SAGE_LIB, 'sage')] 

) 

self.coverage.erase() 

def collect_coverage(self): 

if not (self.coverage and self.coverage.config.parallel): 

return 

self.coverage.combine() 

self.coverage.save() 

os.rename(self.coverage.config.data_file, 

os.path.join(os.path.dirname(self.coverage_dir), 

'.coverage')) 

#shutil.rmtree(self.coverage_dir) 

def second_on_modern_computer(self): 

""" 

Return the wall time equivalent of a second on a modern computer. 

OUTPUT: 

Float. The wall time on your computer that would be equivalent 

to one second on a modern computer. Unless you have kick-ass 

hardware this should always be >= 1.0. Raises a 

``RuntimeError`` if there are no stored timings to use as 

benchmark. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DC = DocTestController(DocTestDefaults(), []) 

sage: DC.second_on_modern_computer() # not tested 

""" 

if len(self.stats) == 0: 

raise RuntimeError('no stored timings available') 

success = [] 

failed = [] 

for mod in self.stats.values(): 

if mod.get('failed', False): 

failed.append(mod['walltime']) 

else: 

success.append(mod['walltime']) 

if len(success) < 2500: 

raise RuntimeError('too few successful tests, not using stored timings') 

if len(failed) > 20: 

raise RuntimeError('too many failed tests, not using stored timings') 

expected = 12800.0 # Core i7 Quad-Core 2014 

return sum(success) / expected 

def _repr_(self): 

""" 

String representation. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DC = DocTestController(DocTestDefaults(), []) 

sage: repr(DC) # indirect doctest 

'DocTest Controller' 

""" 

return "DocTest Controller" 

def load_stats(self, filename): 

""" 

Load stats from the most recent run(s). 

Stats are stored as a JSON file, and include information on 

which files failed tests and the walltime used for execution 

of the doctests. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DC = DocTestController(DocTestDefaults(), []) 

sage: import json 

sage: filename = tmp_filename() 

sage: with open(filename, 'w') as stats_file: 

....: json.dump({'sage.doctest.control':{'walltime':1.0r}}, stats_file) 

sage: DC.load_stats(filename) 

sage: DC.stats['sage.doctest.control'] 

{u'walltime': 1.0} 

If the file doesn't exist, nothing happens. If there is an 

error, print a message. In any case, leave the stats alone:: 

sage: d = tmp_dir() 

sage: DC.load_stats(os.path.join(d)) # Cannot read a directory 

Error loading stats from ... 

sage: DC.load_stats(os.path.join(d, "no_such_file")) 

sage: DC.stats['sage.doctest.control'] 

{u'walltime': 1.0} 

""" 

# Simply ignore non-existing files 

if not os.path.exists(filename): 

return 

try: 

with open(filename) as stats_file: 

self.stats.update(json.load(stats_file)) 

except Exception: 

self.log("Error loading stats from %s"%filename) 

def save_stats(self, filename): 

""" 

Save stats from the most recent run as a JSON file. 

WARNING: This function overwrites the file. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DC = DocTestController(DocTestDefaults(), []) 

sage: DC.stats['sage.doctest.control'] = {'walltime':1.0r} 

sage: filename = tmp_filename() 

sage: DC.save_stats(filename) 

sage: import json 

sage: with open(filename) as f: 

....: D = json.load(f) 

sage: D['sage.doctest.control'] 

{u'walltime': 1.0} 

""" 

from sage.misc.temporary_file import atomic_write 

with atomic_write(filename) as stats_file: 

json.dump(self.stats, stats_file) 

def log(self, s, end="\n"): 

""" 

Log the string ``s + end`` (where ``end`` is a newline by default) 

to the logfile and print it to the standard output. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DD = DocTestDefaults(logfile=tmp_filename()) 

sage: DC = DocTestController(DD, []) 

sage: DC.log("hello world") 

hello world 

sage: DC.logfile.close() 

sage: with open(DD.logfile) as f: 

....: print(f.read()) 

hello world 

In serial mode, check that logging works even if ``stdout`` is 

redirected:: 

sage: DD = DocTestDefaults(logfile=tmp_filename(), serial=True) 

sage: DC = DocTestController(DD, []) 

sage: from sage.doctest.forker import SageSpoofInOut 

sage: with open(os.devnull, 'w') as devnull: 

....: S = SageSpoofInOut(devnull) 

....: S.start_spoofing() 

....: DC.log("hello world") 

....: S.stop_spoofing() 

hello world 

sage: DC.logfile.close() 

sage: with open(DD.logfile) as f: 

....: print(f.read()) 

hello world 

Check that no duplicate logs appear, even when forking (:trac:`15244`):: 

sage: DD = DocTestDefaults(logfile=tmp_filename()) 

sage: DC = DocTestController(DD, []) 

sage: DC.log("hello world") 

hello world 

sage: if os.fork() == 0: 

....: DC.logfile.close() 

....: os._exit(0) 

sage: DC.logfile.close() 

sage: with open(DD.logfile) as f: 

....: print(f.read()) 

hello world 

""" 

self.logger.write(s + end) 

self.logger.flush() 

def test_safe_directory(self, dir=None): 

""" 

Test that the given directory is safe to run Python code from. 

We use the check added to Python for this, which gives a 

warning when the current directory is considered unsafe. We promote 

this warning to an error with ``-Werror``. See 

``sage/tests/cmdline.py`` for a doctest that this works, see 

also :trac:`13579`. 

TESTS:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DD = DocTestDefaults() 

sage: DC = DocTestController(DD, []) 

sage: DC.test_safe_directory() 

sage: d = os.path.join(tmp_dir(), "test") 

sage: os.mkdir(d) 

sage: os.chmod(d, 0o777) 

sage: DC.test_safe_directory(d) 

Traceback (most recent call last): 

... 

RuntimeError: refusing to run doctests... 

""" 

import subprocess 

with open(os.devnull, 'w') as dev_null: 

if subprocess.call(['python', '-Werror', '-c', ''], 

stdout=dev_null, stderr=dev_null, cwd=dir) != 0: 

raise RuntimeError( 

"refusing to run doctests from the current " 

"directory '{}' since untrusted users could put files in " 

"this directory, making it unsafe to run Sage code from" 

.format(os.getcwd())) 

def create_run_id(self): 

""" 

Creates the run id. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: DC = DocTestController(DocTestDefaults(), []) 

sage: DC.create_run_id() 

Running doctests with ID ... 

""" 

self.run_id = time.strftime('%Y-%m-%d-%H-%M-%S-') + "%08x" % random.getrandbits(32) 

self.log("Running doctests with ID %s."%self.run_id) 

def add_files(self): 

r""" 

Checks for the flags '--all', '--new' and '--sagenb'. 

For each one present, this function adds the appropriate directories and files to the todo list. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: from sage.env import SAGE_SRC 

sage: import os 

sage: log_location = os.path.join(SAGE_TMP, 'control_dt_log.log') 

sage: DD = DocTestDefaults(all=True, logfile=log_location) 

sage: DC = DocTestController(DD, []) 

sage: DC.add_files() 

Doctesting entire Sage library. 

sage: os.path.join(SAGE_SRC, 'sage') in DC.files 

True 

:: 

sage: DD = DocTestDefaults(new = True) 

sage: DC = DocTestController(DD, []) 

sage: DC.add_files() 

Doctesting ... 

:: 

sage: DD = DocTestDefaults(sagenb = True) 

sage: DC = DocTestController(DD, []) 

sage: DC.add_files() # py2 

Doctesting the Sage notebook. 

sage: DC.files[0][-6:] # py2 

'sagenb' 

""" 

opj = os.path.join 

from sage.env import SAGE_SRC, SAGE_DOC_SRC, SAGE_ROOT 

DOT_GIT= opj(SAGE_ROOT, '.git') 

have_git = os.path.exists(DOT_GIT) 

def all_files(): 

self.files.append(opj(SAGE_SRC, 'sage')) 

# Don't run these tests when not in the git repository; they are 

# of interest for building sage, but not for runtime behavior and 

# don't make sense to run outside a build environment 

if have_git: 

self.files.append(opj(SAGE_SRC, 'sage_setup')) 

self.files.append(SAGE_DOC_SRC) 

self.options.sagenb = True 

if self.options.all or (self.options.new and not have_git): 

self.log("Doctesting entire Sage library.") 

all_files() 

elif self.options.new and have_git: 

# Get all files changed in the working repo. 

self.log("Doctesting files changed since last git commit") 

import subprocess 

change = subprocess.check_output(["git", 

"--git-dir=" + DOT_GIT, 

"--work-tree=" + SAGE_ROOT, 

"status", 

"--porcelain"]) 

change = change.decode('utf-8') 

for line in change.split("\n"): 

if not line: 

continue 

data = line.strip().split(' ') 

status, filename = data[0], data[-1] 

if (set(status).issubset("MARCU") 

and filename.startswith("src/sage") 

and (filename.endswith(".py") or filename.endswith(".pyx"))): 

self.files.append(os.path.relpath(opj(SAGE_ROOT,filename))) 

if self.options.sagenb: 

if six.PY3: 

if not self.options.all: 

self.log("Skipping doctesting of the Sage notebook: " 

"not installed on Python 3") 

return 

if not self.options.all: 

self.log("Doctesting the Sage notebook.") 

from pkg_resources import Requirement, working_set 

sagenb_loc = working_set.find(Requirement.parse('sagenb')).location 

self.files.append(opj(sagenb_loc, 'sagenb')) 

def expand_files_into_sources(self): 

r""" 

Expands ``self.files``, which may include directories, into a 

list of :class:`sage.doctest.FileDocTestSource` 

This function also handles the optional command line option. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: from sage.env import SAGE_SRC 

sage: import os 

sage: dirname = os.path.join(SAGE_SRC, 'sage', 'doctest') 

sage: DD = DocTestDefaults(optional='all') 

sage: DC = DocTestController(DD, [dirname]) 

sage: DC.expand_files_into_sources() 

sage: len(DC.sources) 

11 

sage: DC.sources[0].options.optional 

True 

:: 

sage: DD = DocTestDefaults(optional='magma,guava') 

sage: DC = DocTestController(DD, [dirname]) 

sage: DC.expand_files_into_sources() 

sage: sorted(DC.sources[0].options.optional) # abs tol 1 

['guava', 'magma', 'py3'] 

We check that files are skipped appropriately:: 

sage: dirname = tmp_dir() 

sage: filename = os.path.join(dirname, 'not_tested.py') 

sage: with open(filename, 'w') as f: 

....: _ = f.write("#"*80 + "\n\n\n\n## nodoctest\n sage: 1+1\n 4") 

sage: DC = DocTestController(DD, [dirname]) 

sage: DC.expand_files_into_sources() 

sage: DC.sources 

[] 

The directory ``sage/doctest/tests`` contains ``nodoctest.py`` 

but the files should still be tested when that directory is 

explicitly given (as opposed to being recursed into):: 

sage: DC = DocTestController(DD, [os.path.join(SAGE_SRC, 'sage', 'doctest', 'tests')]) 

sage: DC.expand_files_into_sources() 

sage: len(DC.sources) >= 10 

True 

""" 

def expand(): 

for path in self.files: 

if os.path.isdir(path): 

for root, dirs, files in os.walk(path): 

for dir in list(dirs): 

if dir[0] == "." or skipdir(os.path.join(root,dir)): 

dirs.remove(dir) 

for file in files: 

if not skipfile(os.path.join(root,file)): 

yield os.path.join(root, file) 

else: 

# the user input this file explicitly, so we don't skip it 

yield path 

self.sources = [FileDocTestSource(path, self.options) for path in expand()] 

def filter_sources(self): 

""" 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: from sage.env import SAGE_SRC 

sage: import os 

sage: dirname = os.path.join(SAGE_SRC, 'sage', 'doctest') 

sage: DD = DocTestDefaults(failed=True) 

sage: DC = DocTestController(DD, [dirname]) 

sage: DC.expand_files_into_sources() 

sage: for i, source in enumerate(DC.sources): 

....: DC.stats[source.basename] = {'walltime': 0.1*(i+1)} 

sage: DC.stats['sage.doctest.control'] = {'failed':True,'walltime':1.0} 

sage: DC.filter_sources() 

Only doctesting files that failed last test. 

sage: len(DC.sources) 

1 

""" 

# Filter the sources to only include those with failing doctests if the --failed option is passed 

if self.options.failed: 

self.log("Only doctesting files that failed last test.") 

def is_failure(source): 

basename = source.basename 

return basename not in self.stats or self.stats[basename].get('failed') 

self.sources = [x for x in self.sources if is_failure(x)] 

def sort_sources(self): 

r""" 

This function sorts the sources so that slower doctests are run first. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: from sage.env import SAGE_SRC 

sage: import os 

sage: dirname = os.path.join(SAGE_SRC, 'sage', 'doctest') 

sage: DD = DocTestDefaults(nthreads=2) 

sage: DC = DocTestController(DD, [dirname]) 

sage: DC.expand_files_into_sources() 

sage: DC.sources.sort(key=lambda s:s.basename) 

sage: for i, source in enumerate(DC.sources): 

....: DC.stats[source.basename] = {'walltime': 0.1*(i+1)} 

sage: DC.sort_sources() 

Sorting sources by runtime so that slower doctests are run first.... 

sage: print("\n".join([source.basename for source in DC.sources])) 

sage.doctest.util 

sage.doctest.test 

sage.doctest.sources 

sage.doctest.reporting 

sage.doctest.parsing 

sage.doctest.forker 

sage.doctest.fixtures 

sage.doctest.external 

sage.doctest.control 

sage.doctest.all 

sage.doctest 

""" 

if self.options.nthreads > 1 and len(self.sources) > self.options.nthreads: 

self.log("Sorting sources by runtime so that slower doctests are run first....") 

default = dict(walltime=0) 

def sort_key(source): 

basename = source.basename 

return -self.stats.get(basename, default).get('walltime'), basename 

self.sources = sorted(self.sources, key=sort_key) 

def run_doctests(self): 

""" 

Actually runs the doctests. 

This function is called by :meth:`run`. 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: from sage.env import SAGE_SRC 

sage: import os 

sage: dirname = os.path.join(SAGE_SRC, 'sage', 'rings', 'homset.py') 

sage: DD = DocTestDefaults() 

sage: DC = DocTestController(DD, [dirname]) 

sage: DC.expand_files_into_sources() 

sage: DC.run_doctests() 

Doctesting 1 file. 

sage -t .../sage/rings/homset.py 

[... tests, ... s] 

---------------------------------------------------------------------- 

All tests passed! 

---------------------------------------------------------------------- 

Total time for all tests: ... seconds 

cpu time: ... seconds 

cumulative wall time: ... seconds 

""" 

nfiles = 0 

nother = 0 

for F in self.sources: 

if isinstance(F, FileDocTestSource): 

nfiles += 1 

else: 

nother += 1 

if self.sources: 

filestr = ", ".join(([count_noun(nfiles, "file")] if nfiles else []) + 

([count_noun(nother, "other source")] if nother else [])) 

threads = " using %s threads"%(self.options.nthreads) if self.options.nthreads > 1 else "" 

iterations = [] 

if self.options.global_iterations > 1: 

iterations.append("%s global iterations"%(self.options.global_iterations)) 

if self.options.file_iterations > 1: 

iterations.append("%s file iterations"%(self.options.file_iterations)) 

iterations = ", ".join(iterations) 

if iterations: 

iterations = " (%s)"%(iterations) 

self.log("Doctesting %s%s%s."%(filestr, threads, iterations)) 

self.reporter = DocTestReporter(self) 

self.dispatcher = DocTestDispatcher(self) 

N = self.options.global_iterations 

for it in range(N): 

try: 

self.timer = Timer().start() 

self.dispatcher.dispatch() 

except KeyboardInterrupt: 

it = N - 1 

break 

finally: 

self.timer.stop() 

self.reporter.finalize() 

self.cleanup(it == N - 1) 

else: 

self.log("No files to doctest") 

self.reporter = DictAsObject(dict(error_status=0)) 

def cleanup(self, final=True): 

""" 

Runs cleanup activities after actually running doctests. 

In particular, saves the stats to disk and closes the logfile. 

INPUT: 

- ``final`` -- whether to close the logfile 

EXAMPLES:: 

sage: from sage.doctest.control import DocTestDefaults, DocTestController 

sage: from sage.env import SAGE_SRC 

sage: import os 

sage: dirname = os.path.join(SAGE_SRC, 'sage', 'rings', 'infinity.py') 

sage: DD = DocTestDefaults() 

sage: DC = DocTestController(DD, [dirname]) 

sage: DC.expand_files_into_sources() 

sage: DC.sources.sort(key=lambda s:s.basename) 

sage: for i, source in enumerate(DC.sources): 

....: DC.stats[source.basename] = {'walltime': 0.1*(i+1)} 

....: 

sage: DC.run() 

Running doctests with ID ... 

Doctesting 1 file. 

sage -t .../rings/infinity.py 

[... tests, ... s] 

---------------------------------------------------------------------- 

All tests passed! 

---------------------------------------------------------------------- 

Total time for all tests: ... seconds 

cpu time: ... seconds 

cumulative wall time: ... seconds 

0 

sage: DC.cleanup() 

""" 

self.stats.update(self.reporter.stats) 

self.save_stats(self.options.stats_path) 

# Close the logfile 

if final and self.logfile is not None: