From 536e74c24d91e8e895836b0c1cdf7d76d19afad1 Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 11 Jun 2026 18:03:47 -0700
Subject: [PATCH 1/9] kegboard_daemon: catch ValueError from malformed serial
 messages

Prevents the daemon from crashing when a corrupted/truncated message
arrives during high-frequency flow pulse bursts.
---
 bin/kegboard_daemon.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/bin/kegboard_daemon.py b/bin/kegboard_daemon.py
index 8350c2b..200b840 100755
--- a/bin/kegboard_daemon.py
+++ b/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

From feecc31983c7642ed06ca9dd7366267c362dc7a3 Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 11 Jun 2026 18:07:53 -0700
Subject: [PATCH 2/9] Dockerfile: pin to python:3.12-alpine

python:3-alpine now resolves to 3.14, which newer pipenv rejects as
an ambiguous match against `python_version = "3"` in --deploy mode.
---
 Dockerfile | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Dockerfile b/Dockerfile
index 14197fd..2a83218 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM python:3-alpine
+FROM python:3.12-alpine
 
 RUN mkdir /app
 WORKDIR /app

From c59d30e60ebec4a375058c34a7e9e52d2280c7b5 Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 11 Jun 2026 18:11:54 -0700
Subject: [PATCH 3/9] Dockerfile: pin pipenv<2024 to fix --deploy python
 version check

pipenv 2024+ rejects python_version = "3" as an ambiguous match
against 3.12.x in --deploy mode. Pin to <2024 to restore prior behavior.
---
 Dockerfile | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Dockerfile b/Dockerfile
index 2a83218..b10eaa0 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -12,7 +12,7 @@ RUN apk update && \
     apk add --no-cache \
       bash \
       curl && \
-   pip install pipenv
+   pip install "pipenv<2024"
 
 ADD Pipfile Pipfile.lock ./
 RUN pipenv install --deploy --system

From ec64094f72ba014162104a866cb444b3ae6f8d20 Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 11 Jun 2026 18:32:42 -0700
Subject: [PATCH 4/9] Dockerfile: use python:3.11-alpine instead of 3.12

Python 3.12 removed the `imp` module; the `future` package (and other
kegbot deps) still depend on it, causing a crash at startup. Python 3.11
retains `imp` and is supported until 2027.
---
 Dockerfile | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Dockerfile b/Dockerfile
index b10eaa0..7f629b8 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM python:3.12-alpine
+FROM python:3.11-alpine
 
 RUN mkdir /app
 WORKDIR /app

From 11d3d384566f32a9fbd6c09f77ced3c4e2f78e6a Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 11 Jun 2026 18:38:37 -0700
Subject: [PATCH 5/9] Add CLAUDE.md with development workflow and deployment
 notes

---
 CLAUDE.md | 66 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 66 insertions(+)
 create mode 100644 CLAUDE.md

