.github/workflows/ci_prepare.yml

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -118,45 +118,29 @@ jobs:
  
            uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5

            with:

              python-version: ${{ matrix.python-version }}

              cache: "poetry"

          - name: Install Poetry

            uses: snok/install-poetry@76e04a911780d5b312d89783f7b1cd627778900a # v1

            id: poetry

            with:

              version: latest

              virtualenvs-create: true

              installer-parallel: true

          - name: Ensure Poetry is available

          - name: Add Poetry to PATH

            run: |

              # Add Poetry to PATH

              echo "$HOME/.local/bin" >> $GITHUB_PATH

              echo "$HOME/.poetry/bin" >> $GITHUB_PATH

              echo "${{ github.workspace }}" >> $GITHUB_PATH

              # Check if Poetry is available

              if ! command -v poetry >/dev/null 2>&1 && [ ! -f "$HOME/.local/bin/poetry" ] && [ ! -f "$HOME/.poetry/bin/poetry" ]; then

                echo "Poetry not found, installing via official installer..."

                curl -sSL https://install.python-poetry.org | python3 -

                export PATH="$HOME/.local/bin:$PATH"

                echo "$HOME/.local/bin" >> $GITHUB_PATH

              fi

              # Verify Poetry is available

              poetry --version || $HOME/.local/bin/poetry --version || $HOME/.poetry/bin/poetry --version || exit 1

              echo "Poetry installation verified"

            shell: bash

          - name: Load cached Poetry dependencies

            uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4

            id: cache

            with:

              path: ~/.cache/pypoetry

              key: ${{ runner.os }}-poetry-${{ hashFiles('**/poetry.lock') }}

              restore-keys: |

                ${{ runner.os }}-poetry-

          - name: Install dependencies

            run: |

              poetry install --with test --no-root --no-interaction

.github/workflows/documentation.yml

-Original file line number
+Diff line change
@@ Expand Up / @@ -21,6 +21,9 @@ jobs: @@
           - name: Set up Python
             uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
+            with:
+              python-version: "3.11"
+              cache: "poetry"
           - name: Install Poetry
             uses: snok/install-poetry@76e04a911780d5b312d89783f7b1cd627778900a # v1
@@ Expand All / @@ -30,9 +33,7 @@ jobs: @@
               installer-parallel: true
           - name: Add Poetry to PATH
-            run: |
-              echo "$HOME/.local/bin" >> $GITHUB_PATH
-              echo "$HOME/.poetry/bin" >> $GITHUB_PATH
+            run: echo "$HOME/.local/bin" >> $GITHUB_PATH
           - name: Load cached Poetry dependencies
             uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4
@@ Expand Down @@

tests/test_asgi_registry.py

