eric: comparison eric7/DebugClients/Python/coverage/sqldata.py

-:d856023fbeb0
+:fcca2fa618bf
 import datetime
 import functools
 import glob
 import itertools
 import os
+import random
 import re
+import socket
 import sqlite3
 import sys
 import threading
 import zlib
 from coverage.debug import NoDebugging, SimpleReprMixin, clipped_repr
-from coverage.exceptions import CoverageException
+from coverage.exceptions import CoverageException, DataError
 from coverage.files import PathAliases
-from coverage.misc import contract, file_be_gone, filename_suffix, isolate_module
+from coverage.misc import contract, file_be_gone, isolate_module
 from coverage.numbits import numbits_to_nums, numbits_union, nums_to_numbits
 from coverage.version import __version__
 os = isolate_module(os)
 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".
+".coverage". This can be a path to a file in another directory.
 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
 """Read the metadata from a database so that we are ready to use it."""
 with self._dbs[threading.get_ident()] as db:
 try:
 schema_version, = db.execute_one("select version from coverage_schema")
 except Exception as exc:
-raise CoverageException(
+raise DataError(
 "Data file {!r} doesn't seem to be a coverage data file: {}".format(
 self._filename, exc
 )
 ) from exc
 else:
 if schema_version != SCHEMA_VERSION:
-raise CoverageException(
+raise DataError(
 "Couldn't use data file {!r}: wrong schema: {} instead of {}".format(
 self._filename, schema_version, SCHEMA_VERSION
 )
 )
 self._open_db()
 else:
 self._create_db()
 return self._dbs[threading.get_ident()]
-def __nonzero__(self):
+def __bool__(self):
 if (threading.get_ident() not in self._dbs and not os.path.exists(self._filename)):
 return False
 try:
 with self._connect() as con:
 rows = con.execute("select * from file limit 1")
 return bool(list(rows))
 except CoverageException:
 return False
-__bool__ = __nonzero__
 @contract(returns="bytes")
 def dumps(self):
 """Serialize the current data to a byte string.
 The format of the serialized data is not documented. It is only
 """
 if self._debug.should("dataio"):
 self._debug.write(f"Loading data into data file {self._filename!r}")
 if data[:1] != b"z":
-raise CoverageException(
+raise DataError(
 f"Unrecognized serialization: {data[:40]!r} (head of {len(data)} bytes)"
 )
 script = zlib.decompress(data[1:]).decode("utf-8")
 self._dbs[threading.get_ident()] = db = SqliteDb(self._filename, self._debug)
 with db:
 db.executescript(script)
 self._read_db()
 def _choose_lines_or_arcs(self, lines=False, arcs=False):
 """Force the data file to choose between lines and arcs."""
 assert lines or arcs
 assert not (lines and arcs)
 if lines and self._has_arcs:
-raise CoverageException("Can't add line measurements to existing branch data")
+raise DataError("Can't add line measurements to existing branch data")
 if arcs and self._has_lines:
-raise CoverageException("Can't add branch measurements to existing line data")
+raise DataError("Can't add branch measurements to existing line data")
 if not self._has_arcs and not self._has_lines:
 self._has_lines = lines
 self._has_arcs = arcs
 with self._connect() as con:
 con.execute(
 self._start_using()
 with self._connect() as con:
 for filename, plugin_name in file_tracers.items():
 file_id = self._file_id(filename)
 if file_id is None:
-raise CoverageException(
+raise DataError(
 f"Can't add file tracer data for unmeasured file '{filename}'"
 )
 existing_plugin = self.file_tracer(filename)
 if existing_plugin:
 if existing_plugin != plugin_name:
-raise CoverageException(
+raise DataError(
 "Conflicting file tracer name for '{}': {!r} vs {!r}".format(
 filename, existing_plugin, plugin_name,
 )
 )
 elif plugin_name:
 if self._debug.should("dataop"):
 self._debug.write(f"Touching {filenames!r}")
 self._start_using()
 with self._connect(): # Use this to get one transaction.
 if not self._has_arcs and not self._has_lines:
-raise CoverageException("Can't touch files in an empty CoverageData")
+raise DataError("Can't touch files in an empty CoverageData")
 for filename in filenames:
 self._file_id(filename, add=True)
 if plugin_name:
 # Set the tracer for this file
 if self._debug.should("dataop"):
 self._debug.write("Updating with data from {!r}".format(
 getattr(other_data, "_filename", "???"),
 ))
 if self._has_lines and other_data._has_arcs:
-raise CoverageException("Can't combine arc data with line data")
+raise DataError("Can't combine arc data with line data")
 if self._has_arcs and other_data._has_lines:
-raise CoverageException("Can't combine line data with arc data")
+raise DataError("Can't combine line data with arc data")
 aliases = aliases or PathAliases()
 # Force the database we're writing to to exist before we start nesting
 # contexts.
 for path in files.values():
 this_tracer = this_tracers.get(path)
 other_tracer = tracers.get(path, "")
 # If there is no tracer, there is always the None tracer.
 if this_tracer is not None and this_tracer != other_tracer:
-raise CoverageException(
+raise DataError(
 "Conflicting file tracer name for '{}': {!r} vs {!r}".format(
 path, this_tracer, other_tracer
 )
 )
 tracer_map[path] = other_tracer
 ("sqlite3_temp_store", temp_store),
 ("sqlite3_compile_options", copts),
 ]
+def filename_suffix(suffix):
+"""Compute a filename suffix for a data file.
+If `suffix` is a string or None, simply return it. If `suffix` is True,
+then build a suffix incorporating the hostname, process id, and a random
+number.
+Returns a string or None.
+"""
+if suffix is True:
+# If data_suffix was a simple true value, then make a suffix with
+# plenty of distinguishing information.  We do this here in
+# `save()` at the last minute so that the pid will be correct even
+# if the process forks.
+dice = random.Random(os.urandom(8)).randint(0, 999999)
+suffix = "%s.%s.%06d" % (socket.gethostname(), os.getpid(), dice)
+return suffix
 class SqliteDb(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::
 if self.debug:
 self.debug.write(f"Connecting to {self.filename!r}")
 try:
 self.con = sqlite3.connect(self.filename, check_same_thread=False)
 except sqlite3.Error as exc:
-raise CoverageException(f"Couldn't use data file {self.filename!r}: {exc}") from exc
+raise DataError(f"Couldn't use data file {self.filename!r}: {exc}") from exc
 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:
 self.con.__exit__(exc_type, exc_value, traceback)
 self.close()
 except Exception as exc:
 if self.debug:
 self.debug.write(f"EXCEPTION from __exit__: {exc}")
-raise CoverageException(f"Couldn't end data file {self.filename!r}: {exc}") from exc
+raise DataError(f"Couldn't end data file {self.filename!r}: {exc}") from exc
 def execute(self, sql, parameters=()):
 """Same as :meth:`python:sqlite3.Connection.execute`."""
 if self.debug:
 tail = f" with {parameters!r}" if parameters else ""
 )
 except Exception:   # pragma: cant happen
 pass
 if self.debug:
 self.debug.write(f"EXCEPTION from execute: {msg}")
-raise CoverageException(f"Couldn't use data file {self.filename!r}: {msg}") from exc
+raise DataError(f"Couldn't use data file {self.filename!r}: {msg}") from exc
 def execute_one(self, sql, parameters=()):
 """Execute a statement and return the one row that results.
 This is like execute(sql, parameters).fetchone(), except it is

Mercurial Repositories > eric / file comparison

comparison: eric7/DebugClients/Python/coverage/sqldata.py

eric7/DebugClients/Python/coverage/sqldata.py