You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
# TODO: make sure all dataop debugging is in place somehow
# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt"""Sqlite coverage data."""# TODO: factor out dataop debugging to a wrapper class?# TODO: make sure all dataop debugging is in place somehowimportcollectionsimportdatetimeimportglobimportitertoolsimportosimportreimportsqlite3importsysimportzlibfromcoverageimportenvfromcoverage.backwardimportget_thread_id, iitems, to_bytes, to_stringfromcoverage.debugimportNoDebugging, SimpleReprMixin, clipped_reprfromcoverage.filesimportPathAliasesfromcoverage.miscimportCoverageException, contract, file_be_gone, filename_suffix, isolate_modulefromcoverage.numbitsimportnumbits_to_nums, numbits_union, nums_to_numbitsfromcoverage.versionimport__version__os=isolate_module(os)
# If you change the schema, increment the SCHEMA_VERSION, and update the# docs in docs/dbschema.rst also.SCHEMA_VERSION=7# Schema versions:# 1: Released in 5.0a2# 2: Added contexts in 5.0a3.# 3: Replaced line table with line_map table.# 4: Changed line_map.bitmap to line_map.numbits.# 5: Added foreign key declarations.# 6: Key-value in meta.# 7: line_map -> line_bitsSCHEMA="""\CREATE TABLE coverage_schema ( -- One row, to record the version of the schema in this db. version integer);CREATE TABLE meta ( -- Key-value pairs, to record metadata about the data key text, value text, unique (key) -- Keys: -- 'has_arcs' boolean -- Is this data recording branches? -- 'sys_argv' text -- The coverage command line that recorded the data. -- 'version' text -- The version of coverage.py that made the file. -- 'when' text -- Datetime when the file was created.);CREATE TABLE file ( -- A row per file measured. id integer primary key, path text, unique (path));CREATE TABLE context ( -- A row per context measured. id integer primary key, context text, unique (context));CREATE TABLE line_bits ( -- If recording lines, a row per context per file executed. -- All of the line numbers for that file/context are in one numbits. file_id integer, -- foreign key to `file`. context_id integer, -- foreign key to `context`. numbits blob, -- see the numbits functions in coverage.numbits foreign key (file_id) references file (id), foreign key (context_id) references context (id), unique (file_id, context_id));CREATE TABLE arc ( -- If recording branches, a row per context per from/to line transition executed. file_id integer, -- foreign key to `file`. context_id integer, -- foreign key to `context`. fromno integer, -- line number jumped from. tono integer, -- line number jumped to. foreign key (file_id) references file (id), foreign key (context_id) references context (id), unique (file_id, context_id, fromno, tono));CREATE TABLE tracer ( -- A row per file indicating the tracer used for that file. file_id integer primary key, tracer text, foreign key (file_id) references file (id));"""classCoverageData(SimpleReprMixin):
"""Manages collected coverage data, including file storage. This class is the public supported API to the data that coverage.py collects during program execution. It includes information about what code was executed. It does not include information from the analysis phase, to determine what lines could have been executed, or what lines were not executed. .. note:: The data file is currently a SQLite database file, with a :ref:`documented schema <dbschema>`. The schema is subject to change though, so be careful about querying it directly. Use this API if you can to isolate yourself from changes. There are a number of kinds of data that can be collected: * **lines**: the line numbers of source lines that were executed. These are always available. * **arcs**: pairs of source and destination line numbers for transitions between source lines. These are only available if branch coverage was used. * **file tracer names**: the module names of the file tracer plugins that handled each file in the data. Lines, arcs, and file tracer names are stored for each source file. File names in this API are case-sensitive, even on platforms with case-insensitive file systems. A data file either stores lines, or arcs, but not both. A data file is associated with the data when the :class:`CoverageData` is created, using the parameters `basename`, `suffix`, and `no_disk`. The base name can be queried with :meth:`base_filename`, and the actual file name being used is available from :meth:`data_filename`. To read an existing coverage.py data file, use :meth:`read`. You can then access the line, arc, or file tracer data with :meth:`lines`, :meth:`arcs`, or :meth:`file_tracer`. The :meth:`has_arcs` method indicates whether arc data is available. You can get a set of the files in the data with :meth:`measured_files`. As with most Python containers, you can determine if there is any data at all by using this object as a boolean value. The contexts for each line in a file can be read with :meth:`contexts_by_lineno`. To limit querying to certain contexts, use :meth:`set_query_context` or :meth:`set_query_contexts`. These will narrow the focus of subsequent :meth:`lines`, :meth:`arcs`, and :meth:`contexts_by_lineno` calls. The set of all measured context names can be retrieved with :meth:`measured_contexts`. Most data files will be created by coverage.py itself, but you can use methods here to create data files if you like. The :meth:`add_lines`, :meth:`add_arcs`, and :meth:`add_file_tracers` methods add data, in ways that are convenient for coverage.py. To record data for contexts, use :meth:`set_context` to set a context to be used for subsequent :meth:`add_lines` and :meth:`add_arcs` calls. To add a source file without any measured data, use :meth:`touch_file`, or :meth:`touch_files` for a list of such files. Write the data to its file with :meth:`write`. You can clear the data in memory with :meth:`erase`. Two data collections can be combined by using :meth:`update` on one :class:`CoverageData`, passing it the other. Data in a :class:`CoverageData` can be serialized and deserialized with :meth:`dumps` and :meth:`loads`. """def__init__(self, basename=None, suffix=None, no_disk=False, warn=None, debug=None):
"""Create a :class:`CoverageData` object to hold coverage-measured data. Arguments: basename (str): the base name of the data file, defaulting to ".coverage". suffix (str or bool): has the same meaning as the `data_suffix` argument to :class:`coverage.Coverage`. no_disk (bool): if True, keep all data in memory, and don't write any disk file. warn: a warning callback function, accepting a warning message argument. debug: a `DebugControl` object (optional) """self._no_disk=no_diskself._basename=os.path.abspath(basenameor".coverage")
self._suffix=suffixself._warn=warnself._debug=debugorNoDebugging()
self._choose_filename()
self._file_map= {}
# Maps thread ids to SqliteDb objects.self._dbs= {}
self._pid=os.getpid()
# Are we in sync with the data file?self._have_used=Falseself._has_lines=Falseself._has_arcs=Falseself._current_context=Noneself._current_context_id=Noneself._query_context_ids=Nonedef_choose_filename(self):
"""Set self._filename based on inited attributes."""ifself._no_disk:
self._filename=":memory:"else:
self._filename=self._basenamesuffix=filename_suffix(self._suffix)
ifsuffix:
self._filename+="."+suffixdef_reset(self):
"""Reset our attributes."""ifself._dbs:
fordbinself._dbs.values():
db.close()
self._dbs= {}
self._file_map= {}
self._have_used=Falseself._current_context_id=Nonedef_create_db(self):
"""Create a db file that doesn't exist yet. Initializes the schema and certain metadata. """ifself._debug.should('dataio'):
self._debug.write("Creating data file {!r}".format(self._filename))
self._dbs[get_thread_id()] =db=SqliteDb(self._filename, self._debug)
withdb:
db.executescript(SCHEMA)
db.execute("insert into coverage_schema (version) values (?)", (SCHEMA_VERSION,))
db.executemany(
"insert into meta (key, value) values (?, ?)",
[
('sys_argv', str(getattr(sys, 'argv', None))),
('version', __version__),
('when', datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')),
]
)
def_open_db(self):
"""Open an existing db file, and read its metadata."""ifself._debug.should('dataio'):
self._debug.write("Opening data file {!r}".format(self._filename))
self._dbs[get_thread_id()] =SqliteDb(self._filename, self._debug)
self._read_db()
def_read_db(self):
"""Read the metadata from a database so that we are ready to use it."""withself._dbs[get_thread_id()] asdb:
try:
schema_version, =db.execute_one("select version from coverage_schema")
exceptExceptionasexc:
raiseCoverageException(
"Data file {!r} doesn't seem to be a coverage data file: {}".format(
self._filename, exc
)
)
else:
ifschema_version!=SCHEMA_VERSION:
raiseCoverageException(
"Couldn't use data file {!r}: wrong schema: {} instead of {}".format(
self._filename, schema_version, SCHEMA_VERSION
)
)
forrowindb.execute("select value from meta where key = 'has_arcs'"):
self._has_arcs=bool(int(row[0]))
self._has_lines=notself._has_arcsforpath, file_idindb.execute("select path, id from file"):
self._file_map[path] =file_iddef_connect(self):
"""Get the SqliteDb object to use."""ifget_thread_id() notinself._dbs:
ifos.path.exists(self._filename):
self._open_db()
else:
self._create_db()
returnself._dbs[get_thread_id()]
def__nonzero__(self):
if (get_thread_id() notinself._dbsandnotos.path.exists(self._filename)):
returnFalsetry:
withself._connect() ascon:
rows=con.execute("select * from file limit 1")
returnbool(list(rows))
exceptCoverageException:
returnFalse__bool__=__nonzero__@contract(returns='bytes')defdumps(self):
"""Serialize the current data to a byte string. The format of the serialized data is not documented. It is only suitable for use with :meth:`loads` in the same version of coverage.py. Returns: A byte string of serialized data. .. versionadded:: 5.0 """ifself._debug.should('dataio'):
self._debug.write("Dumping data from data file {!r}".format(self._filename))
withself._connect() ascon:
returnb'z'+zlib.compress(to_bytes(con.dump()))
@contract(data='bytes')defloads(self, data):
"""Deserialize data from :meth:`dumps` Use with a newly-created empty :class:`CoverageData` object. It's undefined what happens if the object already has data in it. Arguments: data: A byte string of serialized data produced by :meth:`dumps`. .. versionadded:: 5.0 """ifself._debug.should('dataio'):
self._debug.write("Loading data into data file {!r}".format(self._filename))
ifdata[:1] !=b'z':
raiseCoverageException(
"Unrecognized serialization: {!r} (head of {} bytes)".format(data[:40], len(data))
)
script=to_string(zlib.decompress(data[1:]))
self._dbs[get_thread_id()] =db=SqliteDb(self._filename, self._debug)
withdb:
db.executescript(script)
self._read_db()
self._have_used=Truedef_file_id(self, filename, add=False):
"""Get the file id for `filename`. If filename is not in the database yet, add it if `add` is True. If `add` is not True, return None. """iffilenamenotinself._file_map:
ifadd:
withself._connect() ascon:
cur=con.execute("insert or replace into file (path) values (?)", (filename,))
self._file_map[filename] =cur.lastrowidreturnself._file_map.get(filename)
def_context_id(self, context):
"""Get the id for a context."""assertcontextisnotNoneself._start_using()
withself._connect() ascon:
row=con.execute_one("select id from context where context = ?", (context,))
ifrowisnotNone:
returnrow[0]
else:
returnNonedefset_context(self, context):
"""Set the current context for future :meth:`add_lines` etc. `context` is a str, the name of the context to use for the next data additions. The context persists until the next :meth:`set_context`. .. versionadded:: 5.0 """ifself._debug.should('dataop'):
self._debug.write("Setting context: %r"% (context,))
self._current_context=contextself._current_context_id=Nonedef_set_context_id(self):
"""Use the _current_context to set _current_context_id."""context=self._current_contextor""context_id=self._context_id(context)
ifcontext_idisnotNone:
self._current_context_id=context_idelse:
withself._connect() ascon:
cur=con.execute("insert into context (context) values (?)", (context,))
self._current_context_id=cur.lastrowiddefbase_filename(self):
"""The base filename for storing data. .. versionadded:: 5.0 """returnself._basenamedefdata_filename(self):
"""Where is the data stored? .. versionadded:: 5.0 """returnself._filenamedefadd_lines(self, line_data):
"""Add measured line data. `line_data` is a dictionary mapping file names to dictionaries:: { filename: { lineno: None, ... }, ...} """ifself._debug.should('dataop'):
self._debug.write("Adding lines: %d files, %d lines total"% (
len(line_data), sum(len(lines) forlinesinline_data.values())
))
self._start_using()
self._choose_lines_or_arcs(lines=True)
ifnotline_data:
returnwithself._connect() ascon:
self._set_context_id()
forfilename, linenosiniitems(line_data):
linemap=nums_to_numbits(linenos)
file_id=self._file_id(filename, add=True)
query="select numbits from line_bits where file_id = ? and context_id = ?"existing=list(con.execute(query, (file_id, self._current_context_id)))
ifexisting:
linemap=numbits_union(linemap, existing[0][0])
con.execute(
"insert or replace into line_bits "" (file_id, context_id, numbits) values (?, ?, ?)",
(file_id, self._current_context_id, linemap),
)
defadd_arcs(self, arc_data):
"""Add measured arc data. `arc_data` is a dictionary mapping file names to dictionaries:: { filename: { (l1,l2): None, ... }, ...} """ifself._debug.should('dataop'):
self._debug.write("Adding arcs: %d files, %d arcs total"% (
len(arc_data), sum(len(arcs) forarcsinarc_data.values())
))
self._start_using()
self._choose_lines_or_arcs(arcs=True)
ifnotarc_data:
returnwithself._connect() ascon:
self._set_context_id()
forfilename, arcsiniitems(arc_data):
file_id=self._file_id(filename, add=True)
data= [(file_id, self._current_context_id, fromno, tono) forfromno, tonoinarcs]
con.executemany(
"insert or ignore into arc ""(file_id, context_id, fromno, tono) values (?, ?, ?, ?)",
data,
)
def_choose_lines_or_arcs(self, lines=False, arcs=False):
"""Force the data file to choose between lines and arcs."""assertlinesorarcsassertnot (linesandarcs)
iflinesandself._has_arcs:
raiseCoverageException("Can't add line measurements to existing branch data")
ifarcsandself._has_lines:
raiseCoverageException("Can't add branch measurements to existing line data")
ifnotself._has_arcsandnotself._has_lines:
self._has_lines=linesself._has_arcs=arcswithself._connect() ascon:
con.execute(
"insert into meta (key, value) values (?, ?)",
('has_arcs', str(int(arcs)))
)
defadd_file_tracers(self, file_tracers):
"""Add per-file plugin information. `file_tracers` is { filename: plugin_name, ... } """ifself._debug.should('dataop'):
self._debug.write("Adding file tracers: %d files"% (len(file_tracers),))
ifnotfile_tracers:
returnself._start_using()
withself._connect() ascon:
forfilename, plugin_nameiniitems(file_tracers):
file_id=self._file_id(filename)
iffile_idisNone:
raiseCoverageException(
"Can't add file tracer data for unmeasured file '%s'"% (filename,)
)
existing_plugin=self.file_tracer(filename)
ifexisting_plugin:
ifexisting_plugin!=plugin_name:
raiseCoverageException(
"Conflicting file tracer name for '%s': %r vs %r"% (
filename, existing_plugin, plugin_name,
)
)
elifplugin_name:
con.execute(
"insert into tracer (file_id, tracer) values (?, ?)",
(file_id, plugin_name)
)
deftouch_file(self, filename, plugin_name=""):
"""Ensure that `filename` appears in the data, empty if needed. `plugin_name` is the name of the plugin responsible for this file. It is used to associate the right filereporter, etc. """self.touch_files([filename], plugin_name)
deftouch_files(self, filenames, plugin_name=""):
"""Ensure that `filenames` appear in the data, empty if needed. `plugin_name` is the name of the plugin responsible for these files. It is used to associate the right filereporter, etc. """ifself._debug.should('dataop'):
self._debug.write("Touching %r"% (filenames,))
self._start_using()
withself._connect(): # Use this to get one transaction.ifnotself._has_arcsandnotself._has_lines:
raiseCoverageException("Can't touch files in an empty CoverageData")
forfilenameinfilenames:
self._file_id(filename, add=True)
ifplugin_name:
# Set the tracer for this fileself.add_file_tracers({filename: plugin_name})
defupdate(self, other_data, aliases=None):
"""Update this data with data from several other :class:`CoverageData` instances. If `aliases` is provided, it's a `PathAliases` object that is used to re-map paths to match the local machine's. """ifself._debug.should('dataop'):
self._debug.write("Updating with data from %r"% (
getattr(other_data, '_filename', '???'),
))
ifself._has_linesandother_data._has_arcs:
raiseCoverageException("Can't combine arc data with line data")
ifself._has_arcsandother_data._has_lines:
raiseCoverageException("Can't combine line data with arc data")
aliases=aliasesorPathAliases()
# Force the database we're writing to to exist before we start nesting# contexts.self._start_using()
# Collector for all arcs, lines and tracersother_data.read()
withother_data._connect() asconn:
# Get files data.cur=conn.execute('select path from file')
files= {path: aliases.map(path) for (path,) incur}
cur.close()
# Get contexts data.cur=conn.execute('select context from context')
contexts= [contextfor (context,) incur]
cur.close()
# Get arc data.cur=conn.execute(
'select file.path, context.context, arc.fromno, arc.tono ''from arc ''inner join file on file.id = arc.file_id ''inner join context on context.id = arc.context_id'
)
arcs= [(files[path], context, fromno, tono) for (path, context, fromno, tono) incur]
cur.close()
# Get line data.cur=conn.execute(
'select file.path, context.context, line_bits.numbits ''from line_bits ''inner join file on file.id = line_bits.file_id ''inner join context on context.id = line_bits.context_id'
)
lines= {
(files[path], context): numbitsfor (path, context, numbits) incur
}
cur.close()
# Get tracer data.cur=conn.execute(
'select file.path, tracer ''from tracer ''inner join file on file.id = tracer.file_id'
)
tracers= {files[path]: tracerfor (path, tracer) incur}
cur.close()
withself._connect() asconn:
conn.con.isolation_level='IMMEDIATE'# Get all tracers in the DB. Files not in the tracers are assumed# to have an empty string tracer. Since Sqlite does not support# full outer joins, we have to make two queries to fill the# dictionary.this_tracers= {path: ''forpath, inconn.execute('select path from file')}
this_tracers.update({
aliases.map(path): tracerforpath, tracerinconn.execute(
'select file.path, tracer from tracer ''inner join file on file.id = tracer.file_id'
)
})
# Create all file and context rows in the DB.conn.executemany(
'insert or ignore into file (path) values (?)',
((file,) forfileinfiles.values())
)
file_ids= {
path: idforid, pathinconn.execute('select id, path from file')
}
conn.executemany(
'insert or ignore into context (context) values (?)',
((context,) forcontextincontexts)
)
context_ids= {
context: idforid, contextinconn.execute('select id, context from context')
}
# Prepare tracers and fail, if a conflict is found.# tracer_paths is used to ensure consistency over the tracer data# and tracer_map tracks the tracers to be inserted.tracer_map= {}
forpathinfiles.values():
this_tracer=this_tracers.get(path)
other_tracer=tracers.get(path, '')
# If there is no tracer, there is always the None tracer.ifthis_tracerisnotNoneandthis_tracer!=other_tracer:
raiseCoverageException(
"Conflicting file tracer name for '%s': %r vs %r"% (
path, this_tracer, other_tracer
)
)
tracer_map[path] =other_tracer# Prepare arc and line rows to be inserted by converting the file# and context strings with integer ids. Then use the efficient# `executemany()` to insert all rows at once.arc_rows= (
(file_ids[file], context_ids[context], fromno, tono)
forfile, context, fromno, tonoinarcs
)
# Get line data.cur=conn.execute(
'select file.path, context.context, line_bits.numbits ''from line_bits ''inner join file on file.id = line_bits.file_id ''inner join context on context.id = line_bits.context_id'
)
forpath, context, numbitsincur:
key= (aliases.map(path), context)
ifkeyinlines:
numbits=numbits_union(lines[key], numbits)
lines[key] =numbitscur.close()
ifarcs:
self._choose_lines_or_arcs(arcs=True)
# Write the combined data.conn.executemany(
'insert or ignore into arc ''(file_id, context_id, fromno, tono) values (?, ?, ?, ?)',
arc_rows
)
iflines:
self._choose_lines_or_arcs(lines=True)
conn.execute("delete from line_bits")
conn.executemany(
"insert into line_bits ""(file_id, context_id, numbits) values (?, ?, ?)",
[
(file_ids[file], context_ids[context], numbits)
for (file, context), numbitsinlines.items()
]
)
conn.executemany(
'insert or ignore into tracer (file_id, tracer) values (?, ?)',
((file_ids[filename], tracer) forfilename, tracerintracer_map.items())
)
# Update all internal cache data.self._reset()
self.read()
deferase(self, parallel=False):
"""Erase the data in this object. If `parallel` is true, then also deletes data files created from the basename by parallel-mode. """self._reset()
ifself._no_disk:
returnifself._debug.should('dataio'):
self._debug.write("Erasing data file {!r}".format(self._filename))
file_be_gone(self._filename)
ifparallel:
data_dir, local=os.path.split(self._filename)
localdot=local+'.*'pattern=os.path.join(os.path.abspath(data_dir), localdot)
forfilenameinglob.glob(pattern):
ifself._debug.should('dataio'):
self._debug.write("Erasing parallel data file {!r}".format(filename))
file_be_gone(filename)
defread(self):
"""Start using an existing data file."""withself._connect(): # TODO: doesn't look rightself._have_used=Truedefwrite(self):
"""Ensure the data is written to the data file."""passdef_start_using(self):
"""Call this before using the database at all."""ifself._pid!=os.getpid():
# Looks like we forked! Have to start a new data file.self._reset()
self._choose_filename()
self._pid=os.getpid()
ifnotself._have_used:
self.erase()
self._have_used=Truedefhas_arcs(self):
"""Does the database have arcs (True) or lines (False)."""returnbool(self._has_arcs)
defmeasured_files(self):
"""A set of all files that had been measured."""returnset(self._file_map)
defmeasured_contexts(self):
"""A set of all contexts that have been measured. .. versionadded:: 5.0 """self._start_using()
withself._connect() ascon:
contexts= {row[0] forrowincon.execute("select distinct(context) from context")}
returncontextsdeffile_tracer(self, filename):
"""Get the plugin name of the file tracer for a file. Returns the name of the plugin that handles this file. If the file was measured, but didn't use a plugin, then "" is returned. If the file was not measured, then None is returned. """self._start_using()
withself._connect() ascon:
file_id=self._file_id(filename)
iffile_idisNone:
returnNonerow=con.execute_one("select tracer from tracer where file_id = ?", (file_id,))
ifrowisnotNone:
returnrow[0] or""return""# File was measured, but no tracer associated.defset_query_context(self, context):
"""Set a context for subsequent querying. The next :meth:`lines`, :meth:`arcs`, or :meth:`contexts_by_lineno` calls will be limited to only one context. `context` is a string which must match a context exactly. If it does not, no exception is raised, but queries will return no data. .. versionadded:: 5.0 """self._start_using()
withself._connect() ascon:
cur=con.execute("select id from context where context = ?", (context,))
self._query_context_ids= [row[0] forrowincur.fetchall()]
defset_query_contexts(self, contexts):
"""Set a number of contexts for subsequent querying. The next :meth:`lines`, :meth:`arcs`, or :meth:`contexts_by_lineno` calls will be limited to the specified contexts. `contexts` is a list of Python regular expressions. Contexts will be matched using :func:`re.search <python:re.search>`. Data will be included in query results if they are part of any of the contexts matched. .. versionadded:: 5.0 """self._start_using()
ifcontexts:
withself._connect() ascon:
context_clause=' or '.join(['context regexp ?'] *len(contexts))
cur=con.execute("select id from context where "+context_clause, contexts)
self._query_context_ids= [row[0] forrowincur.fetchall()]
else:
self._query_context_ids=Nonedeflines(self, filename):
"""Get the list of lines executed for a file. If the file was not measured, returns None. A file might be measured, and have no lines executed, in which case an empty list is returned. If the file was executed, returns a list of integers, the line numbers executed in the file. The list is in no particular order. """self._start_using()
ifself.has_arcs():
arcs=self.arcs(filename)
ifarcsisnotNone:
all_lines=itertools.chain.from_iterable(arcs)
returnlist({lforlinall_linesifl>0})
withself._connect() ascon:
file_id=self._file_id(filename)
iffile_idisNone:
returnNoneelse:
query="select numbits from line_bits where file_id = ?"data= [file_id]
ifself._query_context_idsisnotNone:
ids_array=', '.join('?'*len(self._query_context_ids))
query+=" and context_id in ("+ids_array+")"data+=self._query_context_idsbitmaps=list(con.execute(query, data))
nums=set()
forrowinbitmaps:
nums.update(numbits_to_nums(row[0]))
returnlist(nums)
defarcs(self, filename):
"""Get the list of arcs executed for a file. If the file was not measured, returns None. A file might be measured, and have no arcs executed, in which case an empty list is returned. If the file was executed, returns a list of 2-tuples of integers. Each pair is a starting line number and an ending line number for a transition from one line to another. The list is in no particular order. Negative numbers have special meaning. If the starting line number is -N, it represents an entry to the code object that starts at line N. If the ending ling number is -N, it's an exit from the code object that starts at line N. """self._start_using()
withself._connect() ascon:
file_id=self._file_id(filename)
iffile_idisNone:
returnNoneelse:
query="select distinct fromno, tono from arc where file_id = ?"data= [file_id]
ifself._query_context_idsisnotNone:
ids_array=', '.join('?'*len(self._query_context_ids))
query+=" and context_id in ("+ids_array+")"data+=self._query_context_idsarcs=con.execute(query, data)
returnlist(arcs)
defcontexts_by_lineno(self, filename):
"""Get the contexts for each line in a file. Returns: A dict mapping line numbers to a list of context names. .. versionadded:: 5.0 """lineno_contexts_map=collections.defaultdict(list)
self._start_using()
withself._connect() ascon:
file_id=self._file_id(filename)
iffile_idisNone:
returnlineno_contexts_mapifself.has_arcs():
query= (
"select arc.fromno, arc.tono, context.context ""from arc, context ""where arc.file_id = ? and arc.context_id = context.id"
)
data= [file_id]
ifself._query_context_idsisnotNone:
ids_array=', '.join('?'*len(self._query_context_ids))
query+=" and arc.context_id in ("+ids_array+")"data+=self._query_context_idsforfromno, tono, contextincon.execute(query, data):
ifcontextnotinlineno_contexts_map[fromno]:
lineno_contexts_map[fromno].append(context)
ifcontextnotinlineno_contexts_map[tono]:
lineno_contexts_map[tono].append(context)
else:
query= (
"select l.numbits, c.context from line_bits l, context c ""where l.context_id = c.id ""and file_id = ?"
)
data= [file_id]
ifself._query_context_idsisnotNone:
ids_array=', '.join('?'*len(self._query_context_ids))
query+=" and l.context_id in ("+ids_array+")"data+=self._query_context_idsfornumbits, contextincon.execute(query, data):
forlinenoinnumbits_to_nums(numbits):
lineno_contexts_map[lineno].append(context)
returnlineno_contexts_map@classmethoddefsys_info(cls):
"""Our information for `Coverage.sys_info`. Returns a list of (key, value) pairs. """withSqliteDb(":memory:", debug=NoDebugging()) asdb:
temp_store= [row[0] forrowindb.execute("pragma temp_store")]
compile_options= [row[0] forrowindb.execute("pragma compile_options")]
return [
('sqlite3_version', sqlite3.version),
('sqlite3_sqlite_version', sqlite3.sqlite_version),
('sqlite3_temp_store', temp_store),
('sqlite3_compile_options', compile_options),
]
classSqliteDb(SimpleReprMixin):
"""A simple abstraction over a SQLite database. Use as a context manager, then you can use it like a :class:`python:sqlite3.Connection` object:: with SqliteDb(filename, debug_control) as db: db.execute("insert into schema (version) values (?)", (SCHEMA_VERSION,)) """def__init__(self, filename, debug):
self.debug=debugifdebug.should('sql') elseNoneself.filename=filenameself.nest=0self.con=Nonedef_connect(self):
"""Connect to the db and do universal initialization."""ifself.conisnotNone:
return# SQLite on Windows on py2 won't open a file if the filename argument# has non-ascii characters in it. Opening a relative file name avoids# a problem if the current directory has non-ascii.filename=self.filenameifenv.WINDOWSandenv.PY2:
try:
filename=os.path.relpath(self.filename)
exceptValueError:
# ValueError can be raised under Windows when os.getcwd() returns a# folder from a different drive than the drive of self.filename in# which case we keep the original value of self.filename unchanged,# hoping that we won't face the non-ascii directory problem.pass# It can happen that Python switches threads while the tracer writes# data. The second thread will also try to write to the data,# effectively causing a nested context. However, given the idempotent# nature of the tracer operations, sharing a connection among threads# is not a problem.ifself.debug:
self.debug.write("Connecting to {!r}".format(self.filename))
self.con=sqlite3.connect(filename, check_same_thread=False)
self.con.create_function('REGEXP', 2, _regexp)
# This pragma makes writing faster. It disables rollbacks, but we never need them.# PyPy needs the .close() calls here, or sqlite gets twisted up:# https://bitbucket.org/pypy/pypy/issues/2872/default-isolation-mode-is-different-onself.execute("pragma journal_mode=off").close()
# This pragma makes writing faster.self.execute("pragma synchronous=off").close()
defclose(self):
"""If needed, close the connection."""ifself.conisnotNoneandself.filename!=":memory:":
self.con.close()
self.con=Nonedef__enter__(self):
ifself.nest==0:
self._connect()
self.con.__enter__()
self.nest+=1returnselfdef__exit__(self, exc_type, exc_value, traceback):
self.nest-=1ifself.nest==0:
try:
self.con.__exit__(exc_type, exc_value, traceback)
self.close()
exceptExceptionasexc:
ifself.debug:
self.debug.write("EXCEPTION from __exit__: {}".format(exc))
raisedefexecute(self, sql, parameters=()):
"""Same as :meth:`python:sqlite3.Connection.execute`."""ifself.debug:
tail=" with {!r}".format(parameters) ifparameterselse""self.debug.write("Executing {!r}{}".format(sql, tail))
try:
try:
returnself.con.execute(sql, parameters)
exceptException:
# In some cases, an error might happen that isn't really an# error. Try again immediately.# https://github.com/nedbat/coveragepy/issues/1010returnself.con.execute(sql, parameters)
exceptsqlite3.Errorasexc:
msg=str(exc)
try:
# `execute` is the first thing we do with the database, so try# hard to provide useful hints if something goes wrong now.withopen(self.filename, "rb") asbad_file:
cov4_sig=b"!coverage.py: This is a private format"ifbad_file.read(len(cov4_sig)) ==cov4_sig:
msg= (
"Looks like a coverage 4.x data file. ""Are you mixing versions of coverage?"
)
exceptException:
passifself.debug:
self.debug.write("EXCEPTION from execute: {}".format(msg))
raiseCoverageException("Couldn't use data file {!r}: {}".format(self.filename, msg))
defexecute_one(self, sql, parameters=()):
"""Execute a statement and return the one row that results. This is like execute(sql, parameters).fetchone(), except it is correct in reading the entire result set. This will raise an exception if more than one row results. Returns a row, or None if there were no rows. """rows=list(self.execute(sql, parameters))
iflen(rows) ==0:
returnNoneeliflen(rows) ==1:
returnrows[0]
else:
raiseCoverageException("Sql {!r} shouldn't return {} rows".format(sql, len(rows)))
defexecutemany(self, sql, data):
"""Same as :meth:`python:sqlite3.Connection.executemany`."""ifself.debug:
data=list(data)
self.debug.write("Executing many {!r} with {} rows".format(sql, len(data)))
returnself.con.executemany(sql, data)
defexecutescript(self, script):
"""Same as :meth:`python:sqlite3.Connection.executescript`."""ifself.debug:
self.debug.write("Executing script with {} chars: {}".format(
len(script), clipped_repr(script, 100),
))
self.con.executescript(script)
defdump(self):
"""Return a multi-line string, the SQL dump of the database."""return"\n".join(self.con.iterdump())
def_regexp(text, pattern):
"""A regexp function for SQLite."""returnre.search(text, pattern) isnotNone
24a002c511e7c3e425c9a78627d5b020c610c729
The text was updated successfully, but these errors were encountered:
make sure all dataop debugging is in place somehow
docs in docs/dbschema.rst also.
1: Released in 5.0a2
2: Added contexts in 5.0a3.
3: Replaced line table with line_map table.
4: Changed line_map.bitmap to line_map.numbits.
5: Added foreign key declarations.
6: Key-value in meta.
7: line_map -> line_bits
contexts.
to have an empty string tracer. Since Sqlite does not support
full outer joins, we have to make two queries to fill the
dictionary.
tracer_paths is used to ensure consistency over the tracer data
and tracer_map tracks the tracers to be inserted.
and context strings with integer ids. Then use the efficient
executemany()
to insert all rows at once.has non-ascii characters in it. Opening a relative file name avoids
a problem if the current directory has non-ascii.
folder from a different drive than the drive of self.filename in
which case we keep the original value of self.filename unchanged,
hoping that we won't face the non-ascii directory problem.
data. The second thread will also try to write to the data,
effectively causing a nested context. However, given the idempotent
nature of the tracer operations, sharing a connection among threads
is not a problem.
PyPy needs the .close() calls here, or sqlite gets twisted up:
https://bitbucket.org/pypy/pypy/issues/2872/default-isolation-mode-is-different-on
error. Try again immediately.
nedbat/coveragepy#1010
hard to provide useful hints if something goes wrong now.
gcp-database-as-a-service-stock-markets/venv/Lib/site-packages/coverage/sqldata.py
Line 7 in 62db596
24a002c511e7c3e425c9a78627d5b020c610c729
The text was updated successfully, but these errors were encountered: