feat: implement check update hook

slack_cli_hooks/error/__init__.py

@@ -1,2 +1,6 @@
 class CliError(Exception):
     """General class for cli error"""
+
+
+class PypiError(Exception):
+    """General class for PyPI package info retrieval error"""

slack_cli_hooks/hooks/check_update.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python
+import json
+from http.client import HTTPResponse
+from types import ModuleType
+from typing import Any, Dict, List, Optional
+from urllib import request
+
+import slack_bolt
+import slack_sdk
+from pkg_resources import parse_version as Version
+
+import slack_cli_hooks.version
+from slack_cli_hooks.error import PypiError
+from slack_cli_hooks.protocol import Protocol, build_protocol
+
+PROTOCOL: Protocol
+
+DEPENDENCIES: List[ModuleType] = [slack_cli_hooks, slack_bolt, slack_sdk]
+
+
+def parse_major(v: Version) -> int:
+    """The first item of :attr:`release` or ``0`` if unavailable.
+
+    >>> parse_major(Version("1.2.3"))
+    1
+    """
+    # This implementation comes directly from the Version implementation since it is not supported in 3.6
+    # source: https://github.com/pypa/packaging/blob/main/src/packaging/version.py
+    return v._version.release[0] if len(v._version) >= 1 else 0  # type: ignore
+
+
+class Release:
+    def __init__(
+        self,
+        name: str,
+        current: Optional[Version] = None,
+        latest: Optional[Version] = None,
+        message: Optional[str] = None,
+        url: Optional[str] = None,
+        error: Optional[Dict[str, str]] = None,
+    ):
+        self.name = name
+        if current and latest:
+            self.current = current.base_version
+            self.latest = latest.base_version
+            self.update = current < latest
+            self.breaking = (parse_major(current) - parse_major(latest)) != 0
+        if error:
+            self.error = error
+        if message:
+            self.message = message
+        if url:
+            self.url = url
+
+
+def pypi_get(project: str, headers={"Accept": "application/json"}) -> HTTPResponse:
+    # Based on https://warehouse.pypa.io/api-reference/json.html
+    url = f"https://pypi.org/pypi/{project}/json"
+    pypi_request = request.Request(method="GET", url=url, headers=headers)
+    return request.urlopen(pypi_request)
+
+
+def pypi_get_json(project: str) -> Dict[str, Any]:
+    pypi_response = pypi_get(project)
+    charset = pypi_response.headers.get_content_charset() or "utf-8"
+    raw_body = pypi_response.read().decode(charset)
+    if pypi_response.status > 200:
+        PROTOCOL.debug(f"Received status {pypi_response.status} from {pypi_response.url}")
+        PROTOCOL.debug(f"Headers {dict(pypi_response.getheaders())}")
+        PROTOCOL.debug(f"Body {raw_body}")
+        raise PypiError(f"Received status {pypi_response.status} from {pypi_response.url}")
+    return json.loads(raw_body)
+
+
+def extract_latest_version(payload: Dict[str, Any]) -> str:
+    if "info" not in payload:
+        raise PypiError("Missing `info` field in pypi payload")
+    if "version" not in payload["info"]:
+        raise PypiError("Missing `version` field in pypi payload['info']")
+    return payload["info"]["version"]
+
+
+def build_release(dependency: ModuleType) -> Release:
+    name = dependency.__name__
+    try:
+        pypi_json_payload = pypi_get_json(name)
+        return Release(
+            name=name,
+            current=Version(dependency.version.__version__),
+            latest=Version(extract_latest_version(pypi_json_payload)),
+        )
+    except PypiError as e:
+        return Release(name=name, error={"message": str(e)})
+
+
+def build_output(dependencies: List[ModuleType] = DEPENDENCIES) -> Dict[str, Any]:
+    output = {"name": "Slack Bolt", "url": "https://api.slack.com/automation/changelog", "releases": []}
+    errors = []
+
+    for dep in dependencies:
+        release = build_release(dep)
+        output["releases"].append(vars(release))
+
+        if hasattr(release, "error"):
+            errors.append(release.name)
+
+    if errors:
+        output["error"] = {"message": f"An error occurred fetching updates for the following packages: {', '.join(errors)}"}
+    return output
+
+
+if __name__ == "__main__":
+    PROTOCOL = build_protocol()
+    PROTOCOL.respond(json.dumps(build_output()))

slack_cli_hooks/hooks/get_hooks.py

@@ -1,6 +1,11 @@
 #!/usr/bin/env python
 import json
