Speak protocol 3.0 and fall back to 2.0 only when that is the reflector's maximum.

A reflector that has moved to 3.0 will not accept a station that always announces 2.0, and a 2.x reflector still needs one explicit plaintext session. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-07 14:24:00 +02:00
parent 0930d4b727
commit 3d1a2a1669
24 changed files with 3180 additions and 244 deletions
@@ -0,0 +1,235 @@
+package station_test
+
+import (
+	"context"
+	"encoding/binary"
+	"testing"
+	"time"
+
+	"gitea.neitzel.de/konrad/funknetz/internal/opus"
+	"gitea.neitzel.de/konrad/funknetz/station"
+)
+
+func TestProtocol2DropsASequenceTooFarBehind(t *testing.T) {
+	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+	enc, err := opus.NewEncoder()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer enc.Close()
+	pkt, err := enc.Encode(make([]int16, opus.FrameSamples))
+	if err != nil {
+		t.Fatal(err)
+	}
+	in.refl.writeUDP(t, 101, 7, 32768, appendBytes(nil, pkt))
+	in.refl.writeUDP(t, 101, 7, 4, appendBytes(nil, pkt))
+	if _, err := in.st.WaitPlayback(t.Context(), opus.FrameSamples); err != nil {
+		t.Fatal(err)
+	}
+	if extra := in.st.TakePlayback(); len(extra) != 0 {
+		t.Fatalf("played %d extra samples from the old sequence", len(extra))
+	}
+}
+
+func TestProtocol2FlushCompletesOnce(t *testing.T) {
+	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+	in.st.HoldButton()
+	in.st.ReleaseButton()
+	typ, id, _, body := in.refl.readUDP(t)
+	if typ != 102 || id != 7 || len(body) != 0 {
+		t.Fatalf("flush type %d id %d len %d", typ, id, len(body))
+	}
+	in.refl.writeUDP(t, 103, 7, 0, nil)
+	in.clock.BlockUntilWaiting(t)
+	in.clock.Advance(3 * time.Second)
+	if pkt, _, ok := tryUDP(in.refl); ok {
+		t.Fatalf("sent %d bytes after flush-complete", len(pkt))
+	}
+	if got := in.st.State(); got.Connection != "Connected" || got.Protocol != "2.0" {
+		t.Fatalf("State = %+v", got)
+	}
+}
+
+func TestProtocol2FlushTimesOutWithoutASecondFlush(t *testing.T) {
+	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+	in.st.HoldButton()
+	in.st.ReleaseButton()
+	typ, _, _, _ := in.refl.readUDP(t)
+	if typ != 102 {
+		t.Fatalf("type %d, want flush", typ)
+	}
+	in.clock.BlockUntilWaiting(t)
+	in.clock.Advance(3 * time.Second)
+	if pkt, _, ok := tryUDP(in.refl); ok {
+		t.Fatalf("sent %d bytes when flush-complete did not arrive", len(pkt))
+	}
+}
+
+func TestSilenceFreesPushToTalkAndKeepsTheTalker(t *testing.T) {
+	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+	enc, err := opus.NewEncoder()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer enc.Close()
+	pkt, err := enc.Encode(make([]int16, opus.FrameSamples))
+	if err != nil {
+		t.Fatal(err)
+	}
+	in.refl.writeUDP(t, 101, 7, 0, appendBytes(nil, pkt))
+	if _, err := in.st.WaitPlayback(t.Context(), opus.FrameSamples); err != nil {
+		t.Fatal(err)
+	}
+	in.refl.writeFrame(t, 104, appendStr(appendU32(nil, 20), "DB0XYZ"))
+	if _, err := in.st.Wait(t.Context(), func(s station.State) bool { return s.Talker == "DB0XYZ" }); err != nil {
+		t.Fatal(err)
+	}
+	in.clock.BlockUntilWaiting(t)
+	in.clock.Advance(3 * time.Second)
+	in.st.HoldButton()
+	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
+	if _, _, ok := tryUDP(in.refl); ok {
+		t.Fatal("push-to-talk sent audio at exactly 3 seconds")
+	}
+	in.st.ReleaseButton()
+	in.clock.BlockUntilWaiting(t)
+	gen := in.clock.Generation()
+	in.clock.Advance(time.Second)
+	waitClock(t, in.clock, gen)
+	if got := in.st.State(); got.Talker != "DB0XYZ" || got.Transmitting {
+		t.Fatalf("State = %+v, want the talker kept and the button free", got)
+	}
+	in.st.HoldButton()
+	if !in.st.State().Transmitting {
+		t.Fatal("push-to-talk stayed blocked after the gap")
+	}
+	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
+	utyp, id, _, _ := in.refl.readUDP(t)
+	if utyp != 101 || id != 7 {
+		t.Fatalf("audio type %d id %d", utyp, id)
+	}
+}
+
+func TestUDPLimitRunsBeforeAnyDatagram(t *testing.T) {
+	refl := newReflector(t)
+	path := writeConfigFile(t, configJSON(t, "127.0.0.1", refl.port, "DL1ABC", "key", 20))
+	clock := newFakeClock()
+	st, err := station.New(station.Options{ConfigPath: path, Clock: clock})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	go func() { _ = st.Run(ctx) }()
+	refl.accept(t)
+	declineTo2(t, refl)
+	for step := 0; step < 6; step++ {
+		clock.BlockUntilWaiting(t)
+		clock.Advance(10 * time.Second)
+		if st.State().Connection == "Disconnected" && step < 5 {
+			t.Fatalf("dropped on step %d", step)
+		}
+		replyTCP(t, refl)
+	}
+	if _, err := st.Wait(t.Context(), func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil {
+		t.Fatal(err)
+	}
+	if got := st.State().Detail; got != "udp receive timeout" {
+		t.Fatalf("Detail = %q, want udp receive timeout", got)
+	}
+	clock.BlockUntilWaiting(t)
+	clock.Advance(2 * time.Second)
+	refl.accept(t)
+	typ, body := refl.readFrame(t)
+	if typ != 5 || !versionIs(body, 3, 0) {
+		t.Fatalf("retry type %d body %v, want 3.0", typ, body)
+	}
+}
+
+func TestOversizedFrameBeforeAuthRetries(t *testing.T) {
+	refl := newReflector(t)
+	path := writeConfigFile(t, configJSON(t, "127.0.0.1", refl.port, "DL1ABC", "key", 20))
+	clock := newFakeClock()
+	st, err := station.New(station.Options{ConfigPath: path, Clock: clock})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	go func() { _ = st.Run(ctx) }()
+	refl.accept(t)
+	declineTo2(t, refl)
+	payload := make([]byte, 65)
+	var hdr [4]byte
+	binary.BigEndian.PutUint32(hdr[:], 65)
+	if _, err := refl.tcp.Write(append(hdr[:], payload...)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil {
+		t.Fatal(err)
+	}
+	clock.BlockUntilWaiting(t)
+	clock.Advance(2 * time.Second)
+	refl.accept(t)
+	typ, body := refl.readFrame(t)
+	if typ != 5 || !versionIs(body, 3, 0) {
+		t.Fatalf("retry type %d body %v, want 3.0", typ, body)
+	}
+}
+
+func TestApplicationMessageBeforeLoginRetries(t *testing.T) {
+	refl := newReflector(t)
+	path := writeConfigFile(t, configJSON(t, "127.0.0.1", refl.port, "DL1ABC", "key", 20))
+	clock := newFakeClock()
+	st, err := station.New(station.Options{ConfigPath: path, Clock: clock})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	go func() { _ = st.Run(ctx) }()
+	refl.accept(t)
+	declineTo2(t, refl)
+	refl.writeFrame(t, 100, serverInfo(1, nil, []string{"OPUS"}))
+	got, err := st.Wait(t.Context(), func(s station.State) bool { return s.Detail == "Protocol error" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Connection != "Disconnected" || got.Protocol != "" {
+		t.Fatalf("State = %+v", got)
+	}
+	clock.BlockUntilWaiting(t)
+	clock.Advance(2 * time.Second)
+	refl.accept(t)
+	typ, body := refl.readFrame(t)
+	if typ != 5 || !versionIs(body, 3, 0) {
+		t.Fatalf("retry type %d body %v", typ, body)
+	}
+	if st.State().Detail != "Protocol error" {
+		t.Fatalf("Detail = %q, want the reason kept", st.State().Detail)
+	}
+}
+
+func TestIgnoredNoticesDoNotChangeTheStation(t *testing.T) {
+	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+	in.refl.writeFrame(t, 101, appendStr(appendU16(nil, 1), "DB0XYZ"))
+	in.refl.writeFrame(t, 102, appendStr(nil, "DB0XYZ"))
+	in.refl.writeFrame(t, 103, appendStr(nil, "DB0XYZ"))
+	in.refl.writeFrame(t, 109, appendU32(nil, 99))
+	body := appendStr(nil, "src")
+	body = appendStr(body, "name")
+	body = appendStr(body, "msg")
+	in.refl.writeFrame(t, 110, body)
+	in.refl.writeFrame(t, 112, append(appendU16(nil, 1), []byte{1, 40, 1}...))
+	in.refl.writeFrame(t, 113, append(appendU16(nil, 1), []byte{1, 1}...))
+	in.clock.BlockUntilWaiting(t)
+	in.clock.Advance(10 * time.Second)
+	typ, _ := in.refl.readFrame(t)
+	if typ != 1 {
+		t.Fatalf("TCP type %d, want a heartbeat", typ)
+	}
+	got := in.st.State()
+	if got.Connection != "Connected" || got.Protocol != "2.0" || got.TalkGroup != 20 || got.Talker != "" {
+		t.Fatalf("State = %+v", got)
+	}
+}