wehub-resource-sync
c56bef871b
Sync docs with Docusaurus / sync (push) Waiting to run
Tests / Check if changed (push) Waiting to run
Tests / format (push) Blocked by required conditions
Tests / check-imports (push) Blocked by required conditions
Tests / Unit / macos-latest (push) Blocked by required conditions
Tests / Unit / ubuntu-latest (push) Blocked by required conditions
Tests / Unit / windows-latest (push) Blocked by required conditions
Tests / mypy (push) Blocked by required conditions
Tests / Integration / ubuntu-latest (push) Blocked by required conditions
Tests / Integration / macos-latest (push) Blocked by required conditions
Tests / Integration / windows-latest (push) Blocked by required conditions
Tests / notify-slack-on-failure (push) Blocked by required conditions
Tests / Mark tests as completed (push) Blocked by required conditions
Docker image release / Build base image (push) Waiting to run
CodeQL / Analyze (python) (push) Has been cancelled
Update Platform Components Table / update (push) Has been cancelled
102 lines
4.2 KiB
Python
102 lines
4.2 KiB
Python
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
|
|
#
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
import pytest
|
|
|
|
from haystack.core.component.sockets import InputSocket, OutputSocket, Sockets
|
|
from haystack.testing.factory import component_class
|
|
|
|
|
|
class TestSockets:
|
|
def test_init(self):
|
|
comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
|
|
sockets: dict[str, InputSocket | OutputSocket] = {
|
|
"input_1": InputSocket("input_1", int),
|
|
"input_2": InputSocket("input_2", int),
|
|
}
|
|
io = Sockets(component=comp, sockets_dict=sockets, sockets_io_type=InputSocket)
|
|
assert io._component == comp
|
|
assert "input_1" in io.__dict__
|
|
assert io.__dict__["input_1"] == comp.__haystack_input__._sockets_dict["input_1"] # type: ignore[attr-defined]
|
|
assert "input_2" in io.__dict__
|
|
assert io.__dict__["input_2"] == comp.__haystack_input__._sockets_dict["input_2"] # type: ignore[attr-defined]
|
|
|
|
def test_init_with_empty_sockets(self):
|
|
comp = component_class("SomeComponent")()
|
|
io = Sockets(component=comp, sockets_dict={}, sockets_io_type=InputSocket)
|
|
|
|
assert io._component == comp
|
|
assert io._sockets_dict == {}
|
|
|
|
def test_getattribute(self):
|
|
comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
|
|
io = Sockets(
|
|
component=comp,
|
|
sockets_dict=comp.__haystack_input__._sockets_dict, # type: ignore[attr-defined]
|
|
sockets_io_type=InputSocket,
|
|
)
|
|
|
|
assert io.input_1 == comp.__haystack_input__._sockets_dict["input_1"] # type: ignore[attr-defined]
|
|
assert io.input_2 == comp.__haystack_input__._sockets_dict["input_2"] # type: ignore[attr-defined]
|
|
|
|
def test_getattribute_non_existing_socket(self):
|
|
comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
|
|
io = Sockets(
|
|
component=comp,
|
|
sockets_dict=comp.__haystack_input__._sockets_dict, # type: ignore[attr-defined]
|
|
sockets_io_type=InputSocket,
|
|
)
|
|
|
|
with pytest.raises(AttributeError):
|
|
io.input_3
|
|
|
|
def test_repr(self):
|
|
comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
|
|
io = Sockets(
|
|
component=comp,
|
|
sockets_dict=comp.__haystack_input__._sockets_dict, # type: ignore[attr-defined]
|
|
sockets_io_type=InputSocket,
|
|
)
|
|
res = repr(io)
|
|
assert res == "Inputs:\n - input_1: int\n - input_2: int"
|
|
|
|
def test_get(self):
|
|
comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
|
|
io = Sockets(
|
|
component=comp,
|
|
sockets_dict=comp.__haystack_input__._sockets_dict, # type: ignore[attr-defined]
|
|
sockets_io_type=InputSocket,
|
|
)
|
|
|
|
assert io.get("input_1") == comp.__haystack_input__._sockets_dict["input_1"] # type: ignore[attr-defined]
|
|
assert io.get("input_2") == comp.__haystack_input__._sockets_dict["input_2"] # type: ignore[attr-defined]
|
|
assert io.get("invalid") is None
|
|
assert io.get("invalid", InputSocket("input_2", int)) == InputSocket("input_2", int)
|
|
|
|
def test_contains(self):
|
|
comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
|
|
io = Sockets(
|
|
component=comp,
|
|
sockets_dict=comp.__haystack_input__._sockets_dict, # type: ignore[attr-defined]
|
|
sockets_io_type=InputSocket,
|
|
)
|
|
|
|
assert "input_1" in io
|
|
assert "input_2" in io
|
|
assert "invalid" not in io
|
|
|
|
|
|
def test_input_socket_variadic_without_type_arg_raises_component_error():
|
|
"""Bare Variadic (no type argument) should raise ComponentError, not IndexError."""
|
|
from collections.abc import Iterable
|
|
from typing import Annotated
|
|
|
|
from haystack.core.component.types import HAYSTACK_VARIADIC_ANNOTATION
|
|
from haystack.core.errors import ComponentError
|
|
|
|
bare_variadic = Annotated[Iterable, HAYSTACK_VARIADIC_ANNOTATION]
|
|
|
|
with pytest.raises(ComponentError, match="must have a type argument"):
|
|
InputSocket("inputs", bare_variadic)
|