diff --git a/CLAUDE.md b/CLAUDE.md
new file mode 100644
index 0000000..d91757f
--- /dev/null
+++ b/CLAUDE.md
@@ -0,0 +1,66 @@
+# 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 |
+
+## 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 ."
+```

From 8661c9dd9d0bc0fae598834eb32b75acd1c1973e Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 11 Jun 2026 21:53:03 -0700
Subject: [PATCH 6/9] Fix Python 3.11 compat: replace isAlive() with is_alive()

Thread.isAlive() was removed in Python 3.9; is_alive() is the replacement.
---
 kegbot/pycore/kb_threads.py | 2 +-
 kegbot/pycore/kegbot_app.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/kegbot/pycore/kb_threads.py b/kegbot/pycore/kb_threads.py
index 4d0669c..3882c8e 100644
--- a/kegbot/pycore/kb_threads.py
+++ b/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)
diff --git a/kegbot/pycore/kegbot_app.py b/kegbot/pycore/kegbot_app.py
index b5b77a5..33f60b1 100644
--- a/kegbot/pycore/kegbot_app.py
+++ b/kegbot/pycore/kegbot_app.py
@@ -112,7 +112,7 @@ def _MainLoop(self):
     while not self._do_quit:
       try:
         watchdog.join(0.5)
-        if not watchdog.isAlive() and not self._do_quit:
+        if not watchdog.is_alive() and not self._do_quit:
           self._logger.error("Watchdog thread exited, quitting")
           self.Quit()
           return

From e55fc184aabaff1d7db815b7a00376f22040ceb2 Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Fri, 12 Jun 2026 10:09:06 -0700
Subject: [PATCH 7/9] Slim image with multi-stage build
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Stage 1 installs pipenv, exports requirements, builds a venv, and
installs the package. Stage 2 copies only the venv and bin/ scripts,
dropping pipenv, curl, and all build cache — 211 MB → 161 MB.
---
 Dockerfile | 55 +++++++++++++++++++++++++++++-------------------------
 1 file changed, 30 insertions(+), 25 deletions(-)

diff --git a/Dockerfile b/Dockerfile
index 7f629b8..f89206a 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,35 +1,40 @@
-FROM python:3.11-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"
+
+COPY Pipfile Pipfile.lock ./
+RUN pipenv requirements > /tmp/requirements.txt
 
-ENV SHELL=/bin/sh \
-   PIP_NO_CACHE_DIR=1 \
-   KEGBOT_IN_DOCKER=True \
-   KEGBOT_ENV=debug
+RUN python -m venv /venv && \
+    /venv/bin/pip install --no-cache-dir -r /tmp/requirements.txt
 
-RUN apk update && \
-    apk add --no-cache \
-      bash \
-      curl && \
-   pip install "pipenv<2024"
+COPY kegbot ./kegbot
+COPY bin ./bin
+COPY setup.py ./
+RUN /venv/bin/pip install --no-cache-dir .
 
-ADD Pipfile Pipfile.lock ./
-RUN pipenv install --deploy --system
 
-ADD bin ./bin
-ADD kegbot ./kegbot
-ADD setup.py ./
-RUN python setup.py develop
+# 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"]

From f177e78657cac0624cb87f163b5e96eba52c9ad0 Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 18 Jun 2026 21:07:13 -0700
Subject: [PATCH 8/9] Fix three latent Python 3 bugs in manager and tap
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Surfaced while adding test coverage:

- TokenRecord defined only __cmp__, which Python 3 ignores, so equality
  fell back to identity. AuthenticationManager._TokenRemoved therefore
  never matched the stored record and silently no-op'd, leaving captive
  auth flows running after token removal. Add __eq__ based on AsTuple().
- TapManager._RemoveTap referenced self.logger (not self._logger) and
  self._meters (which only exists on FlowManager) — two AttributeErrors.
- Tap defined __eq__ without __hash__, making it unhashable under
  Python 3. Add __hash__ based on AsTuple().
---
 kegbot/pycore/manager.py | 8 ++++++--
 kegbot/pycore/tap.py     | 3 +++
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/kegbot/pycore/manager.py b/kegbot/pycore/manager.py
index 888da4d..cd06f97 100644
--- a/kegbot/pycore/manager.py
+++ b/kegbot/pycore/manager.py
@@ -91,9 +91,8 @@ def _RegisterOrUpdateTap(self, name, ml_per_tick, relay_name=None):
 
   def _RemoveTap(self, name):
     tap = self._taps.get(name)
-    self.logger.info('Removing tap: %s' % tap)
+    self._logger.info('Removing tap: %s' % tap)
     del self._taps[name]
-    del self._meters[name]
 
   def GetTap(self, name):
     """Returns the registered tap identified by `name`, or None."""
@@ -495,6 +494,11 @@ def IsRemoved(self):
   def __hash__(self):
     return hash(self.AsTuple())
 
+  def __eq__(self, other):
+    if not isinstance(other, TokenRecord):
+      return NotImplemented
+    return self.AsTuple() == other.AsTuple()
+
   def __cmp__(self, other):
     if not other:
       return -1
diff --git a/kegbot/pycore/tap.py b/kegbot/pycore/tap.py
index d7b67da..3236d3d 100644
--- a/kegbot/pycore/tap.py
+++ b/kegbot/pycore/tap.py
@@ -15,6 +15,9 @@ def __str__(self):
   def __eq__(self, other):
     return not not other and self.AsTuple() == other.AsTuple()
 
+  def __hash__(self):
+    return hash(self.AsTuple())
+
   def AsTuple(self):
     return self._name, self._ml_per_tick, self._relay_name
 

From 5f21a17f601efe6d9e1954b3d89daef03b516ade Mon Sep 17 00:00:00 2001
From: Frode Langelo <frode@langelo.net>
Date: Thu, 18 Jun 2026 21:07:31 -0700
Subject: [PATCH 9/9] Add test coverage, Docker test stage, and refreshed CI

Raise source coverage from ~58% to 85% (14 -> 77 tests):

- New suites: backend, kegnet, kbevent, util, kb_threads, plus expanded
  manager and kegbot env tests. All I/O (Redis, Kegbot API, threads) is
  mocked, so the suite needs no external services.
- Add a `test` stage to the Dockerfile (FROM builder AS test) mirroring
  kegbot-server: installs pytest/coverage, re-installs the package
  editable so coverage measures the source tree, and defaults to
  `coverage run -m pytest && coverage report --fail-under=80`.
- Point CI (pybuild.yml) at `docker build --target test`, exercising the
  real image build. Drop the now-unneeded MySQL/Redis services since the
  suite mocks all I/O.
- Remove the stale Travis config (Python 3.8); GitHub Actions is the
  active CI and the two diverging configs caused version drift.
- Add pytest.ini and .coveragerc; ignore .coverage / .pytest_cache.
- Document the recommended test workflow in CLAUDE.md.
---
 .coveragerc                      |  10 ++
 .github/workflows/pybuild.yml    |  24 ++--
 .gitignore                       |   4 +
 .travis.yml                      |  15 ---
 CLAUDE.md                        |  32 +++++
 Dockerfile                       |  21 ++++
 kegbot/pycore/backend_test.py    | 119 ++++++++++++++++++
 kegbot/pycore/kb_threads_test.py |  68 ++++++++++
 kegbot/pycore/kbevent_test.py    |  85 +++++++++++++
 kegbot/pycore/kegbot_test.py     |   6 +
 kegbot/pycore/kegnet_test.py     | 150 ++++++++++++++++++++++
 kegbot/pycore/manager_test.py    | 206 +++++++++++++++++++++++++++++++
 kegbot/pycore/util_test.py       |  36 ++++++
 pytest.ini                       |   3 +
 14 files changed, 748 insertions(+), 31 deletions(-)
 create mode 100644 .coveragerc
 delete mode 100644 .travis.yml
 create mode 100644 kegbot/pycore/backend_test.py
 create mode 100644 kegbot/pycore/kb_threads_test.py
 create mode 100644 kegbot/pycore/kbevent_test.py
 create mode 100644 kegbot/pycore/kegnet_test.py
 create mode 100644 kegbot/pycore/util_test.py
 create mode 100644 pytest.ini

diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 0000000..a1a4789
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,10 @@
+[run]
+source = kegbot/pycore
+omit =
+    */*_test.py
+
+[report]
+exclude_lines =
+    pragma: no cover
+    raise NotImplementedError
+    if __name__ == .__main__.:
diff --git a/.github/workflows/pybuild.yml b/.github/workflows/pybuild.yml
index 9ae1813..1e21533 100644
--- a/.github/workflows/pybuild.yml
+++ b/.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
diff --git a/.gitignore b/.gitignore
index 77faad0..68ccfde 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,7 @@ distribute-*
 *egg-info
 dist/
 docs/build/
+
+### test artifacts
+.coverage
+.pytest_cache/
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index c7a1b0e..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,15 +0,0 @@
-language: python
-python:
-  - "3.8"
-
-services:
-  - redis-server
-
-install:
-  - pip install pipenv
-  - pipenv install --dev
-  - python setup.py develop
-
-script:
-  - pipenv run pytest
-
diff --git a/CLAUDE.md b/CLAUDE.md
index d91757f..0ac7563 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -46,6 +46,38 @@ ssh kegberry "docker logs kegberry-pycore-1 --tail 50"
 | `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.
diff --git a/Dockerfile b/Dockerfile
index f89206a..4d9a5b7 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -16,6 +16,27 @@ COPY setup.py ./
 RUN /venv/bin/pip install --no-cache-dir .
 
 
+# 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
 
diff --git a/kegbot/pycore/backend_test.py b/kegbot/pycore/backend_test.py
new file mode 100644
index 0000000..d95f416
--- /dev/null
+++ b/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()
diff --git a/kegbot/pycore/kb_threads_test.py b/kegbot/pycore/kb_threads_test.py
new file mode 100644
index 0000000..d8052ed
--- /dev/null
+++ b/kegbot/pycore/kb_threads_test.py
@@ -0,0 +1,68 @@
+"""Unittest for kb_threads module (testable thread logic)."""
+
+import unittest
+from unittest import mock
+
+from kegbot.api import exceptions as api_exceptions
+
+from . import kb_threads
+from . import kbevent
+from .util import AttrDict
+
+
+class FakeEnv(object):
+  """Minimal kb_env stand-in exposing only the event hub."""
+  def __init__(self, hub):
+    self._hub = hub
+
+  def GetEventHub(self):
+    return self._hub
+
+
+class SyncThreadTestCase(unittest.TestCase):
+  def setUp(self):
+    self.hub = kbevent.EventHub()
+    self.env = FakeEnv(self.hub)
+
+  def _make_thread(self, backend):
+    return kb_threads.SyncThread(self.env, 'sync-thread', backend)
+
+  def testSyncPublishesSyncEvent(self):
+    backend = mock.Mock()
+    backend.GetStatus.return_value = {'current_session': {'id': 1}, 'taps': []}
+    thread = self._make_thread(backend)
+
+    received = []
+    self.hub.Subscribe(kbevent.SyncEvent, received.append)
+
+    status = thread.sync_now()
+    self.hub.Flush()
+
+    self.assertEqual({'current_session': {'id': 1}, 'taps': []}, status)
+    self.assertEqual(1, len(received))
+    self.assertIsInstance(received[0].data, AttrDict)
+    self.assertEqual(1, received[0].data.current_session.id)
+
+  def testSyncHandlesApiError(self):
+    backend = mock.Mock()
+    backend.GetStatus.side_effect = api_exceptions.Error('boom')
+    thread = self._make_thread(backend)
+
+    received = []
+    self.hub.Subscribe(kbevent.SyncEvent, received.append)
+
+    status = thread.sync_now()
+    self.hub.Flush()
+
+    self.assertEqual({}, status)
+    self.assertEqual(0, len(received))
+
+  def testHandleQuitStopsThread(self):
+    thread = self._make_thread(mock.Mock())
+    self.assertFalse(thread._quit)
+    thread._HandleQuit(kbevent.QuitEvent())
+    self.assertTrue(thread._quit)
+
+
+if __name__ == '__main__':
+  unittest.main()
diff --git a/kegbot/pycore/kbevent_test.py b/kegbot/pycore/kbevent_test.py
new file mode 100644
index 0000000..378f0f1
--- /dev/null
+++ b/kegbot/pycore/kbevent_test.py
@@ -0,0 +1,85 @@
+"""Unittest for kbevent module"""
+
+import json
+import unittest
+
+from . import kbevent
+
+
+class EventTestCase(unittest.TestCase):
+  def testToDictAndJson(self):
+    e = kbevent.MeterUpdate()
+    e.meter_name = 'flow0'
+    e.reading = 123
+
+    d = e.ToDict()
+    self.assertEqual('MeterUpdate', d['event'])
+    self.assertEqual('flow0', d['data']['meter_name'])
+    self.assertEqual(123, d['data']['reading'])
+
+    # ToJson is just a JSON encoding of ToDict.
+    self.assertEqual(json.loads(e.ToJson()), d)
+
+  def testUnknownFieldRaises(self):
+    e = kbevent.MeterUpdate()
+    with self.assertRaises(AttributeError):
+      e.nonexistent_field
+
+  def testDecodeFromString(self):
+    e = kbevent.MeterUpdate()
+    e.meter_name = 'flow0'
+    e.reading = 50
+
+    decoded = kbevent.DecodeEvent(e.ToJson())
+    self.assertIsInstance(decoded, kbevent.MeterUpdate)
+    self.assertEqual('flow0', decoded.meter_name)
+    self.assertEqual(50, decoded.reading)
+
+  def testDecodeFromDict(self):
+    decoded = kbevent.DecodeEvent(
+        {'event': 'MeterUpdate', 'data': {'meter_name': 'f', 'reading': 1}})
+    self.assertIsInstance(decoded, kbevent.MeterUpdate)
+    self.assertEqual('f', decoded.meter_name)
+
+  def testDecodeUnknownEventRaises(self):
+    with self.assertRaises(ValueError):
+      kbevent.DecodeEvent({'event': 'NoSuchEvent', 'data': {}})
+
+
+class EventHubTestCase(unittest.TestCase):
+  def setUp(self):
+    self.hub = kbevent.EventHub()
+
+  def testSubscribeAndDispatch(self):
+    received = []
+    self.hub.Subscribe(kbevent.Ping, received.append)
+    self.hub.PublishEvent(kbevent.Ping())
+    self.hub.DispatchNextEvent(timeout=0.1)
+    self.assertEqual(1, len(received))
+
+  def testUnsubscribe(self):
+    received = []
+    self.hub.Subscribe(kbevent.Ping, received.append)
+    self.hub.Unsubscribe(kbevent.Ping, received.append)
+    self.hub.PublishEvent(kbevent.Ping())
+    self.hub.Flush()
+    self.assertEqual(0, len(received))
+
+  def testFlushReturnsCount(self):
+    self.hub.PublishEvent(kbevent.Ping())
+    self.hub.PublishEvent(kbevent.Ping())
+    self.assertEqual(2, self.hub.Flush())
+
+  def testDispatchEmptyQueueIsNoOp(self):
+    # Should return without raising even though nothing is queued.
+    self.hub.DispatchNextEvent(timeout=0.01)
+
+  def testSubscribeRejectsPlainClass(self):
+    class Plain(object):
+      pass
+    with self.assertRaises(ValueError):
+      self.hub.Subscribe(Plain, lambda e: None)
+
+
+if __name__ == '__main__':
+  unittest.main()
diff --git a/kegbot/pycore/kegbot_test.py b/kegbot/pycore/kegbot_test.py
index 11280e9..349d02c 100644
--- a/kegbot/pycore/kegbot_test.py
+++ b/kegbot/pycore/kegbot_test.py
@@ -80,5 +80,11 @@ def testPour(self):
     flows = flow_manager.GetActiveFlows()
     self.assertEqual(1, len(flows))
 
