test(client): add comprehensive unit tests for client abstractions

- Implement tests for BaseClient list action and payload delegation
- Add tests for HttpClient service info environment reading
- Include tests for HttpClient fallback behavior with invalid service info
- Add tests for HttpClient display formatting with response envelopes
- Implement tests for MCPClient transport validation

test(common utils): add unit tests for stream utility helpers

- Create tests for chunk formatting across different transport formats
- Add tests for stream task execution with remaining chunk flushing
- Include tests for exception handling in stream tasks

test(dingtalk utils): update tests to include timeout parameter validation

- Modify fake post function to capture timeout value
- Add assertion to verify timeout parameter is passed correctly

test(env utils): add comprehensive tests for environment loading utilities

- Test environment file parsing with comments and invalid lines
- Add tests for explicit path loading with override behavior
- Include tests for cached value handling in implicit loading

test(schema): add unit tests for pydantic schema functionality

- Test embedding node coercion to float16 and JSON serialization
- Add tests for extra field allowance in request/response schemas
- Include tests for component enum key acceptance in configs

test(service utils): add comprehensive tests for service discovery helpers

- Test PID retrieval from port with multiple results
- Add tests for flowllm process scanning and argument parsing
- Include tests for start precheck with existing instances
- Add tests for port occupation detection and error handling

Author: jinliyl

tests/unit/test_client.py

@@ -0,0 +1,86 @@
+"""Unit tests for client abstractions."""
+
+# pylint: disable=protected-access,invalid-overridden-method
+
+import asyncio
+import json
+
+import pytest
+
+from flowllm.components.client import BaseClient, HttpClient, MCPClient
+from flowllm.constants import FLOWLLM_DEFAULT_HOST, FLOWLLM_DEFAULT_PORT, FLOWLLM_SERVICE_INFO
+
+
+class _FakeClient(BaseClient):
+    """Concrete BaseClient for exercising wrapper behavior."""
+
+    def __init__(self, actions=None, **kwargs):
+        super().__init__(**kwargs)
+        self.actions = actions or [{"action": "search"}]
+        self.payloads = []
+
+    async def list_actions(self):
+        return self.actions
+
+    async def _execute(self, action: str, payload: dict):
+        self.payloads.append((action, payload))
+        yield "chunk"
+
+
+def test_base_client_list_action_returns_pretty_json():
+    """The synthetic list action uses list_actions instead of _execute."""
+
+    async def run():
+        client = _FakeClient(actions=[{"action": "health_check", "method": "GET"}])
+        chunks = [chunk async for chunk in client("list")]
+        assert json.loads(chunks[0]) == [{"action": "health_check", "method": "GET"}]
+        assert not client.payloads
+
+    asyncio.run(run())
+
+
+def test_base_client_delegates_non_list_actions():
+    """Regular actions pass through to _execute with kwargs as payload."""
+
+    async def run():
+        client = _FakeClient()
+        assert [chunk async for chunk in client("search", query="hi")] == ["chunk"]
+        assert client.payloads == [("search", {"query": "hi"})]
+
+    asyncio.run(run())
+
+
+def test_http_client_reads_service_info_from_environment(monkeypatch):
+    """HTTP client host/port default from FLOWLLM_SERVICE_INFO when present."""
+    monkeypatch.setenv(FLOWLLM_SERVICE_INFO, json.dumps({"host": "10.0.0.1", "port": 8123}))
+
+    client = HttpClient()
+
+    assert client.base_url == "http://10.0.0.1:8123"
+
+
+def test_http_client_falls_back_on_invalid_service_info(monkeypatch):
+    """Invalid service info is ignored in favor of built-in defaults."""
+    monkeypatch.setenv(FLOWLLM_SERVICE_INFO, "not-json")
+
+    client = HttpClient()
+
+    assert client.base_url == f"http://{FLOWLLM_DEFAULT_HOST}:{FLOWLLM_DEFAULT_PORT}"
+
+
+def test_http_client_format_for_display_handles_response_envelope():
+    """Response envelopes render answer first with status and metadata."""
+    rendered = HttpClient._format_for_display(
+        json.dumps({"answer": "done", "success": True, "metadata": {"cost": 1}, "extra": "x"}),
+    )
+
+    assert rendered.splitlines()[0] == "done"
+    assert "\u2705" in rendered
+    assert '{"cost": 1}' in rendered
+    assert '"extra": "x"' in rendered
+
+
+def test_mcp_client_rejects_unknown_transport():
+    """Unsupported MCP transport names fail during construction."""
+    with pytest.raises(ValueError, match="Unknown transport"):
+        MCPClient(transport="websocket")

tests/unit/test_common_utils.py

