import json import stat from cmdforge.usage import ( build_composite_tool, clear_usage, get_suggestions, is_enabled, record_invocation, set_enabled, ) class Stream: def __init__(self, inode=None): self.inode = inode def configure_paths(tmp_path, monkeypatch): monkeypatch.setattr("cmdforge.usage.CONFIG_DIR", tmp_path) monkeypatch.setattr("cmdforge.usage.USAGE_FILE", tmp_path / "usage.json") monkeypatch.setattr("cmdforge.usage.USAGE_LOCK_FILE", tmp_path / ".usage.lock") def test_tracking_is_disabled_by_default(tmp_path, monkeypatch): configure_paths(tmp_path, monkeypatch) assert not is_enabled() monkeypatch.setattr("cmdforge.usage._pipe_inode", lambda stream: stream.inode) record_invocation("one", Stream(), Stream()) assert not (tmp_path / "usage.json").exists() def test_detects_only_pipe_link_and_stores_no_content(tmp_path, monkeypatch): configure_paths(tmp_path, monkeypatch) set_enabled(True) monkeypatch.setattr("cmdforge.usage._pipe_inode", lambda stream: stream.inode) for inode in range(44, 47): record_invocation("tool-a", Stream(), Stream(inode)) record_invocation("tool-b", Stream(inode), Stream()) assert get_suggestions() == [{ "tools": ["tool-a", "tool-b"], "count": 3, "last_seen": get_suggestions()[0]["last_seen"], }] data = json.loads((tmp_path / "usage.json").read_text()) serialized = json.dumps(data) for forbidden in ("input_text", "output", "arguments", "cwd", "environment"): assert forbidden not in serialized assert stat.S_IMODE((tmp_path / "usage.json").stat().st_mode) == 0o600 def test_disable_and_clear_preserve_consent_state(tmp_path, monkeypatch): configure_paths(tmp_path, monkeypatch) set_enabled(True) clear_usage() assert is_enabled() set_enabled(False) assert not is_enabled() def test_build_composite_tool_preserves_pipeline_order(): tool = build_composite_tool("combined", ["tool-a", "tool-b"]) assert tool.dependencies == ["tool-a", "tool-b"] assert tool.steps[0].input_template == "{input}" assert tool.steps[1].input_template == "{pipeline_1}" assert tool.output == "{pipeline_2}"