+  def testEnvAccessors(self):
+    self.assertIsNotNone(self.kb.GetWatchdogThread())
+    self.assertIsNotNone(self.kb.GetBackend())
+    self.assertIsNotNone(self.kb.GetAuthenticationManager())
+    self.assertTrue(len(self.kb.GetThreads()) > 0)
+
 if __name__ == '__main__':
   unittest.main()
diff --git a/kegbot/pycore/kegnet_test.py b/kegbot/pycore/kegnet_test.py
new file mode 100644
index 0000000..0036f76
--- /dev/null
+++ b/kegbot/pycore/kegnet_test.py
@@ -0,0 +1,150 @@
+"""Unittest for kegnet module (Redis pub/sub client)."""
+
+import json
+import unittest
+from unittest import mock
+
+import redis
+
+from . import kbevent
+from . import kegnet
+
+
+class KegnetClientTestCase(unittest.TestCase):
+  def setUp(self):
+    # KegnetClient connects to Redis in __init__; replace from_url with a mock.
+    self.patcher = mock.patch.object(kegnet.redis, 'from_url')
+    self.mock_from_url = self.patcher.start()
+    self.redis = self.mock_from_url.return_value
+    self.client = kegnet.KegnetClient(redis_url='redis://localhost:6379/0',
+        channel_name='kegnet')
+
+  def tearDown(self):
+    self.patcher.stop()
+
+  def _published(self):
+    self.redis.publish.assert_called_once()
+    (channel, payload), _ = self.redis.publish.call_args
+    return channel, json.loads(payload)
+
+  def testPing(self):
+    self.redis.ping.return_value = True
+    self.assertTrue(self.client.ping())
+
+  def testPingConnectionError(self):
+    self.redis.ping.side_effect = redis.exceptions.ConnectionError()
+    self.assertFalse(self.client.ping())
+
+  def testSendMessageSwallowsConnectionError(self):
+    self.redis.publish.side_effect = redis.exceptions.ConnectionError()
+    # Must not raise; a dropped message is logged and ignored.
+    self.client.SendMeterUpdate('flow0', 100)
+
+  def testSendMeterUpdate(self):
+    self.client.SendMeterUpdate('flow0', 100)
+    channel, msg = self._published()
+    self.assertEqual('kegnet', channel)
+    self.assertEqual('MeterUpdate', msg['event'])
+    self.assertEqual('flow0', msg['data']['meter_name'])
+    self.assertEqual(100, msg['data']['reading'])
+
+  def testSendFlowStart(self):
+    self.client.SendFlowStart('flow0')
+    _, msg = self._published()
+    self.assertEqual('FlowRequest', msg['event'])
+    self.assertEqual(kbevent.FlowRequest.Action.START_FLOW,
+        msg['data']['request'])
+
+  def testSendFlowStop(self):
+    self.client.SendFlowStop('flow0')
+    _, msg = self._published()
+    self.assertEqual(kbevent.FlowRequest.Action.STOP_FLOW,
+        msg['data']['request'])
+
+  def testSendThermoUpdate(self):
+    self.client.SendThermoUpdate('sensor0', 4.0)
+    _, msg = self._published()
+    self.assertEqual('ThermoEvent', msg['event'])
+    self.assertEqual('sensor0', msg['data']['sensor_name'])
+    self.assertEqual(4.0, msg['data']['sensor_value'])
+
+  def testSendAuthTokenAdd(self):
+    self.client.SendAuthTokenAdd('flow0', 'core.rfid', 'tok')
+    _, msg = self._published()
+    self.assertEqual('TokenAuthEvent', msg['event'])
+    self.assertEqual(kbevent.TokenAuthEvent.TokenState.ADDED,
+        msg['data']['status'])
+
+  def testSendAuthTokenRemove(self):
+    self.client.SendAuthTokenRemove('flow0', 'core.rfid', 'tok')
+    _, msg = self._published()
+    self.assertEqual(kbevent.TokenAuthEvent.TokenState.REMOVED,
+        msg['data']['status'])
+
+  def testSendControllerConnected(self):
+    self.client.SendControllerConnectedEvent('kegboard')
+    _, msg = self._published()
+    self.assertEqual('ControllerConnectedEvent', msg['event'])
+    self.assertEqual('kegboard', msg['data']['controller_name'])
+
+
+class HandleMessageTestCase(unittest.TestCase):
+  def setUp(self):
+    self.patcher = mock.patch.object(kegnet.redis, 'from_url')
+    self.patcher.start()
+
+    received = {'new': [], 'flow': [], 'drink': [], 'relay': []}
+    self.received = received
+
+    class RecordingClient(kegnet.KegnetClient):
+      def onNewEvent(self, event):
+        received['new'].append(event)
+
+      def onFlowUpdate(self, event):
+        received['flow'].append(event)
+
+      def onDrinkCreated(self, event):
+        received['drink'].append(event)
+
+      def onSetRelayOutput(self, event):
+        received['relay'].append(event)
+
+    self.client = RecordingClient()
+
+  def tearDown(self):
+    self.patcher.stop()
+
+  def _msg(self, event):
+    return {'type': 'message', 'data': event.ToJson()}
+
+  def testIgnoresNonMessageType(self):
+    self.client._handle_message({'type': 'subscribe', 'data': 1})
+    self.assertEqual(0, len(self.received['new']))
+
+  def testIgnoresUndecodableEvent(self):
+    self.client._handle_message(
+        {'type': 'message', 'data': '{"event": "Nope", "data": {}}'})
+    self.assertEqual(0, len(self.received['new']))
+
+  def testDispatchesFlowUpdate(self):
+    e = kbevent.FlowUpdate()
+    e.flow_id = 1
+    self.client._handle_message(self._msg(e))
+    self.assertEqual(1, len(self.received['new']))
+    self.assertEqual(1, len(self.received['flow']))
+
+  def testDispatchesDrinkCreated(self):
+    e = kbevent.DrinkCreatedEvent()
+    e.drink_id = 1
+    self.client._handle_message(self._msg(e))
+    self.assertEqual(1, len(self.received['drink']))
+
+  def testDispatchesSetRelayOutput(self):
+    e = kbevent.SetRelayOutputEvent()
+    e.output_name = 'relay0'
+    self.client._handle_message(self._msg(e))
+    self.assertEqual(1, len(self.received['relay']))
+
+
+if __name__ == '__main__':
+  unittest.main()
diff --git a/kegbot/pycore/manager_test.py b/kegbot/pycore/manager_test.py
index 2af9ab0..551c8f3 100644
--- a/kegbot/pycore/manager_test.py
+++ b/kegbot/pycore/manager_test.py
@@ -4,10 +4,50 @@
 
 import datetime
 import unittest