@@ -0,0 +1,61 @@
+"""Unit tests for stream utility helpers."""
+
+import asyncio
+import json
+
+import pytest
+
+from flowllm.enumeration import ChunkEnum
+from flowllm.schema import StreamChunk
+from flowllm.utils.common_utils import _format_chunk, execute_stream_task
+
+
+def test_format_chunk_supports_str_bytes_chunk_and_done_marker():
+    """Chunks are rendered correctly for all transport formats."""
+    chunk = StreamChunk(chunk_type=ChunkEnum.CONTENT, chunk="hello")
+
+    as_str = _format_chunk(chunk, "str")
+    assert isinstance(as_str, str)
+    assert as_str.startswith("data:")
+    assert json.loads(as_str.removeprefix("data:").strip())["chunk"] == "hello"
+
+    assert _format_chunk(chunk, "bytes") == as_str.encode()
+    assert _format_chunk(chunk, "chunk") is chunk
+    assert _format_chunk(StreamChunk(done=True), "str") == "data:[DONE]\n\n"
+
+
+def test_execute_stream_task_flushes_remaining_chunks_after_task_finishes():
+    """A completed producer still lets queued chunks drain before DONE."""
+
+    async def run():
+        queue: asyncio.Queue[StreamChunk] = asyncio.Queue()
+        await queue.put(StreamChunk(chunk="one"))
+        await queue.put(StreamChunk(chunk="two"))
+
+        async def producer():
+            return None
+
+        task = asyncio.create_task(producer())
+        items = [item async for item in execute_stream_task(queue, task, output_format="chunk")]
+
+        assert [item.chunk for item in items] == ["one", "two", ""]
+        assert items[-1].done is True
+        assert items[-1].chunk_type == ChunkEnum.DONE
+
+    asyncio.run(run())
+
+
+def test_execute_stream_task_raises_task_exception():
+    """Producer exceptions are surfaced to callers."""
+
+    async def run():
+        queue: asyncio.Queue[StreamChunk] = asyncio.Queue()
+
+        async def producer():
+            raise RuntimeError("boom")
+
+        task = asyncio.create_task(producer())
+        with pytest.raises(RuntimeError, match="boom"):
+            _ = [item async for item in execute_stream_task(queue, task, task_name="job")]
+
+    asyncio.run(run())

tests/unit/test_dingtalk_utils.py

@@ -51,15 +51,17 @@ def test_send_dingtalk_text_message(monkeypatch):
     monkeypatch.setenv("DING_DAILY_API_TOKEN", "daily-token")
     monkeypatch.setattr(dingtalk_utils, "load_env", lambda: {})
 
-    def fake_post(_url, json, _timeout):
+    def fake_post(_url, json, timeout):
         seen["json"] = json
+        seen["timeout"] = timeout
         return FakeResponse()
 
     monkeypatch.setattr(dingtalk_utils.httpx, "post", fake_post)
 
     dingtalk_utils.send_dingtalk_message("Ignored", "hello", msgtype="text")
 
     assert seen["json"] == {"msgtype": "text", "text": {"content": "hello"}}
+    assert seen["timeout"] == 10.0
 
 
 def test_send_dingtalk_message_requires_token(monkeypatch):

tests/unit/test_env_utils.py

