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
new file mode 100644
index 0000000..0ac7563
--- /dev/null
+++ b/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 ."
+```
diff --git a/Dockerfile b/Dockerfile
index 14197fd..4d9a5b7 100644
--- a/Dockerfile
+++ b/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"]
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
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.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/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_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
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.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/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/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
 
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