+from unittest import mock
 
+from kegbot.api import kbapi
+
+from . import backend
 from . import common_defs
 from . import kbevent
 from . import manager
+from .util import AttrDict
+
+
+class FakeBackend(backend.Backend):
+  """In-memory backend that records calls for assertions."""
+  def __init__(self):
+    self.drinks = []
+    self.sensor_readings = []
+    self.tokens = {}  # (auth_device, token_value) -> AttrDict
+    self.record_drink_exception = None
+
+  def RecordDrink(self, meter_name, ticks, volume_ml=None, username=None,
+      pour_time=None, duration=0, auth_token=None, spilled=False, shout=''):
+    if self.record_drink_exception:
+      raise self.record_drink_exception
+    drink = AttrDict({
+      'id': len(self.drinks) + 1,
+      'time': pour_time,
+      'volume_ml': volume_ml if volume_ml is not None else ticks,
+      'ticks': ticks,
+      'keg_id': 1,
+      'user_id': username,
+    })
+    self.drinks.append(drink)
+    return drink
+
+  def LogSensorReading(self, sensor_name, temperature, when=None):
+    self.sensor_readings.append((sensor_name, temperature, when))
+    return True
+
+  def GetAuthToken(self, auth_device, token_value):
+    token = self.tokens.get((auth_device, token_value))
+    if token is None:
+      raise kbapi.NotFoundError('no such token')
+    return token
+
 
 class FlowManagerTestCase(unittest.TestCase):
   def setUp(self):
