Files
internet_use_control/internal/mqtt/payload_test.go
Sam Rolfe 14b304e0b0 Phase 3: MQTT transport — retained config, status/telemetry, LWT presence, TLS/ACL design
- internal/mqtt: publish each device's desired state retained on
  restrict/<device>/config; subscribe to restrict/+/status and
  restrict/+/telemetry and write reports to SQLite; publish on every state
  change, republish on connect and every 15 minutes from SQLite, so the broker
  is repaired from the source of truth.
- store: presence (devices.online and last_seen, audited only on transitions),
  telemetry upsert per (device, day, app, domain) via migration 002, allow and
  block lists read into the config payload, device lookup by topic.
- config: MQTT settings. TLS plus per-device credentials are required; the
  pre-TLS shared broker needs an explicit CONTROL_MQTT_INSECURE escape hatch.
- web: republish config after lock, unlock, expiry, create and edit; broker card
  on the status partial.
- deploy/mosquitto: TLS listener, topic ACL, credential provisioning and a
  verifier that proves the policy on a throwaway broker. Not applied to the
  shared Mosquitto on .13 — that change needs Sam's approval.
- cmd/control: mqtt-clients prints the MQTT identity of every device.
- paho.mqtt.golang v1.5.1, which pulls golang.org/x/net and with it the
  golang.org/x/crypto bump to v0.42.0.

Verified on .13: retained config delivered to a fresh subscriber; a client
killed without DISCONNECT drove the broker's retained will and the service
flipped the device offline; telemetry landed in SQLite; make mqtt-verify passed
13/13; the service connected over TLS with credentials and a CA.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-13 14:37:43 +11:00

212 lines
6.2 KiB
Go

