wehub-resource-sync
c56bef871b
CodeQL / Analyze (python) (push) Has been cancelled
Update Platform Components Table / update (push) Has been cancelled
Docker image release / Build base image (push) Has been cancelled
Sync docs with Docusaurus / sync (push) Has been cancelled
Tests / Check if changed (push) Has been cancelled
Tests / format (push) Has been cancelled
Tests / check-imports (push) Has been cancelled
Tests / Unit / macos-latest (push) Has been cancelled
Tests / Unit / ubuntu-latest (push) Has been cancelled
Tests / Unit / windows-latest (push) Has been cancelled
Tests / mypy (push) Has been cancelled
Tests / Integration / ubuntu-latest (push) Has been cancelled
Tests / Integration / macos-latest (push) Has been cancelled
Tests / Integration / windows-latest (push) Has been cancelled
Tests / notify-slack-on-failure (push) Has been cancelled
Tests / Mark tests as completed (push) Has been cancelled
514 lines
22 KiB
Python
514 lines
22 KiB
Python
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
|
|
#
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
import io
|
|
import mimetypes
|
|
import sys
|
|
from pathlib import PosixPath
|
|
from unittest.mock import mock_open, patch
|
|
|
|
import pytest
|
|
|
|
from haystack import Pipeline, component
|
|
from haystack.components.converters import PyPDFToDocument, TextFileToDocument
|
|
from haystack.components.routers.file_type_router import FileTypeRouter
|
|
from haystack.dataclasses import ByteStream
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform in ["win32", "cygwin"],
|
|
reason="Can't run on Windows Github CI, need access to registry to get mime types",
|
|
)
|
|
class TestFileTypeRouter:
|
|
def test_init(self):
|
|
"""
|
|
Test that the component initializes correctly.
|
|
"""
|
|
router = FileTypeRouter(mime_types=["text/plain", "audio/x-wav", "image/jpeg"])
|
|
assert router.mime_types == ["text/plain", "audio/x-wav", "image/jpeg"]
|
|
assert router._additional_mimetypes is None
|
|
|
|
router = FileTypeRouter(
|
|
mime_types=["text/plain"],
|
|
additional_mimetypes={"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"},
|
|
)
|
|
assert router.mime_types == ["text/plain"]
|
|
assert router._additional_mimetypes == {
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"
|
|
}
|
|
|
|
def test_init_fail_wo_mime_types(self):
|
|
"""
|
|
Test that the component raises an error if no mime types are provided.
|
|
"""
|
|
with pytest.raises(ValueError):
|
|
FileTypeRouter(mime_types=[])
|
|
|
|
def test_to_dict(self):
|
|
router = FileTypeRouter(
|
|
mime_types=["text/plain", "audio/x-wav", "image/jpeg"],
|
|
additional_mimetypes={"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"},
|
|
)
|
|
expected_dict = {
|
|
"type": "haystack.components.routers.file_type_router.FileTypeRouter",
|
|
"init_parameters": {
|
|
"mime_types": ["text/plain", "audio/x-wav", "image/jpeg"],
|
|
"additional_mimetypes": {
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"
|
|
},
|
|
"raise_on_failure": False,
|
|
},
|
|
}
|
|
assert router.to_dict() == expected_dict
|
|
|
|
def test_from_dict(self):
|
|
router_dict = {
|
|
"type": "haystack.components.routers.file_type_router.FileTypeRouter",
|
|
"init_parameters": {
|
|
"mime_types": ["text/plain", "audio/x-wav", "image/jpeg"],
|
|
"additional_mimetypes": {
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"
|
|
},
|
|
},
|
|
}
|
|
loaded_router = FileTypeRouter.from_dict(router_dict)
|
|
|
|
expected_router = FileTypeRouter(
|
|
mime_types=["text/plain", "audio/x-wav", "image/jpeg"],
|
|
additional_mimetypes={"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"},
|
|
)
|
|
|
|
assert loaded_router.mime_types == expected_router.mime_types
|
|
assert loaded_router._additional_mimetypes == expected_router._additional_mimetypes
|
|
|
|
def test_run(self, test_files_path):
|
|
"""
|
|
Test if the component runs correctly in the simplest happy path.
|
|
"""
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "txt" / "doc_2.txt",
|
|
test_files_path / "audio" / "the context for this answer is here.wav",
|
|
test_files_path / "images" / "apple.jpg",
|
|
test_files_path / "msg" / "sample.msg",
|
|
]
|
|
|
|
router = FileTypeRouter(mime_types=[r"text/plain", r"audio/x-wav", r"image/jpeg", "application/vnd.ms-outlook"])
|
|
output = router.run(sources=file_paths)
|
|
assert output
|
|
assert len(output[r"text/plain"]) == 2
|
|
assert len(output[r"audio/x-wav"]) == 1
|
|
assert len(output[r"image/jpeg"]) == 1
|
|
assert len(output["application/vnd.ms-outlook"]) == 1
|
|
assert not output.get("unclassified")
|
|
|
|
def test_run_with_single_meta(self, test_files_path):
|
|
"""
|
|
Test if the component runs correctly when a single metadata dictionary is provided.
|
|
"""
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "txt" / "doc_2.txt",
|
|
test_files_path / "audio" / "the context for this answer is here.wav",
|
|
]
|
|
|
|
meta = {"meta_field": "meta_value"}
|
|
|
|
router = FileTypeRouter(mime_types=[r"text/plain", r"audio/x-wav"])
|
|
output = router.run(sources=file_paths, meta=meta)
|
|
assert output
|
|
|
|
assert len(output[r"text/plain"]) == 2
|
|
assert len(output[r"audio/x-wav"]) == 1
|
|
assert not output.get("unclassified")
|
|
|
|
for elements in output.values():
|
|
for el in elements:
|
|
assert isinstance(el, ByteStream)
|
|
assert el.meta["meta_field"] == "meta_value"
|
|
|
|
def test_run_with_meta_list(self, test_files_path):
|
|
"""
|
|
Test if the component runs correctly when a list of metadata dictionaries is provided.
|
|
"""
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "images" / "apple.jpg",
|
|
test_files_path / "audio" / "the context for this answer is here.wav",
|
|
]
|
|
|
|
meta = [{"key1": "value1"}, {"key2": "value2"}, {"key3": "value3"}]
|
|
|
|
router = FileTypeRouter(mime_types=[r"text/plain", r"audio/x-wav", r"image/jpeg"])
|
|
output = router.run(sources=file_paths, meta=meta)
|
|
assert output
|
|
|
|
assert len(output[r"text/plain"]) == 1
|
|
assert len(output[r"audio/x-wav"]) == 1
|
|
assert len(output[r"image/jpeg"]) == 1
|
|
assert not output.get("unclassified")
|
|
|
|
for i, elements in enumerate(output.values()):
|
|
for el in elements:
|
|
assert isinstance(el, ByteStream)
|
|
expected_meta_key, expected_meta_value = list(meta[i].items())[0]
|
|
assert el.meta[expected_meta_key] == expected_meta_value
|
|
|
|
def test_run_with_meta_and_bytestreams(self):
|
|
"""
|
|
Test if the component runs correctly with ByteStream inputs and meta.
|
|
The original meta is preserved and the new meta is added.
|
|
"""
|
|
|
|
bs = ByteStream.from_string("Haystack!", mime_type="text/plain", meta={"foo": "bar"})
|
|
meta = {"another_key": "another_value"}
|
|
router = FileTypeRouter(mime_types=[r"text/plain"])
|
|
output = router.run(sources=[bs], meta=meta)
|
|
|
|
assert output
|
|
assert len(output[r"text/plain"]) == 1
|
|
assert not output.get("unclassified")
|
|
|
|
assert isinstance(output[r"text/plain"][0], ByteStream)
|
|
assert output[r"text/plain"][0].meta["foo"] == "bar"
|
|
assert output[r"text/plain"][0].meta["another_key"] == "another_value"
|
|
|
|
def test_run_fails_if_meta_length_does_not_match_sources(self, test_files_path):
|
|
"""
|
|
Test that the component raises an error if the length of the metadata list does not match the number of sources.
|
|
"""
|
|
file_paths = [test_files_path / "txt" / "doc_1.txt"]
|
|
meta = [{"key1": "value1"}, {"key2": "value2"}, {"key3": "value3"}]
|
|
router = FileTypeRouter(mime_types=[r"text/plain"])
|
|
|
|
with pytest.raises(ValueError):
|
|
router.run(sources=file_paths, meta=meta)
|
|
|
|
def test_run_with_bytestreams(self, test_files_path):
|
|
"""
|
|
Test if the component runs correctly with ByteStream inputs.
|
|
"""
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "txt" / "doc_2.txt",
|
|
test_files_path / "audio" / "the context for this answer is here.wav",
|
|
test_files_path / "images" / "apple.jpg",
|
|
]
|
|
mime_types = [r"text/plain", r"text/plain", r"audio/x-wav", r"image/jpeg"]
|
|
# Convert file paths to ByteStream objects and set metadata
|
|
byte_streams = []
|
|
for path, mime_type in zip(file_paths, mime_types, strict=True):
|
|
byte_streams.append(ByteStream(path.read_bytes(), mime_type=mime_type))
|
|
|
|
# add unclassified ByteStream
|
|
bs = ByteStream(b"unclassified content")
|
|
bs.meta["content_type"] = "unknown_type"
|
|
byte_streams.append(bs)
|
|
|
|
router = FileTypeRouter(mime_types=[r"text/plain", r"audio/x-wav", r"image/jpeg"])
|
|
output = router.run(sources=byte_streams)
|
|
assert output
|
|
assert len(output[r"text/plain"]) == 2
|
|
assert len(output[r"audio/x-wav"]) == 1
|
|
assert len(output[r"image/jpeg"]) == 1
|
|
assert len(output.get("unclassified")) == 1
|
|
|
|
def test_run_with_bytestreams_and_file_paths(self, test_files_path):
|
|
"""
|
|
Test if the component raises an error for unsupported data source types.
|
|
"""
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "audio" / "the context for this answer is here.wav",
|
|
test_files_path / "txt" / "doc_2.txt",
|
|
test_files_path / "images" / "apple.jpg",
|
|
test_files_path / "markdown" / "sample.md",
|
|
]
|
|
mime_types = [r"text/plain", r"audio/x-wav", r"text/plain", r"image/jpeg", r"text/markdown"]
|
|
byte_stream_sources = []
|
|
for path, mime_type in zip(file_paths, mime_types, strict=True):
|
|
byte_stream_sources.append(ByteStream(path.read_bytes(), mime_type=mime_type))
|
|
|
|
mixed_sources = file_paths[:2] + byte_stream_sources[2:]
|
|
|
|
router = FileTypeRouter(mime_types=[r"text/plain", r"audio/x-wav", r"image/jpeg", r"text/markdown"])
|
|
output = router.run(sources=mixed_sources)
|
|
assert len(output[r"text/plain"]) == 2
|
|
assert len(output[r"audio/x-wav"]) == 1
|
|
assert len(output[r"image/jpeg"]) == 1
|
|
assert len(output[r"text/markdown"]) == 1
|
|
|
|
def test_no_files(self):
|
|
"""
|
|
Test that the component runs correctly when no files are provided.
|
|
"""
|
|
router = FileTypeRouter(mime_types=[r"text/plain", r"audio/x-wav", r"image/jpeg"])
|
|
output = router.run(sources=[])
|
|
assert not output
|
|
|
|
def test_unlisted_extensions(self, test_files_path):
|
|
"""
|
|
Test that the component correctly handles files with non specified mime types.
|
|
"""
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "audio" / "ignored.mp3",
|
|
test_files_path / "audio" / "this is the content of the document.wav",
|
|
]
|
|
router = FileTypeRouter(mime_types=[r"text/plain"], raise_on_failure=False)
|
|
output = router.run(sources=file_paths)
|
|
assert len(output[r"text/plain"]) == 1
|
|
assert "mp3" not in output
|
|
assert len(output.get("unclassified")) == 1
|
|
assert output.get("failed")[0].name == "ignored.mp3"
|
|
|
|
def test_no_extension(self, test_files_path):
|
|
"""
|
|
Test that the component ignores files with no extension.
|
|
"""
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "txt" / "doc_2.txt",
|
|
test_files_path / "txt" / "doc_4",
|
|
]
|
|
router = FileTypeRouter(mime_types=[r"text/plain"])
|
|
output = router.run(sources=file_paths)
|
|
assert len(output[r"text/plain"]) == 2
|
|
assert len(output.get("unclassified")) == 1
|
|
|
|
def test_unsupported_source_type(self):
|
|
"""
|
|
Test if the component raises an error for unsupported data source types.
|
|
"""
|
|
router = FileTypeRouter(mime_types=[r"text/plain", r"audio/x-wav", r"image/jpeg"])
|
|
with pytest.raises(TypeError, match="Unsupported data source type:"):
|
|
router.run(sources=[{"unsupported": "type"}])
|
|
|
|
def test_invalid_regex_pattern(self):
|
|
"""
|
|
Test that the component raises a ValueError for invalid regex patterns.
|
|
"""
|
|
with pytest.raises(ValueError, match="Invalid MIME type or regex pattern"):
|
|
FileTypeRouter(mime_types=["[Invalid-Regex"])
|
|
|
|
def test_regex_mime_type_matching(self, test_files_path):
|
|
"""
|
|
Test if the component correctly matches mime types using regex.
|
|
"""
|
|
router = FileTypeRouter(mime_types=[r"text\/.*", r"audio\/.*", r"image\/.*"])
|
|
file_paths = [
|
|
test_files_path / "txt" / "doc_1.txt",
|
|
test_files_path / "audio" / "the context for this answer is here.wav",
|
|
test_files_path / "images" / "apple.jpg",
|
|
]
|
|
output = router.run(sources=file_paths)
|
|
assert len(output[r"text\/.*"]) == 1, "Failed to match text file with regex"
|
|
assert len(output[r"audio\/.*"]) == 1, "Failed to match audio file with regex"
|
|
assert len(output[r"image\/.*"]) == 1, "Failed to match image file with regex"
|
|
|
|
@patch("pathlib.Path.open", new_callable=mock_open, read_data=b"Mock file content.")
|
|
def test_exact_mime_type_matching(self, mock_file):
|
|
"""
|
|
Test if the component correctly matches mime types exactly, without regex patterns.
|
|
"""
|
|
txt_stream = ByteStream(io.BytesIO(b"Text file content").read(), mime_type="text/plain")
|
|
jpg_stream = ByteStream(io.BytesIO(b"JPEG file content").read(), mime_type="image/jpeg")
|
|
mp3_stream = ByteStream(io.BytesIO(b"MP3 file content").read(), mime_type="audio/mpeg")
|
|
|
|
byte_streams = [txt_stream, jpg_stream, mp3_stream]
|
|
router = FileTypeRouter(mime_types=["text/plain", "image/jpeg"])
|
|
output = router.run(sources=byte_streams)
|
|
|
|
assert len(output["text/plain"]) == 1, "Failed to match 'text/plain' MIME type exactly"
|
|
assert txt_stream in output["text/plain"], "'doc_1.txt' ByteStream not correctly classified as 'text/plain'"
|
|
|
|
assert len(output["image/jpeg"]) == 1, "Failed to match 'image/jpeg' MIME type exactly"
|
|
assert jpg_stream in output["image/jpeg"], "'apple.jpg' ByteStream not correctly classified as 'image/jpeg'"
|
|
|
|
assert len(output.get("unclassified")) == 1, "Failed to handle unclassified file types"
|
|
assert mp3_stream in output["unclassified"], "'sound.mp3' ByteStream should be unclassified but is not"
|
|
|
|
@pytest.mark.parametrize(
|
|
"literal_mime",
|
|
[
|
|
"image/svg+xml",
|
|
"application/ld+json",
|
|
"application/atom+xml",
|
|
"application/vnd.api+json",
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
],
|
|
)
|
|
def test_literal_mime_with_regex_metacharacters_matches_self(self, literal_mime):
|
|
"""MIME types with regex metacharacters (`+`, `.`) match themselves via the equality short-circuit."""
|
|
router = FileTypeRouter(mime_types=[literal_mime])
|
|
bs = ByteStream(b"x", mime_type=literal_mime)
|
|
output = router.run(sources=[bs])
|
|
|
|
assert output[literal_mime] == [bs]
|
|
assert "unclassified" not in output
|
|
|
|
def test_to_dict_from_dict_preserves_literal_and_regex_mix(self):
|
|
"""A literal + regex mix survives serde and still routes correctly."""
|
|
router = FileTypeRouter(mime_types=["image/svg+xml", r"audio/.*"])
|
|
|
|
loaded = FileTypeRouter.from_dict(router.to_dict())
|
|
assert loaded.mime_types == ["image/svg+xml", r"audio/.*"]
|
|
|
|
svg = ByteStream(b"<svg/>", mime_type="image/svg+xml")
|
|
wav = ByteStream(b"x", mime_type="audio/x-wav")
|
|
output = loaded.run(sources=[svg, wav])
|
|
|
|
assert output["image/svg+xml"] == [svg]
|
|
assert output[r"audio/.*"] == [wav]
|
|
assert "unclassified" not in output
|
|
|
|
def test_pipeline_output_socket_name_matches_literal_mime_with_plus(self):
|
|
"""A downstream component wired to `router.image/svg+xml` actually receives the SVG stream."""
|
|
|
|
@component
|
|
class _Sink:
|
|
@component.output_types(received=list)
|
|
def run(self, value: list) -> dict:
|
|
return {"received": value}
|
|
|
|
pipe = Pipeline()
|
|
pipe.add_component(instance=FileTypeRouter(mime_types=["image/svg+xml"]), name="router")
|
|
pipe.add_component(instance=_Sink(), name="sink")
|
|
pipe.connect("router.image/svg+xml", "sink.value")
|
|
|
|
svg = ByteStream(b"<svg/>", mime_type="image/svg+xml")
|
|
output = pipe.run(data={"router": {"sources": [svg]}})
|
|
|
|
assert output["sink"]["received"] == [svg]
|
|
|
|
def test_additional_mimetypes_with_literal_plus(self, tmp_path):
|
|
"""Custom `additional_mimetypes` registration still works for `+`-containing literal MIMEs."""
|
|
custom_mime = "application/x-custom+xml"
|
|
custom_extension = ".cxml"
|
|
test_file = tmp_path / f"test{custom_extension}"
|
|
test_file.touch()
|
|
|
|
router = FileTypeRouter(mime_types=[custom_mime], additional_mimetypes={custom_mime: custom_extension})
|
|
output = router.run(sources=[test_file])
|
|
|
|
assert custom_mime in output
|
|
assert test_file in output[custom_mime]
|
|
assert "unclassified" not in output
|
|
|
|
def test_serde_in_pipeline(self):
|
|
"""
|
|
Test if a pipeline containing the component can be serialized and deserialized without errors.
|
|
"""
|
|
|
|
file_type_router = FileTypeRouter(mime_types=["text/plain", "application/pdf"])
|
|
|
|
pipeline = Pipeline()
|
|
pipeline.add_component(instance=file_type_router, name="file_type_router")
|
|
|
|
pipeline_dict = pipeline.to_dict()
|
|
|
|
assert pipeline_dict == {
|
|
"metadata": {},
|
|
"max_runs_per_component": 100,
|
|
"connection_type_validation": True,
|
|
"components": {
|
|
"file_type_router": {
|
|
"type": "haystack.components.routers.file_type_router.FileTypeRouter",
|
|
"init_parameters": {
|
|
"mime_types": ["text/plain", "application/pdf"],
|
|
"additional_mimetypes": None,
|
|
"raise_on_failure": False,
|
|
},
|
|
}
|
|
},
|
|
"connections": [],
|
|
}
|
|
|
|
pipeline_yaml = pipeline.dumps()
|
|
|
|
new_pipeline = Pipeline.loads(pipeline_yaml)
|
|
assert new_pipeline == pipeline
|
|
|
|
@pytest.mark.integration
|
|
def test_pipeline_with_converters(self, test_files_path):
|
|
"""
|
|
Test if the component runs correctly in a pipeline with converters and passes metadata correctly.
|
|
"""
|
|
file_type_router = FileTypeRouter(
|
|
mime_types=["text/plain", "application/pdf"],
|
|
additional_mimetypes={"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"},
|
|
)
|
|
|
|
pipe = Pipeline()
|
|
pipe.add_component(instance=file_type_router, name="file_type_router")
|
|
pipe.add_component(instance=TextFileToDocument(), name="text_file_converter")
|
|
pipe.add_component(instance=PyPDFToDocument(), name="pypdf_converter")
|
|
pipe.connect("file_type_router.text/plain", "text_file_converter.sources")
|
|
pipe.connect("file_type_router.application/pdf", "pypdf_converter.sources")
|
|
|
|
file_paths = [test_files_path / "txt" / "doc_1.txt", test_files_path / "pdf" / "sample_pdf_1.pdf"]
|
|
|
|
meta = [{"meta_field_1": "meta_value_1"}, {"meta_field_2": "meta_value_2"}]
|
|
|
|
output = pipe.run(data={"file_type_router": {"sources": file_paths, "meta": meta}})
|
|
|
|
assert output["text_file_converter"]["documents"][0].meta["meta_field_1"] == "meta_value_1"
|
|
assert output["pypdf_converter"]["documents"][0].meta["meta_field_2"] == "meta_value_2"
|
|
|
|
def test_additional_mimetypes_integration(self, tmp_path):
|
|
"""
|
|
Test if the component runs correctly in a pipeline with additional mimetypes correctly.
|
|
"""
|
|
custom_mime_type = "application/x-spam"
|
|
custom_extension = ".spam"
|
|
test_file = tmp_path / f"test.{custom_extension}"
|
|
test_file.touch()
|
|
|
|
# confirm that mimetypes module doesn't know about this extension by default
|
|
assert custom_mime_type not in mimetypes.types_map.values()
|
|
|
|
# make haystack aware of the custom mime type
|
|
router = FileTypeRouter(
|
|
mime_types=[custom_mime_type], additional_mimetypes={custom_mime_type: custom_extension}
|
|
)
|
|
mappings = router.run(sources=[test_file])
|
|
|
|
# assert the file was classified under the custom mime type
|
|
assert custom_mime_type in mappings
|
|
assert test_file in mappings[custom_mime_type]
|
|
|
|
def test_non_existent_file(self):
|
|
"""
|
|
Test conditional FileNotFoundError behavior in FileTypeRouter.
|
|
"""
|
|
router = FileTypeRouter(mime_types=[r"text/plain"], raise_on_failure=True)
|
|
|
|
# no metadata
|
|
with pytest.raises(FileNotFoundError):
|
|
router.run(sources=["non_existent.txt"])
|
|
|
|
# with metadata
|
|
with pytest.raises(FileNotFoundError):
|
|
router.run(sources=["non_existent.txt"], meta={"spam": "eggs"})
|
|
|
|
def test_logging_for_non_existent_file(self, caplog: pytest.LogCaptureFixture):
|
|
"""
|
|
Test that a logging warning is triggered when a non-existent file is encountered
|
|
and raise_on_failure is False.
|
|
"""
|
|
import logging
|
|
|
|
router = FileTypeRouter(mime_types=[r"text/plain"], raise_on_failure=False)
|
|
|
|
# Capture log messages to verify they are triggered
|
|
with caplog.at_level(logging.WARNING):
|
|
result = router.run(sources=["non_existent_file.txt"])
|
|
assert "File not found:" in caplog.text
|
|
assert "Skipping it." in caplog.text
|
|
|
|
# Verify the file is added to the "failed" category
|
|
assert "failed" in result
|
|
assert PosixPath("non_existent_file.txt") in result["failed"]
|
|
assert "text/plain" not in result
|