@@ -124,5 +164,171 @@ def t(stamp):
     self.assertTrue(len(idle_flows) == 1)
 
 
+class DrinkManagerTestCase(unittest.TestCase):
+  def setUp(self):
+    self.hub = kbevent.EventHub()
+    self.backend = FakeBackend()
+    self.drink_manager = manager.DrinkManager(self.hub, self.backend)
+
+  def _completed_event(self, ticks=100, volume_ml=50, username='alice'):
+    e = kbevent.FlowUpdate()
+    e.flow_id = 0x1234
+    e.meter_name = 'flow0'
+    e.state = kbevent.FlowUpdate.FlowState.COMPLETED
+    e.username = username
+    e.start_time = datetime.datetime.fromtimestamp(0)
+    e.last_activity_time = datetime.datetime.fromtimestamp(5)
+    e.ticks = ticks
+    e.volume_ml = volume_ml
+    return e
+
+  def testRecordsCompletedFlow(self):
+    created = []
+    self.hub.Subscribe(kbevent.DrinkCreatedEvent, created.append)
+
+    self.drink_manager.HandleFlowUpdateEvent(self._completed_event())
+
+    self.assertEqual(1, len(self.backend.drinks))
+    self.assertEqual(100, self.backend.drinks[0].ticks)
+    self.assertEqual('alice', self.backend.drinks[0].user_id)
+
+    # A DrinkCreatedEvent should have been published for downstream listeners.
+    self.hub.Flush()
+    self.assertEqual(1, len(created))
+    self.assertEqual(0x1234, created[0].flow_id)
+
+  def testIgnoresNonCompletedFlow(self):
+    event = self._completed_event()
+    event.state = kbevent.FlowUpdate.FlowState.ACTIVE
+    self.drink_manager.HandleFlowUpdateEvent(event)
+    self.assertEqual(0, len(self.backend.drinks))
+
+  def testSkipsTinyPour(self):
+    self.drink_manager.HandleFlowUpdateEvent(
+        self._completed_event(volume_ml=common_defs.MIN_VOLUME_TO_RECORD - 1))
+    self.assertEqual(0, len(self.backend.drinks))
+
+  def testSkipsZeroTicks(self):
+    self.drink_manager.HandleFlowUpdateEvent(self._completed_event(ticks=0))
+    self.assertEqual(0, len(self.backend.drinks))
+
+  def testRetriesThenDropsOnBackendError(self):
+    self.backend.record_drink_exception = backend.BackendException('boom')
+
+    # First attempt (via the event handler) seeds the retry counter and requeues.
+    self.drink_manager.HandleFlowUpdateEvent(self._completed_event())
+    self.assertEqual(1, len(self.drink_manager._pending))
+
+    # Default maximum_event_retries is 3, so the event survives two more flushes
+    # before being dropped.
+    self.drink_manager._FlushPending()
+    self.assertEqual(1, len(self.drink_manager._pending))
+    self.drink_manager._FlushPending()
+    self.assertEqual(0, len(self.drink_manager._pending))
+    self.assertEqual(0, len(self.backend.drinks))
+
+
+class ThermoManagerTestCase(unittest.TestCase):
+  def setUp(self):
+    self.hub = kbevent.EventHub()
+    self.backend = FakeBackend()
+    self.thermo_manager = manager.ThermoManager(self.hub, self.backend)
+
+  def _event(self, name='sensor0', value=4.0):
+    e = kbevent.ThermoEvent()
+    e.sensor_name = name
+    e.sensor_value = value
+    return e
+
+  def testRecordsReading(self):
+    self.thermo_manager._HandleThermoUpdateEvent(self._event(value=4.0))
+    self.assertEqual(1, len(self.backend.sensor_readings))
+    self.assertEqual('sensor0', self.backend.sensor_readings[0][0])
+    self.assertEqual(4.0, self.backend.sensor_readings[0][1])
+
+  def testRejectsOutOfRange(self):
+    too_hot = common_defs.THERMO_SENSOR_RANGE[1] + 100
+    self.thermo_manager._HandleThermoUpdateEvent(self._event(value=too_hot))
+    self.assertEqual(0, len(self.backend.sensor_readings))
+
+  def testDropsDuplicateWithinSameMinute(self):
+    with mock.patch('kegbot.pycore.manager.datetime') as mock_dt:
+      mock_dt.datetime.now.return_value = datetime.datetime(2024, 1, 1, 12, 30, 15)
+      mock_dt.timedelta = datetime.timedelta
+      self.thermo_manager._HandleThermoUpdateEvent(self._event(value=4.0))
+      self.thermo_manager._HandleThermoUpdateEvent(self._event(value=5.0))
+    self.assertEqual(1, len(self.backend.sensor_readings))
+
+
+class AuthenticationManagerTestCase(unittest.TestCase):
+  def setUp(self):
+    self.hub = kbevent.EventHub()
+    self.backend = FakeBackend()
+    self.tap_manager = manager.TapManager(self.hub, self.backend)
+    self.flow_manager = manager.FlowManager(self.hub, self.tap_manager)
+    self.auth_manager = manager.AuthenticationManager(
+        self.hub, self.flow_manager, self.tap_manager, self.backend)
+    self.tap_manager._RegisterOrUpdateTap('flow0', ml_per_tick=0.5,
+        relay_name='relay0')
+
+  def _auth_event(self, device, status, meter_name='flow0', token_value='tok1'):
+    e = kbevent.TokenAuthEvent()
+    e.meter_name = meter_name
+    e.auth_device_name = device
+    e.token_value = token_value
+    e.status = status
+    return e
+
+  def testCaptiveTokenStartsAndEndsFlow(self):
+    self.backend.tokens[(common_defs.AUTH_MODULE_CORE_ONEWIRE, 'tok1')] = \
+        AttrDict({'username': 'alice', 'enabled': True})
+
+    self.auth_manager.HandleAuthTokenEvent(self._auth_event(
+        common_defs.AUTH_MODULE_CORE_ONEWIRE,
+        kbevent.TokenAuthEvent.TokenState.ADDED))
+
+    flow = self.flow_manager.GetFlow('flow0')
+    self.assertIsNotNone(flow)
+    self.assertEqual('alice', flow.GetUsername())
+
+    # Regression test: TokenRecord equality was broken under Python 3 (only
+    # __cmp__ was defined), so removing a captive token silently no-op'd and
+    # the flow was never stopped.
+    self.auth_manager.HandleAuthTokenEvent(self._auth_event(
+        common_defs.AUTH_MODULE_CORE_ONEWIRE,
+        kbevent.TokenAuthEvent.TokenState.REMOVED))
+    self.assertIsNone(self.flow_manager.GetFlow('flow0'))
+
+  def testNonCaptiveTokenRemovalKeepsFlow(self):
+    self.backend.tokens[(common_defs.AUTH_MODULE_CORE_RFID, 'tok1')] = \
+        AttrDict({'username': 'bob', 'enabled': True})
+
+    self.auth_manager.HandleAuthTokenEvent(self._auth_event(
+        common_defs.AUTH_MODULE_CORE_RFID,
+        kbevent.TokenAuthEvent.TokenState.ADDED))
+    self.assertIsNotNone(self.flow_manager.GetFlow('flow0'))
+
+    # Non-captive (contactless) devices leave the flow running; it times out
+    # rather than ending immediately on token removal.
+    self.auth_manager.HandleAuthTokenEvent(self._auth_event(
+        common_defs.AUTH_MODULE_CORE_RFID,
+        kbevent.TokenAuthEvent.TokenState.REMOVED))
+    self.assertIsNotNone(self.flow_manager.GetFlow('flow0'))
+
+  def testUnknownTokenStartsNoFlow(self):
+    self.auth_manager.HandleAuthTokenEvent(self._auth_event(
+        common_defs.AUTH_MODULE_CORE_ONEWIRE,
+        kbevent.TokenAuthEvent.TokenState.ADDED))
+    self.assertIsNone(self.flow_manager.GetFlow('flow0'))
+
+  def testDisabledTokenStartsNoFlow(self):
+    self.backend.tokens[(common_defs.AUTH_MODULE_CORE_ONEWIRE, 'tok1')] = \
+        AttrDict({'username': 'alice', 'enabled': False})
+    self.auth_manager.HandleAuthTokenEvent(self._auth_event(
+        common_defs.AUTH_MODULE_CORE_ONEWIRE,
+        kbevent.TokenAuthEvent.TokenState.ADDED))
+    self.assertIsNone(self.flow_manager.GetFlow('flow0'))
+
+
 if __name__ == '__main__':
   unittest.main()
diff --git a/kegbot/pycore/util_test.py b/kegbot/pycore/util_test.py
new file mode 100644
index 0000000..1f78eb3
--- /dev/null
+++ b/kegbot/pycore/util_test.py
@@ -0,0 +1,36 @@
+"""Unittest for util module"""
+
+import unittest
+
+from .util import AttrDict
+
+
+class AttrDictTestCase(unittest.TestCase):
+  def testAttributeAccess(self):
+    d = AttrDict({'a': 1})
+    self.assertEqual(1, d.a)
+    self.assertEqual(1, d['a'])
+
+  def testNestedDictBecomesAttrDict(self):
+    d = AttrDict({'outer': {'inner': 5}})
+    self.assertIsInstance(d.outer, AttrDict)
+    self.assertEqual(5, d.outer.inner)
+
+  def testMissingKeyRaisesAttributeError(self):
+    d = AttrDict({'a': 1})
+    with self.assertRaises(AttributeError):
+      d.missing
+
+  def testSetAttribute(self):
+    d = AttrDict()
+    d.x = 10
+    self.assertEqual(10, d['x'])
+    self.assertEqual(10, d.x)
+
+  def testEmptyInit(self):
+    d = AttrDict()
+    self.assertEqual(0, len(d))
+
+
+if __name__ == '__main__':
+  unittest.main()
diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..e3d2c86
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,3 @@
+[pytest]
+testpaths = kegbot/pycore
+python_files = *_test.py