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_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
"spacy_model": load_spacy_model(SPACY_MODELS.en_core_web_sm)[0],
"jobs": 2,
},
),
(generate_report, "libratom.lib.report.Message", {}),
],
)
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)
def test_entities_with_bad_model(enron_dataset_part001):
with tempfile.TemporaryDirectory() as tmpdir:
assert 1 == subcommands.entities(
out=Path(tmpdir),
spacy_model_name="no_such_model",
jobs=2,
src=enron_dataset_part001,
progress=False,
)
assert not any(load_spacy_model(spacy_model_name="no_such_model"))
desc="Initial file scan",
unit="files",
color="green",
leave=False,
) as file_bar, db_session(Session) as session:
status = scan_files(
files, session, jobs=jobs, progress_callback=file_bar.update
)
if status == 1:
logger.warning("Aborting")
return status
# Get spaCy model
logger.info(f"Loading spaCy model: {spacy_model_name}")
spacy_model, spacy_model_version = load_spacy_model(spacy_model_name)
if not spacy_model:
return 1
# Try to see if we're using a stale model version
try:
latest_version = get_spacy_models()[spacy_model_name][0]
if parse(latest_version) > parse(spacy_model_version):
logger.info(
f"Model {spacy_model_name} {spacy_model_version} will be used, but {latest_version} is available"
)
except Exception as exc:
logger.debug(exc, exc_info=True)
# Get messages and extract entities
with db_session(Session) as session: