Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_zero_sequence_length(self):
# Mangle a sample data file to force a zero sequence length.
ts = msprime.simulate(10, mutation_rate=2, random_seed=5)
with tempfile.TemporaryDirectory(prefix="tsinf_format_test") as tempdir:
filename = os.path.join(tempdir, "samples.tmp")
with tsinfer.SampleData(path=filename) as sample_data:
for var in ts.variants():
sample_data.add_site(var.site.position, var.genotypes)
store = zarr.LMDBStore(filename, subdir=False)
data = zarr.open(store=store, mode="w+")
data.attrs["sequence_length"] = 0
store.close()
sample_data = tsinfer.load(filename)
self.assertEqual(sample_data.sequence_length, 0)
self.assertRaises(ValueError, tsinfer.generate_ancestors, sample_data)
def _open_lmbd_readonly(self):
# We set the mapsize here because LMBD will map 1TB of virtual memory if
# we don't, making it hard to figure out how much memory we're actually
# using.
map_size = None
try:
map_size = os.path.getsize(self.path)
except OSError as e:
raise exceptions.FileFormatError(str(e)) from e
try:
store = zarr.LMDBStore(
self.path, map_size=map_size, readonly=True, subdir=False, lock=False
)
except lmdb.InvalidError as e:
raise exceptions.FileFormatError(
"Unknown file format:{}".format(str(e))
) from e
except lmdb.Error as e:
raise exceptions.FileFormatError(str(e)) from e
return store
def _new_lmdb_store(self, map_size=None):
if os.path.exists(self.path):
os.unlink(self.path)
# The existence of a lock-file can confuse things, so delete it.
remove_lmdb_lockfile(self.path)
if map_size is None:
map_size = DEFAULT_MAX_FILE_SIZE
else:
map_size = int(map_size)
if map_size <= 0:
raise ValueError("max_file_size must be > 0")
return zarr.LMDBStore(self.path, subdir=False, map_size=map_size)