test: add unit tests for github

anakin87 · anakin87 · commit 3f08634a4746 · 2026-05-06T14:11:53.000+02:00
diff --git a/integrations/github/tests/test_file_editor.py b/integrations/github/tests/test_file_editor.py
@@ -280,3 +280,94 @@ def test_get_request_headers_with_empty_token(self, monkeypatch):
         assert "Authorization" not in headers
         assert headers["Accept"] == "application/vnd.github.v3+json"
         assert headers["User-Agent"] == "Haystack/GitHubFileEditor"
+
+    @pytest.mark.parametrize(
+        "file_b64,original,expected",
+        [
+            ("SGVsbG8gV29ybGQ=", "missing", "Error: Original string not found in file"),
+            ("SGVsbG8gSGVsbG8=", "Hello", "Error: Original string appears multiple times. Please provide more context"),
+        ],
+    )
+    @patch("requests.get")
+    def test_run_edit_string_edge_cases(self, mock_get, file_b64, original, expected, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        mock_get.return_value.json.return_value = {"content": file_b64, "sha": "abc123"}
+        mock_get.return_value.raise_for_status.return_value = None
+
+        editor = GitHubFileEditor()
+        result = editor.run(
+            command=Command.EDIT,
+            payload={"path": "f.txt", "original": original, "replacement": "X", "message": "m"},
+            repo="owner/repo",
+            branch="main",
+        )
+        assert result["result"] == expected
+
+    @patch("requests.get")
+    def test_run_undo_not_last_user(self, mock_get, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+
+        mock_get.return_value.raise_for_status.return_value = None
+        mock_get.return_value.json.side_effect = [
+            [{"author": {"login": "different_user"}, "sha": "sha1"}],
+            {"login": "another_user"},
+        ]
+
+        editor = GitHubFileEditor()
+        result = editor.run(command=Command.UNDO, payload={"message": "m"}, repo="owner/repo", branch="main")
+        assert result["result"] == "Error: Last commit was not made by the current user"
+
+    @pytest.mark.parametrize(
+        "kwargs,expected_substring",
+        [
+            ({"command": Command.EDIT, "payload": {}}, "Error: No repository specified"),
+            (
+                {"command": "bogus", "payload": {}, "repo": "owner/repo"},
+                None,
+            ),
+        ],
+    )
+    def test_run_validation_errors(self, kwargs, expected_substring, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        editor = GitHubFileEditor()
+        if expected_substring is None:
+            with pytest.raises(ValueError):
+                editor.run(**kwargs)
+        else:
+            result = editor.run(**kwargs)
+            assert expected_substring in result["result"]
+
+    @patch("requests.get")
+    def test_run_command_as_string(self, mock_get, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        mock_get.return_value.json.return_value = {"content": "SGVsbG8=", "sha": "abc"}
+        mock_get.return_value.raise_for_status.return_value = None
+
+        editor = GitHubFileEditor(repo="owner/repo")
+        with patch.object(editor, "_update_file", return_value=True):
+            result = editor.run(
+                command="EDIT",
+                payload={"path": "f.txt", "original": "Hello", "replacement": "Hi", "message": "m"},
+            )
+        assert result["result"] == "Edit successful"
+
+    @pytest.mark.parametrize(
+        "command,payload",
+        [
+            (Command.UNDO, {"message": "m"}),
+            (Command.CREATE, {"path": "n.txt", "content": "x", "message": "m"}),
+            (Command.DELETE, {"path": "n.txt", "message": "m"}),
+        ],
+    )
+    @patch("requests.get")
+    def test_run_command_error_handling_no_raise(self, mock_get, command, payload, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        mock_get.side_effect = requests.RequestException("API Error")
+
+        editor = GitHubFileEditor(raise_on_failure=False)
+        with (
+            patch("requests.put", side_effect=requests.RequestException("API Error")),
+            patch("requests.delete", side_effect=requests.RequestException("API Error")),
+        ):
+            result = editor.run(command=command, payload=payload, repo="owner/repo", branch="main")
+        assert "Error: API Error" in result["result"]
diff --git a/integrations/github/tests/test_pr_creator.py b/integrations/github/tests/test_pr_creator.py
@@ -143,3 +143,82 @@ def test_get_request_headers_with_empty_token(self, monkeypatch):
         assert "Authorization" not in headers
         assert headers["Accept"] == "application/vnd.github.v3+json"
         assert headers["User-Agent"] == "Haystack/GitHubPRCreator"
+
+    def test_parse_issue_url_invalid(self, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        pr_creator = GitHubPRCreator()
+        with pytest.raises(ValueError, match="Invalid GitHub issue URL format"):
+            pr_creator._parse_issue_url("https://github.com/owner/repo/pull/1")
+
+    @patch("requests.get")
+    def test_run_returns_error_when_fork_not_found(self, mock_get, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        mock_get.return_value.json.return_value = {"login": "test_user"}
+        mock_get.return_value.raise_for_status.return_value = None
+
+        pr_creator = GitHubPRCreator()
+        with patch.object(pr_creator, "_check_fork_exists", return_value=False):
+            result = pr_creator.run(
+                issue_url="https://github.com/owner/repo/issues/456",
+                title="Test PR",
+                branch="feature-branch",
+                base="main",
+            )
+        assert result == {"result": "Error: Fork not found at test_user/repo"}
+
+    @pytest.mark.parametrize(
+        "fork_payload,raises_request,expected",
+        [
+            ({"fork": True}, False, True),
+            ({"fork": False}, False, False),
+            ({}, False, False),
+            (None, True, False),
+        ],
+    )
+    @patch("requests.get")
+    def test_check_fork_exists(self, mock_get, fork_payload, raises_request, expected, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        if raises_request:
+            mock_get.side_effect = requests.RequestException("boom")
+        else:
+            mock_get.return_value.json.return_value = fork_payload
+            mock_get.return_value.raise_for_status.return_value = None
+
+        pr_creator = GitHubPRCreator()
+        assert pr_creator._check_fork_exists("repo", "test_user") is expected
+
+    @pytest.mark.parametrize("raise_on_failure", [True, False])
+    @patch("requests.post")
+    def test_create_fork(self, mock_post, raise_on_failure, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        mock_post.return_value.json.return_value = {"owner": {"login": "test_user"}}
+        mock_post.return_value.raise_for_status.return_value = None
+
+        pr_creator = GitHubPRCreator(raise_on_failure=raise_on_failure)
+        assert pr_creator._create_fork("owner", "repo") == "test_user"
+
+        mock_post.side_effect = requests.RequestException("boom")
+        if raise_on_failure:
+            with pytest.raises(RuntimeError, match="Failed to create fork"):
+                pr_creator._create_fork("owner", "repo")
+        else:
+            assert pr_creator._create_fork("owner", "repo") is None
+
+    @pytest.mark.parametrize("raise_on_failure", [True, False])
+    @patch("requests.post")
+    @patch("requests.get")
+    def test_create_branch(self, mock_get, mock_post, raise_on_failure, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        mock_get.return_value.json.return_value = {"object": {"sha": "abc123"}}
+        mock_get.return_value.raise_for_status.return_value = None
+        mock_post.return_value.raise_for_status.return_value = None
+
+        pr_creator = GitHubPRCreator(raise_on_failure=raise_on_failure)
+        assert pr_creator._create_branch("owner", "repo", "feature", "main") is True
+
+        mock_get.side_effect = requests.RequestException("boom")
+        if raise_on_failure:
+            with pytest.raises(RuntimeError, match="Failed to create branch"):
+                pr_creator._create_branch("owner", "repo", "feature", "main")
+        else:
+            assert pr_creator._create_branch("owner", "repo", "feature", "main") is False
diff --git a/integrations/github/tests/test_repo_forker.py b/integrations/github/tests/test_repo_forker.py
@@ -272,3 +272,57 @@ def test_get_request_headers_with_empty_token(self, monkeypatch):
         assert "Authorization" not in headers
         assert headers["Accept"] == "application/vnd.github.v3+json"
         assert headers["User-Agent"] == "Haystack/GitHubRepoForker"
+
+    @pytest.mark.parametrize(
+        "status_code,raises,expected",
+        [
+            (200, False, True),
+            (404, False, False),
+            (None, True, False),
+        ],
+    )
+    @patch("requests.get")
+    def test_check_fork_status(self, mock_get, status_code, raises, expected, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        if raises:
+            mock_get.side_effect = requests.RequestException("boom")
+        else:
+            mock_get.return_value.status_code = status_code
+
+        forker = GitHubRepoForker()
+        assert forker._check_fork_status("test_user/repo") is expected
+
+    def test_get_existing_repository_returns_none_on_exception(self, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+
+        forker = GitHubRepoForker()
+        with (
+            patch.object(forker, "_get_authenticated_user", return_value="test_user"),
+            patch("requests.get", side_effect=requests.RequestException("boom")),
+        ):
+            assert forker._get_existing_repository("repo") is None
+
+    @pytest.mark.parametrize("succeeds_within_timeout", [True, False])
+    def test_run_wait_for_completion(self, succeeds_within_timeout, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+
+        forker = GitHubRepoForker(
+            wait_for_completion=True,
+            max_wait_seconds=5 if succeeds_within_timeout else 0,
+            poll_interval=0,
+            create_branch=False,
+        )
+        with (
+            patch.object(forker, "_parse_github_url", return_value=("owner", "repo", "123")),
+            patch.object(forker, "_get_authenticated_user", return_value="test_user"),
+            patch.object(forker, "_get_existing_repository", return_value=None),
+            patch.object(forker, "_create_fork", return_value="test_user/repo"),
+            patch.object(forker, "_check_fork_status", side_effect=[False, True]),
+            patch("time.sleep", lambda _: None),
+        ):
+            if succeeds_within_timeout:
+                result = forker.run(url="https://github.com/owner/repo/issues/123")
+                assert result == {"repo": "test_user/repo", "issue_branch": None}
+            else:
+                with pytest.raises(TimeoutError):
+                    forker.run(url="https://github.com/owner/repo/issues/123")
diff --git a/integrations/github/tests/test_repo_viewer.py b/integrations/github/tests/test_repo_viewer.py
@@ -186,3 +186,52 @@ def test_get_request_headers_with_empty_token(self, monkeypatch):
         assert "Authorization" not in headers
         assert headers["Accept"] == "application/vnd.github.v3+json"
         assert headers["User-Agent"] == "Haystack/GitHubRepoViewer"
+
+    def test_run_no_repo_raises(self, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        viewer = GitHubRepoViewer()
+        with pytest.raises(ValueError, match="Repository not provided"):
+            viewer.run(path="README.md")
+
+    @pytest.mark.parametrize(
+        "file_payload,error_substring",
+        [
+            (
+                {
+                    "name": "big.bin",
+                    "path": "big.bin",
+                    "size": 2_000_000,
+                    "html_url": "https://x",
+                    "content": "",
+                    "encoding": "base64",
+                },
+                "exceeds limit",
+            ),
+            (
+                {
+                    "name": "raw.txt",
+                    "path": "raw.txt",
+                    "size": 5,
+                    "html_url": "https://x",
+                    "content": "Hello",
+                    "encoding": "utf-8",
+                },
+                None,
+            ),
+        ],
+    )
+    @patch("requests.get")
+    def test_run_file_size_and_encoding(self, mock_get, file_payload, error_substring, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "test-token")
+        mock_get.return_value.json.return_value = file_payload
+        mock_get.return_value.raise_for_status.return_value = None
+
+        viewer = GitHubRepoViewer(raise_on_failure=False)
+        result = viewer.run(repo="owner/repo", path=file_payload["path"])
+        doc = result["documents"][0]
+        if error_substring is None:
+            assert doc.content == "Hello"
+            assert doc.meta["type"] == "file_content"
+        else:
+            assert doc.meta["type"] == "error"
+            assert error_substring in doc.content
diff --git a/integrations/github/tests/test_utils.py b/integrations/github/tests/test_utils.py
@@ -0,0 +1,63 @@
+# SPDX-FileCopyrightText: 2023-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+import pytest
+from haystack import Document
+
+from haystack_integrations.tools.github.utils import message_handler, serialize_handlers
+
+
+class TestMessageHandler:
+    @pytest.mark.parametrize(
+        "documents,expected",
+        [
+            (
+                [Document(content="error message", meta={"type": "error"})],
+                "error message",
+            ),
+            (
+                [
+                    Document(content="docs", meta={"type": "dir"}),
+                    Document(content="README.md", meta={"type": "file"}),
+                ],
+                "docsREADME.md",
+            ),
+            (
+                [Document(content="print('hi')", meta={"type": "file_content", "path": "main.py"})],
+                "File Content for main.py\n\nprint('hi')",
+            ),
+        ],
+    )
+    def test_message_handler_renders_documents(self, documents, expected):
+        assert message_handler(documents) == expected
+
+    def test_message_handler_truncates_to_max_length(self):
+        big_content = "x" * 200
+        doc = Document(content=big_content, meta={"type": "file_content", "path": "main.py"})
+        result = message_handler([doc], max_length=50)
+        assert result.endswith("...(large file can't be fully displayed)")
+        assert len(result) == 50 + len("...(large file can't be fully displayed)")
+
+
+class TestSerializeHandlers:
+    @pytest.mark.parametrize(
+        "outputs_to_state,outputs_to_string,error_match",
+        [
+            (
+                {"documents": {"handler": "not_callable"}},
+                None,
+                "outputs_to_state\\[documents\\] is not a callable",
+            ),
+            (
+                None,
+                {"handler": "not_callable"},
+                "outputs_to_string is not a callable",
+            ),
+        ],
+    )
+    def test_serialize_handlers_raises_when_handler_not_callable(
+        self, outputs_to_state, outputs_to_string, error_match
+    ):
+        serialized: dict = {}
+        with pytest.raises(ValueError, match=error_match):
+            serialize_handlers(serialized, outputs_to_state, outputs_to_string)