package mqtt
import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"time"
"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store"
)
func TestTopicHelpers(t *testing.T) {
const config = "restrict/harry_phone_2/config"
if got := StatusTopic(config); got != "restrict/harry_phone_2/status" {
t.Errorf("StatusTopic = %q", got)
}
if got := TelemetryTopic(config); got != "restrict/harry_phone_2/telemetry" {
t.Errorf("TelemetryTopic = %q", got)
}
for _, tc := range []struct {
topic string
suffix string
want string
ok bool
}{
{"restrict/harry_phone_2/status", SuffixStatus, config, true},
{"restrict/harry_phone_2/telemetry", SuffixTelemetry, config, true},
{"restrict/harry_phone_2/config", SuffixStatus, "", false},
{"restrict/status", SuffixStatus, "", false},
{"restrict/a/b/status", SuffixStatus, "", false},
{"other/harry_phone_2/status", SuffixStatus, "", false},
} {
got, ok := deviceFromTopic(tc.topic, tc.suffix)
if ok != tc.ok || got != tc.want {
t.Errorf("deviceFromTopic(%q, %q) = (%q, %v), want (%q, %v)",
tc.topic, tc.suffix, got, ok, tc.want, tc.ok)
}
}
}
func TestConfigMarshal(t *testing.T) {
until := time.Date(2026, 10, 13, 21, 0, 0, 0, time.UTC)
b, err := Config{
V: ConfigVersion,
DeviceID: 2,
Name: "harry_phone",
Platform: store.PlatformAndroid,
Locked: true,
Until: &until,
GeneratedAt: time.Date(2026, 10, 13, 3, 0, 0, 0, time.UTC),
}.Marshal()
if err != nil {
t.Fatalf("Marshal: %v", err)
}
// The lists must be present as arrays, never null, so a client can parse
// this without null checks.
if !strings.Contains(string(b), `"allow":[]`) || !strings.Contains(string(b), `"block":[]`) {
t.Errorf("payload has null lists: %s", b)
}
var back Config
if err := json.Unmarshal(b, &back); err != nil {
t.Fatalf("Unmarshal: %v", err)
}
if !back.Locked || back.Until == nil || !back.Until.Equal(until) {
t.Errorf("round trip lost the lock: %+v", back)
}
if back.V != ConfigVersion || back.DeviceID != 2 {
t.Errorf("round trip lost the identity: %+v", back)
}
}
func TestParseStatus(t *testing.T) {
if _, err := ParseStatus(nil); err == nil {
t.Error("expected an error for an empty payload")
}
if _, err := ParseStatus([]byte("{oops")); err == nil {
t.Error("expected an error for malformed JSON")
}
// An explicit offline (the will payload the broker publishes on a dropped
// connection) must read as offline.
off, err := ParseStatus([]byte(`{"online":false}`))
if err != nil {
t.Fatalf("ParseStatus: %v", err)
}
if off.IsOnline() {
t.Error(`{"online":false} read as online`)
}
// A report with no "online" field means the device is talking.
on, err := ParseStatus([]byte(`{"app":"com.spotify.music"}`))
if err != nil {
t.Fatalf("ParseStatus: %v", err)
}
if !on.IsOnline() {
t.Error("a report with no online field read as offline")
}
if on.App != "com.spotify.music" {
t.Errorf("App = %q", on.App)
}
at := on.Timestamp(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))
if !at.Equal(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)) {
t.Errorf("Timestamp with no at field = %v, want the fallback", at)
}
at = Status{At: "2026-10-13T07:30:00Z"}.Timestamp(time.Now())
if at.UTC().Format(time.RFC3339) != "2026-10-13T07:30:00Z" {
t.Errorf("Timestamp = %v", at)
}
}
func TestTelemetryRows(t *testing.T) {
payload := []byte(`{
"day": "2026-10-13",
"apps": [{"name": "com.spotify.music", "seconds": 600}, {"name": "", "seconds": 5}],
"domains": [{"name": "youtube.com", "seconds": 120}]
}`)
report, err := ParseTelemetry(payload)
if err != nil {
t.Fatalf("ParseTelemetry: %v", err)
}
rows := report.Rows()
if len(rows) != 2 {
t.Fatalf("Rows = %+v, want 2 rows", rows)
}
// Apps first, then domains.
if rows[0].App != "com.spotify.music" || rows[0].Seconds != 600 {
t.Errorf("rows[0] = %+v", rows[0])
}
if rows[1].Domain != "youtube.com" || rows[1].Seconds != 120 {
t.Errorf("rows[1] = %+v", rows[1])
}
if got := report.DayOrDefault(time.Now()); got != "2026-10-13" {
t.Errorf("DayOrDefault = %q, want the reported day", got)
}
// The flat single-row form, and a bad day falling back to today.
flat, err := ParseTelemetry([]byte(`{"app":"roblox","seconds":60,"day":"yesterday"}`))
if err != nil {
t.Fatalf("ParseTelemetry flat: %v", err)
}
flatRows := flat.Rows()
if len(flatRows) != 1 || flatRows[0].App != "roblox" {
t.Fatalf("flat Rows = %+v", flatRows)
}
now := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC)
if got := flat.DayOrDefault(now); got != "2026-03-04" {
t.Errorf("DayOrDefault with a bad day = %q, want 2026-03-04", got)
}
// Negative totals are dropped, not stored.
bad, err := ParseTelemetry([]byte(`{"apps":[{"name":"x","seconds":-1}]}`))
if err != nil {
t.Fatalf("ParseTelemetry: %v", err)
}
if len(bad.Rows()) != 0 {
t.Errorf("negative usage stored: %+v", bad.Rows())
}
if _, err := ParseTelemetry([]byte("not json")); err == nil {
t.Error("expected an error for malformed JSON")
}
}
func TestNewRejectsIncompleteOptions(t *testing.T) {
if _, err := New(Options{Broker: "tcp://127.0.0.1:1883"}); err == nil {
t.Error("expected an error without a device lookup, source and sink")
}
_, err := New(Options{
Broker: "tls://broker:8883",
CAFile: "/nonexistent/ca.crt",
Devices: stubLookup{},
Source: stubSource{},
Reports: stubSink{},
})
if err == nil {
t.Fatal("expected an error for a missing CA bundle")
}
if !strings.Contains(err.Error(), "ca.crt") {
t.Errorf("error = %v, want it to name the CA file", err)
}
}
type stubLookup struct{}
func (stubLookup) DeviceByConfigTopic(ctx context.Context, topic string) (*store.Device, error) {
return nil, errors.New("not implemented")
}
type stubSource struct{}
func (stubSource) ListDevices(ctx context.Context) ([]store.Device, error) { return nil, nil }
func (stubSource) DeviceLists(ctx context.Context, id int64) ([]string, []string, error) {
return nil, nil, nil
}
type stubSink struct{}
func (stubSink) SetPresence(ctx context.Context, id int64, online bool, at time.Time, actor string) (bool, error) {
return false, nil
}
func (stubSink) InsertTelemetry(ctx context.Context, id int64, day string, e []store.TelemetryEntry) (int, error) {
return 0, nil
}