-Original file line number
+Diff line change
@@ -0,0 +1,139 @@
+    from unittest.mock import MagicMock, patch
+    import pytest
+    from temporal_boost.workers.asgi_registry import (
+        ASGIWorkerRegistry,
+        ASGIWorkerType,
+        asgi_worker_registry,
+        get_asgi_worker_class,
+    )
+    class MockASGIWorker:
+        pass
+    class TestASGIWorkerRegistry:
+        def test_init(self):
+            registry = ASGIWorkerRegistry()
+            assert registry._data == {}
+        def test_register(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker")
+            class TestWorker(MockASGIWorker):
+                pass
+            assert "test_worker" in registry._data
+            assert registry._data["test_worker"] == TestWorker
+        def test_register_with_packages(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker", packages=["os"])
+            class TestWorker(MockASGIWorker):
+                pass
+            assert "test_worker" in registry._data
+        def test_register_with_missing_package(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker", packages=["nonexistent_package_12345"])
+            class TestWorker(MockASGIWorker):
+                pass
+            assert "test_worker" not in registry._data
+        def test_get_existing(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker")
+            class TestWorker(MockASGIWorker):
+                pass
+            worker_class = registry.get("test_worker")
+            assert worker_class == TestWorker
+        def test_get_nonexistent(self):
+            registry = ASGIWorkerRegistry()
+            with pytest.raises(RuntimeError, match="is not available"):
+                registry.get("nonexistent")
+        def test_available_keys(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("worker1")
+            class Worker1(MockASGIWorker):
+                pass
+            @registry.register("worker2")
+            class Worker2(MockASGIWorker):
+                pass
+            keys = registry.available_keys()
+            assert "worker1" in keys
+            assert "worker2" in keys
+            assert len(keys) == 2
+    class TestASGIWorkerType:
+        def test_enum_values(self):
+            assert ASGIWorkerType.uvicorn.value == "uvicorn"
+            assert ASGIWorkerType.hypercorn.value == "hypercorn"
+            assert ASGIWorkerType.granian.value == "granian"
+            assert ASGIWorkerType.auto.value == "auto"
+    class TestGetASGIWorkerClass:
+        def test_get_uvicorn(self):
+            with patch.object(asgi_worker_registry, "get") as mock_get:
+                mock_get.return_value = MockASGIWorker
+                worker_class = get_asgi_worker_class(ASGIWorkerType.uvicorn)
+                mock_get.assert_called_once_with("uvicorn")
+                assert worker_class == MockASGIWorker
+        def test_get_hypercorn(self):
+            with patch.object(asgi_worker_registry, "get") as mock_get:
+                mock_get.return_value = MockASGIWorker
+                worker_class = get_asgi_worker_class(ASGIWorkerType.hypercorn)
+                mock_get.assert_called_once_with("hypercorn")
+                assert worker_class == MockASGIWorker
+        def test_get_granian(self):
+            with patch.object(asgi_worker_registry, "get") as mock_get:
+                mock_get.return_value = MockASGIWorker
+                worker_class = get_asgi_worker_class(ASGIWorkerType.granian)
+                mock_get.assert_called_once_with("granian")
+                assert worker_class == MockASGIWorker
+        def test_get_auto_with_uvicorn_available(self):
+            with patch.object(asgi_worker_registry, "available_keys") as mock_keys:
+                mock_keys.return_value = ["uvicorn", "hypercorn"]
+                with patch.object(asgi_worker_registry, "get") as mock_get:
+                    mock_get.return_value = MockASGIWorker
+                    worker_class = get_asgi_worker_class(ASGIWorkerType.auto)
+                    mock_get.assert_called_once_with("uvicorn")
+                    assert worker_class == MockASGIWorker
+        def test_get_auto_without_uvicorn(self):
+            with patch.object(asgi_worker_registry, "available_keys") as mock_keys:
+                mock_keys.return_value = ["hypercorn", "granian"]
+                with patch.object(asgi_worker_registry, "get") as mock_get:
+                    mock_get.return_value = MockASGIWorker
+                    worker_class = get_asgi_worker_class(ASGIWorkerType.auto)
+                    mock_get.assert_called_once_with("hypercorn")
+                    assert worker_class == MockASGIWorker
+        def test_get_auto_with_no_workers(self):
+            with patch.object(asgi_worker_registry, "available_keys") as mock_keys:
+                mock_keys.return_value = []
+                with pytest.raises(RuntimeError, match="No ASGI worker implementation is available"):
+                    get_asgi_worker_class(ASGIWorkerType.auto)

chore: streamline CI workflow for Poetry installation and path manage… #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

northpowered merged 3 commits into main from fix-tests-ci-jobs

Nov 1, 2025

-Original file line number
+Diff line change
@@ Expand Up / @@ -21,6 +21,9 @@ jobs: @@
           - name: Set up Python
             uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
+            with:
+              python-version: "3.11"
+              cache: "poetry"
           - name: Install Poetry
             uses: snok/install-poetry@76e04a911780d5b312d89783f7b1cd627778900a # v1
@@ Expand All / @@ -30,9 +33,7 @@ jobs: @@
               installer-parallel: true
           - name: Add Poetry to PATH
-            run: |
-              echo "$HOME/.local/bin" >> $GITHUB_PATH
-              echo "$HOME/.poetry/bin" >> $GITHUB_PATH
+            run: echo "$HOME/.local/bin" >> $GITHUB_PATH
           - name: Load cached Poetry dependencies
             uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4
@@ Expand Down @@

-Original file line number
+Diff line change
@@ -0,0 +1,139 @@
+    from unittest.mock import MagicMock, patch
+    import pytest
+    from temporal_boost.workers.asgi_registry import (
+        ASGIWorkerRegistry,
+        ASGIWorkerType,
+        asgi_worker_registry,
+        get_asgi_worker_class,
+    )
+    class MockASGIWorker:
+        pass
+    class TestASGIWorkerRegistry:
+        def test_init(self):
+            registry = ASGIWorkerRegistry()
+            assert registry._data == {}
+        def test_register(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker")
+            class TestWorker(MockASGIWorker):
+                pass
+            assert "test_worker" in registry._data
+            assert registry._data["test_worker"] == TestWorker
+        def test_register_with_packages(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker", packages=["os"])
+            class TestWorker(MockASGIWorker):
+                pass
+            assert "test_worker" in registry._data
+        def test_register_with_missing_package(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker", packages=["nonexistent_package_12345"])
+            class TestWorker(MockASGIWorker):
+                pass
+            assert "test_worker" not in registry._data
+        def test_get_existing(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("test_worker")
+            class TestWorker(MockASGIWorker):
+                pass
+            worker_class = registry.get("test_worker")
+            assert worker_class == TestWorker
+        def test_get_nonexistent(self):
+            registry = ASGIWorkerRegistry()
+            with pytest.raises(RuntimeError, match="is not available"):
+                registry.get("nonexistent")
+        def test_available_keys(self):
+            registry = ASGIWorkerRegistry()
+            @registry.register("worker1")
+            class Worker1(MockASGIWorker):
+                pass
+            @registry.register("worker2")
+            class Worker2(MockASGIWorker):
+                pass
+            keys = registry.available_keys()
+            assert "worker1" in keys
+            assert "worker2" in keys
+            assert len(keys) == 2
+    class TestASGIWorkerType:
+        def test_enum_values(self):
+            assert ASGIWorkerType.uvicorn.value == "uvicorn"
+            assert ASGIWorkerType.hypercorn.value == "hypercorn"
+            assert ASGIWorkerType.granian.value == "granian"
+            assert ASGIWorkerType.auto.value == "auto"
+    class TestGetASGIWorkerClass:
+        def test_get_uvicorn(self):
+            with patch.object(asgi_worker_registry, "get") as mock_get:
+                mock_get.return_value = MockASGIWorker
+                worker_class = get_asgi_worker_class(ASGIWorkerType.uvicorn)
+                mock_get.assert_called_once_with("uvicorn")
+                assert worker_class == MockASGIWorker
+        def test_get_hypercorn(self):
+            with patch.object(asgi_worker_registry, "get") as mock_get:
+                mock_get.return_value = MockASGIWorker
+                worker_class = get_asgi_worker_class(ASGIWorkerType.hypercorn)
+                mock_get.assert_called_once_with("hypercorn")
+                assert worker_class == MockASGIWorker
+        def test_get_granian(self):
+            with patch.object(asgi_worker_registry, "get") as mock_get:
+                mock_get.return_value = MockASGIWorker
+                worker_class = get_asgi_worker_class(ASGIWorkerType.granian)
+                mock_get.assert_called_once_with("granian")
+                assert worker_class == MockASGIWorker
+        def test_get_auto_with_uvicorn_available(self):
+            with patch.object(asgi_worker_registry, "available_keys") as mock_keys:
+                mock_keys.return_value = ["uvicorn", "hypercorn"]
+                with patch.object(asgi_worker_registry, "get") as mock_get:
+                    mock_get.return_value = MockASGIWorker
+                    worker_class = get_asgi_worker_class(ASGIWorkerType.auto)
+                    mock_get.assert_called_once_with("uvicorn")
+                    assert worker_class == MockASGIWorker
+        def test_get_auto_without_uvicorn(self):
+            with patch.object(asgi_worker_registry, "available_keys") as mock_keys:
+                mock_keys.return_value = ["hypercorn", "granian"]
+                with patch.object(asgi_worker_registry, "get") as mock_get:
+                    mock_get.return_value = MockASGIWorker
+                    worker_class = get_asgi_worker_class(ASGIWorkerType.auto)
+                    mock_get.assert_called_once_with("hypercorn")
+                    assert worker_class == MockASGIWorker
+        def test_get_auto_with_no_workers(self):
+            with patch.object(asgi_worker_registry, "available_keys") as mock_keys:
+                mock_keys.return_value = []
+                with pytest.raises(RuntimeError, match="No ASGI worker implementation is available"):
+                    get_asgi_worker_class(ASGIWorkerType.auto)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

chore: streamline CI workflow for Poetry installation and path manage… #39

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!