Test coverage

.coveragerc

@@ -0,0 +1,10 @@
+[run]
+source = kegbot/pycore
+omit =
+    */*_test.py
+
+[report]
+exclude_lines =
+    pragma: no cover
+    raise NotImplementedError
+    if __name__ == .__main__.:

.github/workflows/pybuild.yml

@@ -12,21 +12,13 @@ jobs:
   py_build_and_test:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
-      - name: Setup Python
-        uses: actions/setup-python@v1
-        with:
-          python-version: 3.7
+      # Build the Dockerfile's `test` stage and run it. The suite mocks all I/O
+      # (Redis, the Kegbot API, threads), so no services are needed. The image's
+      # default CMD runs `coverage run -m pytest && coverage report --fail-under=80`.
+      - name: Build test image
+        run: docker build --target test -t kegbot/pycore:test .
 
-      - name: Install dependencies
-        run: |
-          pip install pipenv docker-compose
-          pipenv install --deploy --dev
-
-      - name: Run mysql & redis
-        run: |
-          docker-compose -f testdata/test-docker-compose.yml up -d
-
-      - name: pytest
-        run: KEGBOT_DATABASE_URL=mysql://root:changeme@127.0.0.1:3306/kegbot_dev pipenv run pytest
+      - name: Run tests with coverage
+        run: docker run --rm kegbot/pycore:test

.gitignore

@@ -7,3 +7,7 @@ distribute-*
 *egg-info
 dist/
 docs/build/
+
+### test artifacts
+.coverage
+.pytest_cache/

CLAUDE.md

@@ -0,0 +1,98 @@
+# kegbot-pycore — Claude Code context
+
+## What this is
+Python daemon that bridges a Kegboard Arduino device to the Kegbot server via Redis.
+Two long-running processes built from this repo:
+- **kegboard daemon** (`bin/kegboard_daemon.py`) — reads serial messages from the Arduino and publishes flow/temperature events to Redis
+- **pycore** (`bin/kegbot_core.py`) — consumes those events and drives the kegbot server API
+
+## Development workflow
+Code is written on the **MacBook** (`/Users/frodelangelo/src/kegbot-pycore`), then built and deployed on the **Pi** (`frode@kegberry`).
+
+```
+# 1. Make changes locally, commit, push
+git add <files> && git commit -m "..." && git push
+
+# 2. SSH to Pi, pull, build
+ssh kegberry "cd ~/src/kegbot-pycore && git pull && docker build -t kegbot/pycore:latest ."
+
+# 3. Deploy (docker-compose lives in ~/kegberry on the Pi)
+ssh kegberry "cd ~/kegberry && docker compose up -d kegboard"
+# or to restart both services:
+ssh kegberry "cd ~/kegberry && docker compose up -d kegboard pycore"
+
+# 4. Check logs
+ssh kegberry "docker logs kegberry-kegboard-1 --tail 50"
+ssh kegberry "docker logs kegberry-pycore-1 --tail 50"
+```
+
+## Pi directory layout
+| Path | Purpose |
+|------|---------|
+| `~/src/kegbot-pycore` | this repo |
+| `~/src/kegbot-server` | kegbot Django server |
+| `~/src/kegboard` | Arduino firmware + kegboard Python library |
+| `~/kegberry/` | docker-compose deployment (docker-compose.yml, nginx.conf, data/) |
+
+## Docker containers (docker-compose project: kegberry)
+| Container | Image | Role |
+|-----------|-------|------|
+| `kegberry-kegboard-1` | `kegbot/pycore:latest` | kegboard serial daemon |
+| `kegberry-pycore-1` | `kegbot/pycore:latest` | pycore event processor |
+| `kegberry-kegnet-listener-1` | `ghcr.io/flangelo/kegbot-server:latest` | kegnet Redis listener |
+| `kegberry-kegbot-1` | `ghcr.io/flangelo/kegbot-server:latest` | Django app |
+| `kegberry-workers-1` | `ghcr.io/flangelo/kegbot-server:latest` | RQ background workers |
+| `kegberry-nginx-1` | `nginx:alpine` | reverse proxy (port 8000) |
+| `kegberry-redis-1` | `redis:7.2` | message bus + task queue |
+| `kegberry-mysql-1` | `mariadb:10.11` | database |
+
+## Running tests
+Tests are plain `unittest.TestCase` classes in `*_test.py` files alongside the
+code, run via pytest. They mock all I/O (Redis, the Kegbot API, threads), so no
+services are required. Config lives in `pytest.ini` and `.coveragerc`.
+
+**Recommended:** build and run the Dockerfile's `test` stage. This runs the
+suite against the same venv/deps as the production image (`python:3.11-alpine`)
+and is exactly what CI (`.github/workflows/pybuild.yml`) does:
+```bash
+docker build --target test -t kegbot/pycore:test .
+docker run --rm kegbot/pycore:test
+```
+The image's default `CMD` runs `coverage run -m pytest && coverage report -m
+--fail-under=80`. Scope it down by overriding the command, e.g.:
+```bash
+docker run --rm kegbot/pycore:test pytest kegbot/pycore/manager_test.py
+```
+
+Quicker iteration (mounts the working tree into a slim image so edits don't
+require a rebuild):
+```bash
+docker run --rm -v "$(pwd)":/app -w /app python:3.11-slim bash -c '
+  pip install -q "pipenv<2024" && pipenv requirements > /tmp/req.txt
+  pip install -q -r /tmp/req.txt pytest coverage && pip install -q -e .
+  coverage run -m pytest && coverage report -m'
+```
+
+**Gotcha (mount workflow only):** mount/run with an explicit absolute path. If
+the working directory drifts to the kegbot-server repo, pytest picks up *its*
+`setup.cfg` addopts (`-p pykeg.test.plugin`), which needs Django and aborts
+collection.
+
+## Known build constraints
+- **Base image must be `python:3.11-alpine`** — Python 3.12 removed the `imp` module, which the `future` package (and other kegbot deps) still use. 3.11 retains it; 3.11 is supported until 2027.
+- **Pin `pipenv<2024`** — pipenv 2024+ rejects `python_version = "3"` (the spec in Pipfile/Pipfile.lock) as ambiguous in `--deploy` mode. Older pipenv accepts it.
+
+## Known runtime issue (fixed)
+`kegboard_daemon.py` used to crash with `ValueError: Bad length, must be exactly 4 bytes` during high-frequency flow pulse bursts (serial framing corruption). Fixed by catching `ValueError` in `service_devices()` — the daemon now logs a warning and continues rather than aborting.
+
+## Useful debugging
+```bash
+# Watch kegboard live
+ssh kegberry "docker logs -f kegberry-kegboard-1"
+
+# Check all container health
+ssh kegberry "docker ps"
+
+# Rebuild without cache (if packages seem stale)
+ssh kegberry "cd ~/src/kegbot-pycore && docker build --no-cache -t kegbot/pycore:latest ."
+```

Dockerfile

@@ -1,35 +1,61 @@
-FROM python:3-alpine
+# Stage 1: install dependencies and the package into a venv
+FROM python:3.11-alpine AS builder
 
-RUN mkdir /app
 WORKDIR /app
+RUN pip install --no-cache-dir "pipenv<2024"
 
-ENV SHELL=/bin/sh \
-   PIP_NO_CACHE_DIR=1 \
-   KEGBOT_IN_DOCKER=True \
-   KEGBOT_ENV=debug
+COPY Pipfile Pipfile.lock ./
+RUN pipenv requirements > /tmp/requirements.txt
 
-RUN apk update && \
-    apk add --no-cache \
-      bash \
-      curl && \
-   pip install pipenv
+RUN python -m venv /venv && \
+    /venv/bin/pip install --no-cache-dir -r /tmp/requirements.txt
 
-ADD Pipfile Pipfile.lock ./
-RUN pipenv install --deploy --system
+COPY kegbot ./kegbot
+COPY bin ./bin
+COPY setup.py ./
+RUN /venv/bin/pip install --no-cache-dir .
 
-ADD bin ./bin
-ADD kegbot ./kegbot
-ADD setup.py ./
-RUN python setup.py develop
+
+# Stage: test — layers test tooling on top of the builder venv so the suite runs
+# against the same installed deps as production. Never part of the runtime image
+# (`docker build` defaults to the final stage). Build/run it explicitly:
+#   docker build --target test -t kegbot/pycore:test . && docker run --rm kegbot/pycore:test
+# The suite mocks all I/O (Redis, the Kegbot API, threads), so no services are needed.
+FROM builder AS test
+
+ENV PATH="/venv/bin:$PATH" \
+    KEGBOT_ENV=test
+
+# Install test tooling, then re-install the package editable so coverage measures
+# the source tree at /app/kegbot/pycore rather than the copy in site-packages.
+RUN /venv/bin/pip install --no-cache-dir pytest coverage && \
+    /venv/bin/pip install --no-cache-dir -e .
+
+COPY pytest.ini .coveragerc ./
+
+# Default to the full suite with coverage and the CI gate; override to scope down.
+CMD ["sh", "-c", "coverage run -m pytest && coverage report -m --fail-under=80"]
+
+
+# Stage 2: lean runtime image — no pipenv, no curl, no build cache
+FROM python:3.11-alpine
+
+WORKDIR /app
+
+ENV PATH="/venv/bin:$PATH" \
+    KEGBOT_IN_DOCKER=True \
+    KEGBOT_ENV=debug
+
+RUN apk add --no-cache bash
+
+COPY --from=builder /venv /venv
+COPY bin ./bin
 
 ARG GIT_SHORT_SHA="unknown"
 ARG VERSION="unknown"
 ARG BUILD_DATE="unknown"
-RUN echo "GIT_SHORT_SHA=${GIT_SHORT_SHA}" > /etc/kegbot-pycore-version
-RUN echo "VERSION=${VERSION}" >> /etc/kegbot-pycore-version
-RUN echo "BUILD_DATE=${BUILD_DATE}" >> /etc/kegbot-pycore-version
-
-CMD [ \
-   "python", \
-   "bin/kegbot_core.py" \
-]
+RUN printf "GIT_SHORT_SHA=%s\nVERSION=%s\nBUILD_DATE=%s\n" \
+    "${GIT_SHORT_SHA}" "${VERSION}" "${BUILD_DATE}" \
+    > /etc/kegbot-pycore-version
+
+CMD ["python", "bin/kegbot_core.py"]

bin/kegboard_daemon.py

@@ -143,7 +143,12 @@ def active_devices(self):
   def service_devices(self):
     message_posted = False
     for kb in self.active_devices():
-      for message in kb.drain_messages():
+      try:
+        messages = kb.drain_messages()
+      except ValueError as e:
+        self._logger.warning('Skipping malformed message from %s: %s' % (kb, e))
+        continue
+      for message in messages:
         self.handle_message(kb, message)
         message_posted = True
     return message_posted

kegbot/pycore/backend_test.py

@@ -0,0 +1,119 @@
+"""Unittest for backend module (WebBackend exception translation)."""
+
+import socket
+import unittest
+from unittest import mock
+
+from kegbot.api import kbapi
+
+from . import backend
+from . import common_defs
+
+
+class WebBackendTestCase(unittest.TestCase):
+  def setUp(self):
+    # WebBackend builds a kbapi.Client in __init__; replace it with a mock so
+    # no network access occurs.
+    self.client_patcher = mock.patch.object(backend.kbapi, 'Client')
+    self.MockClient = self.client_patcher.start()
+    self.client = self.MockClient.return_value
+    self.wb = backend.WebBackend(api_url='http://example/api/', api_key='key')
+
+  def tearDown(self):
+    self.client_patcher.stop()
+
+  # --- RecordDrink ---
+  def testRecordDrinkSuccess(self):
+    self.client.record_drink.return_value = {'id': 1}
+    result = self.wb.RecordDrink('flow0', ticks=100, username='alice')
+    self.assertEqual({'id': 1}, result)
+    _, kwargs = self.client.record_drink.call_args
+    self.assertEqual('flow0', kwargs['tap_name'])
+    self.assertEqual(100, kwargs['ticks'])
+    self.assertEqual('alice', kwargs['username'])
+
+  def testRecordDrinkNotFound(self):
+    self.client.record_drink.side_effect = kbapi.NotFoundError('nope')
+    with self.assertRaises(backend.DoesNotExistException):
+      self.wb.RecordDrink('flow0', ticks=100)
+
+  def testRecordDrinkError(self):
+    self.client.record_drink.side_effect = kbapi.Error('boom')
+    with self.assertRaises(backend.BackendException):
+      self.wb.RecordDrink('flow0', ticks=100)
+
+  # --- LogSensorReading ---
+  def testLogSensorReadingSuccess(self):
+    self.client.log_sensor_reading.return_value = 'ok'
+    self.assertEqual('ok', self.wb.LogSensorReading('sensor0', 4.0))
+
+  def testLogSensorReadingOutOfRange(self):
+    too_hot = common_defs.THERMO_SENSOR_RANGE[1] + 100
+    with self.assertRaises(ValueError):
+      self.wb.LogSensorReading('sensor0', too_hot)
+    self.client.log_sensor_reading.assert_not_called()
+
+  def testLogSensorReadingNotFound(self):
+    self.client.log_sensor_reading.side_effect = kbapi.NotFoundError()
+    self.assertIsNone(self.wb.LogSensorReading('sensor0', 4.0))
+
+  def testLogSensorReadingServerError(self):
+    self.client.log_sensor_reading.side_effect = kbapi.ServerError()
+    self.assertIsNone(self.wb.LogSensorReading('sensor0', 4.0))
+
+  def testLogSensorReadingSocketError(self):
+    self.client.log_sensor_reading.side_effect = socket.error()
+    self.assertIsNone(self.wb.LogSensorReading('sensor0', 4.0))
+
+  # --- GetAuthToken ---
+  def testGetAuthTokenSuccess(self):
+    self.client.get_token.return_value = {'username': 'alice'}
+    self.assertEqual({'username': 'alice'},
+        self.wb.GetAuthToken('core.rfid', 'tok'))
+
+  def testGetAuthTokenNotFoundReRaised(self):
+    self.client.get_token.side_effect = kbapi.NotFoundError()
+    with self.assertRaises(kbapi.NotFoundError):
+      self.wb.GetAuthToken('core.rfid', 'tok')
+
+  def testGetAuthTokenSocketErrorBecomesNotFound(self):
+    self.client.get_token.side_effect = socket.error()
+    with self.assertRaises(kbapi.NotFoundError):
+      self.wb.GetAuthToken('core.rfid', 'tok')
+
+  # --- CreateController ---
+  def testCreateControllerCreatesDefaultMeters(self):
+    self.client.create_controller.return_value = {'id': 7}
+    result = self.wb.CreateController('kegboard')
+    self.assertEqual({'id': 7}, result)
+    self.assertEqual(2, self.client.create_flow_meter.call_count)
+    self.client.create_flow_meter.assert_any_call(7, 'flow0')
+    self.client.create_flow_meter.assert_any_call(7, 'flow1')
+
+  def testCreateControllerError(self):
+    self.client.create_controller.side_effect = kbapi.Error('boom')
+    with self.assertRaises(backend.BackendException):
+      self.wb.CreateController('kegboard')
+
+  # --- CancelDrink ---
+  def testCancelDrinkSuccess(self):
+    self.client.cancel_drink.return_value = 'ok'
+    self.assertEqual('ok', self.wb.CancelDrink(5))
+
+  def testCancelDrinkError(self):
+    self.client.cancel_drink.side_effect = kbapi.Error('boom')
+    with self.assertRaises(backend.BackendException):
+      self.wb.CancelDrink(5)
+
+  # --- pass-throughs ---
+  def testGetStatus(self):
+    self.client.status.return_value = {'ok': True}
+    self.assertEqual({'ok': True}, self.wb.GetStatus())
+
+  def testGetAllTaps(self):
+    self.client.taps.return_value = ['a', 'b']
+    self.assertEqual(['a', 'b'], self.wb.GetAllTaps())
+
+
+if __name__ == '__main__':
+  unittest.main()

kegbot/pycore/kb_threads.py

@@ -29,7 +29,7 @@ def ThreadMain(self):
       for thr in self._kb_env.GetThreads():
         if not thr.hasStarted():
           continue
-        if not thr.isAlive():
+        if not thr.is_alive():
           self._logger.error('Thread %s died unexpectedly' % thr.getName())
           self.Quit()
       time.sleep(0.5)