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_scan_files_with_interrupt(directory_of_mbox_files):
tmp_filename = "test.sqlite3"
with TemporaryDirectory() as tmpdir:
destination = Path(tmpdir) / tmp_filename
Session = db_init(destination)
with db_session(Session) as session, patch(
"libratom.lib.report.FileReport",
new=MagicMock(side_effect=KeyboardInterrupt),
):
assert (
scan_files(
files=get_set_of_files(directory_of_mbox_files),
session=session,
jobs=2,
)
== 1
)
def test_extract_entities_from_mbox_files(directory_of_mbox_files):
tmp_filename = "test.sqlite3"
with TemporaryDirectory() as tmpdir:
destination = Path(tmpdir) / tmp_filename
Session = db_init(destination)
with db_session(Session) as session:
status = extract_entities(
files=get_set_of_files(directory_of_mbox_files),
session=session,
spacy_model=load_spacy_model(SPACY_MODELS.en_core_web_sm)[0],
jobs=2,
)
assert status == 0
def test_run_function_with_interrupt(
directory_of_mbox_files, function, patched, kwargs
):
tmp_filename = "test.sqlite3"
with TemporaryDirectory() as tmpdir:
destination = Path(tmpdir) / tmp_filename
Session = db_init(destination)
with db_session(Session) as session, patch(
patched, new=MagicMock(side_effect=KeyboardInterrupt),
):
status = function(
files=get_set_of_files(directory_of_mbox_files),
session=session,
**kwargs,
)
assert status == 1
# Resolve output file based on src parameter
if out.is_dir():
out = out / OUTPUT_FILENAME_TEMPLATE.format(
src.name,
"entities",
datetime.now()
.isoformat(timespec="seconds")
.translate(str.maketrans({"-": "", ":": ""})),
)
# Make DB file's parents if needed
out.parent.mkdir(parents=True, exist_ok=True)
# DB setup
Session = db_init(out)
# Get set of PST files from the source
files = get_set_of_files(src)
if not files:
logger.info(f"No PST file found in {src}")
# Compute and store file information
with progress_bar_context(
total=len(files),
desc="Initial file scan",
unit="files",
color="green",
leave=False,
) as file_bar, db_session(Session) as session:
status = scan_files(
# Resolve output file based on src parameter
if out.is_dir():
out = out / OUTPUT_FILENAME_TEMPLATE.format(
src.name,
"report",
datetime.now()
.isoformat(timespec="seconds")
.translate(str.maketrans({"-": "", ":": ""})),
)
# Make DB file's parents if needed
out.parent.mkdir(parents=True, exist_ok=True)
# DB setup
Session = db_init(out)
# Get set of PST files from the source
files = get_set_of_files(src)
if not files:
logger.info(f"No PST file found in {src}")
# Compute and store file information
with progress_bar_context(
total=len(files),
desc="Initial file scan",
unit="files",
color="green",
leave=False,
) as file_bar, db_session(Session) as session:
status = scan_files(