-from slack_cli_hooks.protocol import Protocol, MessageBoundaryProtocol, DefaultProtocol, build_protocol
+from slack_cli_hooks.protocol import (
+    Protocol,
+    MessageBoundaryProtocol,
+    DefaultProtocol,
+    build_protocol,
+)
 
 PROTOCOL: Protocol
 EXEC = "python3"
@@ -10,6 +15,7 @@
     "hooks": {
         "get-manifest": f"{EXEC} -m slack_cli_hooks.hooks.get_manifest",
         "start": f"{EXEC} -X dev -m slack_cli_hooks.hooks.start",
+        "check-update": f"{EXEC} -m slack_cli_hooks.hooks.check_update",
     },
     "config": {
         "watch": {"filter-regex": "(^manifest\\.json$)", "paths": ["."]},

slack_cli_hooks/hooks/utils/managed_os_env_vars.py

@@ -10,9 +10,7 @@ def __init__(self, protocol: Protocol) -> None:
 
     def set_if_absent(self, os_env_var: str, value: str) -> None:
         if os_env_var in os.environ:
-            self._protocol.info(
-                f"{os_env_var} environment variable detected in session, using it over the provided one!"
-            )
+            self._protocol.info(f"{os_env_var} environment variable detected in session, using it over the provided one!")
             return
         self._os_env_vars.append(os_env_var)
         os.environ[os_env_var] = value

tests/scenario_test/test_check_update.py

@@ -0,0 +1,47 @@
+from unittest.mock import patch
+from urllib import request
+
+from slack_cli_hooks.hooks import check_update
+from slack_cli_hooks.hooks.check_update import build_output
+from slack_cli_hooks.protocol.default_protocol import DefaultProtocol
+from tests.utils import build_fake_dependency, build_fake_pypi_urlopen
+
+
+class TestGetManifest:
+    def setup_method(self):
+        check_update.PROTOCOL = DefaultProtocol()
+
+    def test_build_output(self):
+        test_project = "test_proj"
+        fake_pypi_urlopen = build_fake_pypi_urlopen(status=200, body={"info": {"version": "0.0.1"}})
+        test_dependency = build_fake_dependency(test_project, "0.0.0")
+
+        with patch.object(request, "urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = fake_pypi_urlopen
+            actual = build_output([test_dependency])
+
+        assert actual["name"] == "Slack Bolt"
+        assert len(actual["releases"]) == 1
+        assert actual["releases"][0]["name"] == test_project
+        assert actual["releases"][0]["current"] == "0.0.0"
+        assert actual["releases"][0]["latest"] == "0.0.1"
+        assert actual["releases"][0]["update"] is True
+        assert actual["releases"][0]["breaking"] is False
+        assert "error" not in actual["releases"][0]
+
+    def test_build_output_error(self):
+        test_project = "test_proj"
+        fake_pypi_urlopen = build_fake_pypi_urlopen(status=200, body={"info": {}})
+        test_dependency = build_fake_dependency(test_project, "0.0.0")
+
+        with patch.object(request, "urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = fake_pypi_urlopen
+            actual = build_output([test_dependency])
+
+        assert actual["name"] == "Slack Bolt"
+        assert len(actual["releases"]) == 1
+        assert actual["releases"][0]["name"] == test_project
+        assert "error" in actual["releases"][0]
+        assert "message" in actual["releases"][0]["error"]
+        assert "error" in actual
+        assert "message" in actual["error"]

tests/slack_cli_hooks/hooks/test_check_update.py

@@ -0,0 +1,111 @@
+from unittest.mock import patch
+from urllib import request
+
+import pytest
+
+from slack_cli_hooks.error import PypiError
+from slack_cli_hooks.hooks import check_update
+from slack_cli_hooks.hooks.check_update import (
+    build_output,
+    build_release,
+    extract_latest_version,
+    pypi_get,
+    pypi_get_json,
+)
+from slack_cli_hooks.protocol.default_protocol import DefaultProtocol
+from tests.utils import build_fake_dependency, build_fake_pypi_urlopen
+
+
+class TestGetManifest:
+    def setup_method(self):
+        check_update.PROTOCOL = DefaultProtocol()
+
+    def test_pypi_get(self):
+        test_project = "test_proj"
+        fake_pypi_urlopen = build_fake_pypi_urlopen()
+
+        with patch.object(request, "urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = fake_pypi_urlopen
+            response = pypi_get(test_project)
+
+        assert response.url == f"https://pypi.org/pypi/{test_project}/json"
+        assert response.status == 200
+        assert response.read().decode("utf-8") == "{}"
+
+    def test_pypi_get_json(self):
+        project = "my_test_project"
+        fake_pypi_urlopen = build_fake_pypi_urlopen(body={"info": {}, "releases": {}})
+
+        with patch.object(request, "urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = fake_pypi_urlopen
+            json_response = pypi_get_json(project)
+
+        assert json_response == {"info": {}, "releases": {}}
+
+    def test_pypi_get_json_fail(self):
+        project = "my_test_project"
+        fake_pypi_urlopen = build_fake_pypi_urlopen(status=300)
+
+        with patch.object(request, "urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = fake_pypi_urlopen
+            with pytest.raises(PypiError) as e:
+                pypi_get_json(project)
+
+            assert "300" in str(e)
+            assert f"https://pypi.org/pypi/{project}/json" in str(e)
+
+    def test_extract_latest_version(self):
+        test_payload = {"info": {"version": "0.0.0"}}
+        actual = extract_latest_version(test_payload)
+        assert actual == "0.0.0"
+
+    def test_extract_latest_version_missing_info(self):
+        test_payload = {}
+        with pytest.raises(PypiError) as e:
+            extract_latest_version(test_payload)
+            assert "info" in str(e)
+
+    def test_extract_latest_version_missing_version(self):
+        test_payload = {"info": {}}
+        with pytest.raises(PypiError) as e:
+            extract_latest_version(test_payload)
+            assert "version" in str(e)
+            assert "payload['info']" in str(e)
+
+    def test_build_release(self):
+        test_project = "test-dependency"
+        test_dependency = build_fake_dependency(test_project, "0.0.0")
+
+        with patch.object(check_update, pypi_get_json.__name__) as mock_pypi_get_json:
+            mock_pypi_get_json.return_value = {"info": {"version": "0.0.1"}}
+            actual = build_release(test_dependency)
+
+        assert vars(actual) == {
+            "name": test_project,
+            "current": "0.0.0",
+            "latest": "0.0.1",
+            "update": True,
+            "breaking": False,
+        }
+
+    def test_build_release_error(self):
+        test_project = "test-dependency"
+        test_dependency = build_fake_dependency(test_project, "0.0.0")
+
+        with patch.object(check_update, pypi_get_json.__name__) as mock_pypi_get_json:
+            mock_pypi_get_json.return_value = {}
+            actual = build_release(test_dependency)
+
+        assert vars(actual) == {
+            "name": test_project,
+            "error": {"message": "Missing `info` field in pypi payload"},
+        }
+
+    def test_build_output(self):
+        actual = build_output([])
+
+        assert actual == {
+            "name": "Slack Bolt",
+            "url": "https://api.slack.com/automation/changelog",
+            "releases": [],
+        }

tests/slack_cli_hooks/hooks/test_get_hooks.py

@@ -8,6 +8,7 @@ def test_hooks_payload(self):
 
         assert "slack_cli_hooks.hooks.get_manifest" in hooks["get-manifest"]
         assert "slack_cli_hooks.hooks.start" in hooks["start"]
+        assert "slack_cli_hooks.hooks.check_update" in hooks["check-update"]
 
     def test_hooks_payload_config(self):
         config = hooks_payload["config"]

tests/utils.py

@@ -1,4 +1,9 @@
+import json
 import os
+from http.client import HTTPMessage, HTTPResponse
+from typing import Callable, Union
+from unittest.mock import MagicMock
+from urllib.request import Request
 
 
 def remove_os_env_temporarily() -> dict:
@@ -9,3 +14,25 @@ def remove_os_env_temporarily() -> dict:
 
 def restore_os_env(old_env: dict) -> None:
     os.environ.update(old_env)
+
+
+def build_fake_pypi_urlopen(status: int = 200, headers=HTTPMessage(), body={}) -> Callable[..., HTTPResponse]:
+    headers.add_header("Content-Type", 'application/json; charset="UTF-8"')
+
+    mock_resp = HTTPResponse(MagicMock())
+    mock_resp.headers = headers
+    mock_resp.status = status
+    mock_resp.read = MagicMock(return_value=json.dumps(body).encode("UTF-8"))
+
+    def fake_urlopen(url: Union[str, Request]):
+        mock_resp.url = url.full_url if isinstance(url, Request) else url
+        return mock_resp
+
+    return fake_urlopen
+
+
+def build_fake_dependency(name: str, version: str):
+    fake_dependency = MagicMock()
+    fake_dependency.version.__version__ = version
+    fake_dependency.__name__ = name
+    return fake_dependency