body
"): + super().__init__(custom_labels=[], join_input_with_list=False) + self._labels = labels + self._body = body + + def format_labels(self, labels: list[str]): + return self._labels + + def format_body(self, body: str) -> str: + return self._body + + +def _make_render_result(text: str) -> sulguk.RenderResult: + # Minimal object compatible with what TelegramGateway expects + return sulguk.RenderResult( + text=text, + entities=[], + ) + + +@patch("notifier.application.interactors.sulguk.transform_html") +def test_send_issue_uses_default_template(mock_transform_html): + github = _GithubStub() + telegram = _TelegramStub() + render_service = _RenderServiceStub() + + mock_transform_html.return_value = _make_render_result("rendered") + + interactor = SendIssue( + template="", + github=github, + telegram=telegram, + render_service=render_service, + ) + + interactor.handler() + + # Verify transform_html was called with correct message + mock_transform_html.assert_called_once() + call_args = mock_transform_html.call_args + captured_message = call_args[0][0] + + assert ISSUE_TEMPLATE.format( + id=github.get_issue().id, + user=github.get_issue().user, + title=github.get_issue().title, + labels=render_service._labels, + url=github.get_issue().url, + body=render_service._body, + repository=github.get_issue().repository, + promo="sent via relator", + ) == captured_message + # ensure telegram was called with rendered result + assert len(telegram.sent) == 1 + assert telegram.sent[0].text == "rendered" + + +@patch("notifier.application.interactors.sulguk.transform_html") +def test_send_issue_truncates_long_messages(mock_transform_html): + github = _GithubStub() + telegram = _TelegramStub() + # create very long body + long_body = "x" * (TG_MESSAGE_LIMIT + 10) + render_service = _RenderServiceStub(body=long_body) + + # render_result.text should reflect the original message length + def _side_effect(message: str, base_url: str): + return _make_render_result(text=message) + + mock_transform_html.side_effect = _side_effect + + interactor = SendIssue( + template=ISSUE_TEMPLATE, + github=github, + telegram=telegram, + render_service=render_service, + ) + + interactor.handler() + + # when message length exceeds limit, we do not send via telegram + assert telegram.sent == [] + # second transform_html call should be without description + assert mock_transform_html.call_count == 2 + calls = [call[0][0] for call in mock_transform_html.call_args_list] + assert "" in calls[1] + + +@patch("notifier.application.interactors.sulguk.transform_html") +def test_send_pr_uses_default_template(mock_transform_html): + github = _GithubStub() + telegram = _TelegramStub() + render_service = _RenderServiceStub() + + mock_transform_html.return_value = _make_render_result("rendered") + + interactor = SendPR( + template="", + github=github, + telegram=telegram, + render_service=render_service, + ) + + interactor.handler() + + pr = github.get_pull_request() + expected_message = PR_TEMPLATE.format( + id=pr.id, + user=pr.user, + title=pr.title, + labels=render_service._labels, + url=pr.url, + body=render_service._body, + repository=pr.repository, + additions=pr.additions, + deletions=pr.deletions, + head_ref=pr.head_ref, + base_ref=pr.base_ref, + promo="sent via relator", + ) + call_args = mock_transform_html.call_args + captured_message = call_args[0][0] + assert captured_message == expected_message + assert len(telegram.sent) == 1 + assert telegram.sent[0].text == "rendered" + + +@patch("notifier.application.interactors.sulguk.transform_html") +def test_send_pr_truncates_long_messages(mock_transform_html): + github = _GithubStub() + telegram = _TelegramStub() + long_body = "x" * (TG_MESSAGE_LIMIT + 10) + render_service = _RenderServiceStub(body=long_body) + + def _side_effect(message: str, base_url: str): + return _make_render_result(text=message) + + mock_transform_html.side_effect = _side_effect + + interactor = SendPR( + template=PR_TEMPLATE, + github=github, + telegram=telegram, + render_service=render_service, + ) + + interactor.handler() + + assert telegram.sent == [] + assert mock_transform_html.call_count == 2 + calls = [call[0][0] for call in mock_transform_html.call_args_list] + assert "" in calls[1] + + diff --git a/tests/test_application_services.py b/tests/test_application_services.py new file mode 100644 index 0000000..9144377 --- /dev/null +++ b/tests/test_application_services.py @@ -0,0 +1,82 @@ +from unittest.mock import patch + +import bs4 +import pytest +import sulguk + +from notifier.application.services import RenderService + + +def test_format_body_empty_returns_input(): + service = RenderService(custom_labels=[], join_input_with_list=False) + + assert service.format_body("") == "" + + +def test_format_body_strips_blob_wrappers(): + service = RenderService(custom_labels=[], join_input_with_list=False) + + html = """ +code block
real content
+broken
") == "" + + +@pytest.mark.parametrize( + ("labels", "custom", "expected"), + [ + (["bug"], [], "#bugbody
", + } + response.raise_for_status.return_value = None + mock_get.return_value = response + + gw = GithubGateway(token="TOKEN", event_url="https://api.github.com/issue") + issue = gw.get_issue() + + mock_get.assert_called_once_with( + "https://api.github.com/issue", + headers={ + "Accept": "application/vnd.github.v3.html+json", + "X-GitHub-Api-Version": "2022-11-28", + "Authorization": "Bearer TOKEN", + }, + timeout=30, + ) + assert issue.title == "Issue" + assert issue.labels == ["bug"] + assert issue.user == "user" + assert issue.url == "https://github.com/owner/repo/issues/1" + assert issue.body == "body
" + + +@patch("notifier.infrastructure.github_gateway.requests.get") +def test_github_gateway_get_pull_request_builds_entity(mock_get): + response = Mock() + response.json.return_value = { + "number": 2, + "title": "PR", + "labels": [{"name": "enhancement"}], + "html_url": "https://github.com/owner/repo/pull/2", + "user": {"login": "user"}, + "body_html": "body
", + "additions": 5, + "deletions": 1, + "head": {"label": "feature", "ref": "feature"}, + "base": {"ref": "main", "repo": {"full_name": "owner/repo"}}, + } + response.raise_for_status.return_value = None + mock_get.return_value = response + + gw = GithubGateway(token="TOKEN", event_url="https://api.github.com/pr") + pr = gw.get_pull_request() + + assert pr.id == 2 + assert pr.additions == 5 + assert pr.deletions == 1 + assert pr.repository == "owner/repo" + + +@patch("notifier.infrastructure.telegram_gateway.requests.post") +def test_telegram_gateway_send_message_success(mock_post, capsys): + result = sulguk.RenderResult(text="hi", entities=[{"offset": 0, "length": 2, "type": "bold"}]) + + response = Mock() + response.raise_for_status.return_value = None + response.json.return_value = {"ok": True} + mock_post.return_value = response + + gw = TelegramGateway( + chat_id="123", + bot_token="TOKEN", + attempt_count=1, + message_thread_id=None, + ) + + gw.send_message(result) + + mock_post.assert_called_once() + call_args, call_kwargs = mock_post.call_args + assert "sendMessage" in call_args[0] + assert call_kwargs["json"]["text"] == "hi" + # language field should be removed if exists + for e in call_kwargs["json"]["entities"]: + assert "language" not in e + + captured = capsys.readouterr() + # Should print response json on success + assert "ok" in captured.out + + +@patch("notifier.infrastructure.telegram_gateway.time.sleep") +@patch("notifier.infrastructure.telegram_gateway.requests.post") +def test_telegram_gateway_send_message_retries_on_error(mock_post, mock_sleep, capsys): + result = sulguk.RenderResult(text="hi", entities=[]) + + response = Mock() + response.raise_for_status.side_effect = requests.exceptions.HTTPError("fail") + response.content = b"error" + mock_post.return_value = response + + gw = TelegramGateway( + chat_id="123", + bot_token="TOKEN", + attempt_count=3, + message_thread_id="456", + ) + + gw.send_message(result) + + # ensure we retried attempt_count times + assert mock_post.call_count == 3 + captured = capsys.readouterr() + assert "error" in captured.err + + diff --git a/tests/test_main_module.py b/tests/test_main_module.py new file mode 100644 index 0000000..34b8e21 --- /dev/null +++ b/tests/test_main_module.py @@ -0,0 +1,91 @@ +import os +from unittest.mock import patch + +import pytest + +from notifier.__main__ import get_interactor +from notifier.application.interactors import SendIssue, SendPR + + +@pytest.mark.parametrize( + ("url", "expected"), + [ + ("https://api.github.com/repos/owner/repo/issues/1", SendIssue), + ("https://api.github.com/repos/owner/repo/pulls/2", SendPR), + ], +) +def test_get_interactor_detects_type(url, expected): + """get_interactor should recognize API URLs for issues and PRs.""" + assert get_interactor(url) is expected + + +def test_get_interactor_raises_on_unknown(): + with pytest.raises(ValueError): + get_interactor("https://github.com/owner/repo/unknown/1") + + +@patch.dict( + os.environ, + { + "TELEGRAM_CHAT_ID": "123", + "TELEGRAM_BOT_TOKEN": "token", + "ATTEMPT_COUNT": "3", + "EVENT_URL": "https://api.github.com/repos/o/r/issues/1", + "GITHUB_TOKEN": "gtoken", + "CUSTOM_LABELS": "custom1,custom2", + "JOIN_INPUT_WITH_LIST": "1", + "HTML_TEMPLATE": "", + }, +) +@patch("notifier.__main__.GithubGateway") +@patch("notifier.__main__.TelegramGateway") +@patch("notifier.__main__.RenderService") +@patch("notifier.__main__.get_interactor") +def test_main_module_env_parsing( + mock_get_interactor, mock_render_service, mock_telegram, mock_github +): + """ + Smoke-test the main module wiring by simulating environment and + checking that the selected interactor's handler is invoked. + """ + + # prepare fake gateways and interactor + class _FakeGithub: + def __init__(self, *_, **__): + pass + + class _FakeTelegram: + def __init__(self, *_, **__): + pass + + class _FakeRenderService: + def __init__(self, *_, **__): + pass + + class _FakeInteractor: + def __init__(self, *_, **__): + self.called = False + + def handler(self): + self.called = True + + mock_github.return_value = _FakeGithub() + mock_telegram.return_value = _FakeTelegram() + mock_render_service.return_value = _FakeRenderService() + mock_get_interactor.return_value = _FakeInteractor + + # Import and execute main logic + import importlib + + main_mod = importlib.import_module("notifier.__main__") + + fake_interactor = main_mod.get_interactor(os.environ["EVENT_URL"])( + template="", + github=_FakeGithub(), + telegram=_FakeTelegram(), + render_service=_FakeRenderService(), + ) + # Ensure handler exists and can be called without errors + fake_interactor.handler() + +