add integration tests

Author: snopoke

.github/workflows/integration_tests.yml

@@ -0,0 +1,29 @@
+name: Integration Tests
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  integration-tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.11"
+      - name: Install Poetry
+        uses: Gr1N/setup-poetry@v8
+      - uses: actions/cache@v3
+        with:
+          path: ~/.cache/pypoetry/virtualenvs
+          key: ${{ runner.os }}-poetry-${{ hashFiles('poetry.lock') }}
+      - name: Checks
+        run: |
+          poetry --version
+          poetry check --no-interaction
+      - name: Install project
+        run: poetry install --no-interaction
+      - name: Run integration tests
+        run: poetry run pytest integration_tests -v

.gitignore

@@ -11,6 +11,7 @@ dist/
 .env
 .venv
 .envrc
+.env.integration
 
 # mypy
 .mypy_cache/

integration_tests/__init__.py

@@ -0,0 +1,33 @@
+import os
+from pathlib import Path
+
+import pytest
+
+import taskbadger as badger
+
+
+def _load_config():
+    path = Path(__file__).parent.parent / ".env.integration"
+    if not path.exists():
+        return
+    with path.open() as f:
+        for line in f:
+            if line.startswith("#") or not line.strip():
+                continue
+            key, value = line.strip().split("=", 1)
+            os.environ[key] = value
+
+
+_load_config()
+ORG = os.environ.get("TASKBADGER_ORG", "")
+PROJECT = os.environ.get("TASKBADGER_PROJECT", "")
+API_KEY = os.environ.get("TASKBADGER_API_KEY", "")
+
+if not ORG or not PROJECT or not API_KEY:
+    pytest.fail("Integration test config missing", pytrace=False)
+else:
+    badger.init(
+        os.environ.get("TASKBADGER_ORG", ""),
+        os.environ.get("TASKBADGER_PROJECT", ""),
+        os.environ.get("TASKBADGER_API_KEY", ""),
+    )

integration_tests/tasks.py

@@ -0,0 +1,10 @@
+from celery import shared_task
+
+import taskbadger.celery
+
+
+@shared_task(bind=True, base=taskbadger.celery.Task)
+def add(self, x, y):
+    assert self.taskbadger_task is not None, "missing task on self"
+    self.taskbadger_task.update(value=100, data={"result": x + y})
+    return x + y

integration_tests/test_basics.py

@@ -0,0 +1,16 @@
+from datetime import datetime
+
+import taskbadger as badger
+from taskbadger import StatusEnum
+
+
+def test_basics():
+    data = {"now": datetime.utcnow().isoformat()}
+    task = badger.create_task("test basics", data=data)
+    task.success(100)
+    assert task.status == StatusEnum.SUCCESS
+
+    fresh = badger.get_task(task.id)
+    assert fresh.status == StatusEnum.SUCCESS
+    assert fresh.value == 100
+    assert fresh.data == data

integration_tests/test_celery.py

@@ -0,0 +1,26 @@
+import random
+
+import pytest
+
+from taskbadger import StatusEnum
+
+from .tasks import add
+
+
+@pytest.fixture(scope="session", autouse=True)
+def celery_includes():
+    return [
+        "integration_tests.tasks",
+    ]
+
+
+def test_celery(celery_worker):
+    a, b = random.randint(1, 1000), random.randint(1, 1000)
+    result = add.delay(a, b)
+    assert result.get(timeout=10, propagate=True) == a + b
+
+    tb_task = result.get_taskbadger_task()
+    assert tb_task is not None
+    assert tb_task.status == StatusEnum.SUCCESS
+    assert tb_task.value == 100
+    assert tb_task.data == {"result": a + b}

pyproject.toml

@@ -77,3 +77,7 @@ exclude = '''
 [tool.isort]
 line_length = 120
 profile = "black"
+
+[tool.pytest.ini_options]
+# don't run integration tests unless specifically requested
+norecursedirs = ".* integration_tests"