@@ -0,0 +1,58 @@
+"""Unit tests for environment loading helpers."""
+
+# pylint: disable=protected-access
+
+from flowllm.utils import env_utils
+
+
+def test_parse_env_file_skips_comments_and_invalid_lines(tmp_path):
+    """Only valid KEY=VALUE entries are returned."""
+    env_file = tmp_path / ".env"
+    env_file.write_text(
+        """
+        # comment
+        FOO=bar
+        QUOTED="hello world"
+        EMPTY_KEY_IGNORED
+        =missing
+        SPACED = 'trimmed'
+        """,
+        encoding="utf-8",
+    )
+
+    assert env_utils._parse_env_file(env_file) == {
+        "FOO": "bar",
+        "QUOTED": "hello world",
+        "SPACED": "trimmed",
+    }
+
+
+def test_load_env_from_explicit_path_respects_override(monkeypatch, tmp_path):
+    """Explicit .env loading does not overwrite existing values when override=False."""
+    env_file = tmp_path / ".env"
+    env_file.write_text("FLOWLLM_ENV_TEST=file\nNEW_VALUE=created\n", encoding="utf-8")
+    monkeypatch.setenv("FLOWLLM_ENV_TEST", "existing")
+
+    loaded = env_utils.load_env(env_file, override=False)
+
+    assert loaded == {"NEW_VALUE": "created"}
+    assert env_utils.os.environ["FLOWLLM_ENV_TEST"] == "existing"
+    assert env_utils.os.environ["NEW_VALUE"] == "created"
+
+
+def test_load_env_searches_cwd_once_and_returns_cached_values(monkeypatch, tmp_path):
+    """Implicit .env loading caches the discovered values."""
+    env_file = tmp_path / ".env"
+    env_file.write_text("FLOWLLM_CACHED_ENV=first\n", encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.delenv("FLOWLLM_CACHED_ENV", raising=False)
+    monkeypatch.setattr(env_utils, "_LOADED", False)
+    monkeypatch.setattr(env_utils, "_LOADED_VALUES", {})
+
+    assert env_utils.load_env() == {"FLOWLLM_CACHED_ENV": "first"}
+
+    env_file.write_text("FLOWLLM_CACHED_ENV=second\n", encoding="utf-8")
+    monkeypatch.setenv("FLOWLLM_CACHED_ENV", "changed")
+
+    assert env_utils.load_env() == {"FLOWLLM_CACHED_ENV": "first"}
+    assert env_utils.os.environ["FLOWLLM_CACHED_ENV"] == "changed"

tests/unit/test_schema.py

@@ -0,0 +1,40 @@
+"""Unit tests for pydantic schema helpers."""
+
+import numpy as np
+
+from flowllm.enumeration import ComponentEnum
+from flowllm.schema import ApplicationConfig, ComponentConfig, EmbNode, Request, Response
+
+
+def test_emb_node_coerces_embedding_to_float16_and_serializes_to_list():
+    """Embedding vectors accept plain lists and remain JSON serializable."""
+    node = EmbNode(text="hello", embedding=[1.25, 2.5], metadata={"source": "unit"})
+
+    assert isinstance(node.embedding, np.ndarray)
+    assert node.embedding.dtype == np.float16
+    assert node.model_dump()["embedding"] == [1.25, 2.5]
+    assert node.metadata == {"source": "unit"}
+
+
+def test_request_response_component_config_allow_extra_fields():
+    """Schemas intentionally preserve backend-specific payload fields."""
+    request = Request(metadata={"trace_id": "1"}, query="hello")
+    response = Response(answer="ok", score=0.9)
+    component = ComponentConfig(backend="custom", host="127.0.0.1")
+
+    assert request.query == "hello"
+    assert response.score == 0.9
+    assert component.host == "127.0.0.1"
+
+
+def test_application_config_accepts_component_enum_keys():
+    """Component config maps can be keyed by ComponentEnum values."""
+    config = ApplicationConfig(
+        components={
+            ComponentEnum.CLIENT: {
+                "http": ComponentConfig(backend="http"),
+            },
+        },
+    )
+
+    assert config.components[ComponentEnum.CLIENT]["http"].backend == "http"

tests/unit/test_service_utils.py

@@ -0,0 +1,61 @@
+"""Unit tests for service discovery helpers."""
+
+# pylint: disable=protected-access
+
+import pytest
+
+from flowllm.constants import FLOWLLM_DEFAULT_HOST, FLOWLLM_DEFAULT_PORT
+from flowllm.utils import service_utils
+
+
+def test_pid_on_port_returns_first_pid(monkeypatch):
+    """Only the first lsof result is used."""
+    monkeypatch.setattr(service_utils, "_sh", lambda _cmd: "123\n456\n")
+
+    assert service_utils._pid_on_port(8080) == 123
+
+
+def test_scan_flowllm_procs_extracts_host_and_port(monkeypatch):
+    """flowllm start process args are parsed for service host and port."""
+    monkeypatch.setattr(
+        service_utils,
+        "_sh",
+        lambda _cmd: (
+            "111 flowllm start service.host=0.0.0.0 service.port=9000\n"
+            "222 python -m flowllm start\n"
+            "bad not-a-pid\n"
+        ),
+    )
+
+    assert service_utils._scan_flowllm_procs() == [
+        (111, "0.0.0.0", 9000),
+        (222, FLOWLLM_DEFAULT_HOST, FLOWLLM_DEFAULT_PORT),
+    ]
+
+
+def test_precheck_start_returns_false_when_flowllm_already_running(monkeypatch, capsys):
+    """Existing FlowLLM instance prevents a second start."""
+
+    async def fake_find_flowllm(host, port):
+        assert (host, port) == ("127.0.0.9", 7777)
+        return "flowllm"
+
+    monkeypatch.setattr(service_utils, "find_flowllm", fake_find_flowllm)
+
+    assert service_utils.precheck_start({"host": "127.0.0.9", "port": "7777"}) is False
+    assert "flowllm already running at 127.0.0.9:7777" in capsys.readouterr().out
+
+
+def test_precheck_start_exits_when_port_occupied(monkeypatch, capsys):
+    """Non-FlowLLM listeners on the requested port are reported as errors."""
+
+    async def fake_find_flowllm(_host, _port):
+        return "occupied"
+
+    monkeypatch.setattr(service_utils, "find_flowllm", fake_find_flowllm)
+
+    with pytest.raises(SystemExit) as exc:
+        service_utils.precheck_start({"port": 8888})
+
+    assert exc.value.code == 1
+    assert "port 8888 occupied" in capsys.readouterr().err