1package tuner23import (4 "bytes"5 "context"6 "errors"7 "sync"8 "testing"9 "time"1011 "airwaves/internal/hdhr"12 "airwaves/internal/hdhr/hdhrtest"13 "airwaves/internal/ts"14 "airwaves/internal/ts/tstest"15)1617// Three RF channels: KMGH's two programs on 177 MHz, KUSA's on 189 MHz,18// KTVD's on 557 MHz; 509 MHz has nothing to lock.19const (20 rfKMGH = 177_000_00021 rfKUSA = 189_000_00022 rfKTVD = 557_000_00023 rfNone = 509_000_00024)2526var muxes = map[int64][]byte{27 rfKMGH: tstest.Mux{TSID: 1, Programs: []tstest.Program{28 {Number: 3, PMTPID: 0x30, VideoPID: 0x31, Audio: []tstest.Audio{{PID: 0x34, Lang: "eng"}, {PID: 0x35, Lang: "spa"}}, Major: 7, Minor: 1, Name: "KMGH"},29 {Number: 4, PMTPID: 0x40, VideoPID: 0x41, Audio: []tstest.Audio{{PID: 0x44, Lang: "eng"}}, Major: 7, Minor: 2, Name: "Grit"},30 }}.Packets(700),31 rfKUSA: tstest.Mux{TSID: 2, Programs: []tstest.Program{{Number: 3, PMTPID: 0x30, VideoPID: 0x31, Audio: []tstest.Audio{{PID: 0x34, Lang: "eng"}}, Major: 9, Minor: 1, Name: "KUSA"}}}.Packets(700),32 rfKTVD: tstest.Mux{TSID: 3, Programs: []tstest.Program{{Number: 1, PMTPID: 0x30, VideoPID: 0x31, Audio: []tstest.Audio{{PID: 0x34, Lang: "eng"}}, Major: 20, Minor: 1, Name: "KTVD"}}}.Packets(700),33}3435func setup(t *testing.T, rate ...int) (*Manager, *hdhrtest.Server) {36 t.Helper()37 d := hdhrtest.Device{Mux: func(f int64) []byte { return muxes[f] }}38 if len(rate) > 0 {39 d.Rate = rate[0]40 }41 srv := hdhrtest.New(t, d)42 dev, err := hdhr.Open(context.Background(), srv.URL(), nil)43 if err != nil {44 t.Fatal(err)45 }46 m := NewManager()47 m.Linger = 048 m.SetDevice(dev, dev.Tuners)49 return m, srv50}5152func open(t *testing.T, m *Manager, r Request) *Sub {53 t.Helper()54 s, err := m.Open(context.Background(), r)55 if err != nil {56 t.Fatalf("open %+v: %v", r, err)57 }58 t.Cleanup(s.Close)59 return s60}6162func wait(t *testing.T, s *Sub) ts.Program {63 t.Helper()64 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)65 defer cancel()66 p, err := s.Wait(ctx)67 if err != nil {68 t.Fatalf("wait for program %d: %v", s.program, err)69 }70 return p71}7273// read copies what s receives for a moment and lists the programs in it.74func read(t *testing.T, s *Sub) []ts.Program {75 t.Helper()76 var buf bytes.Buffer77 ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)78 defer cancel()79 _ = s.Copy(ctx, &buf)80 d := ts.NewDemux()81 var split ts.Splitter82 split.Write(buf.Bytes(), d.Packet)83 return d.Programs()84}8586func eventually(t *testing.T, what string, ok func() bool) {87 t.Helper()88 for deadline := time.Now().Add(5 * time.Second); !ok(); time.Sleep(20 * time.Millisecond) {89 if time.Now().After(deadline) {90 t.Fatalf("timed out waiting for %s", what)91 }92 }93}9495// TestShareRF: two programs of one RF channel take one tuner, and each96// user gets their own program alone, with its audio tracks described.97func TestShareRF(t *testing.T) {98 m, srv := setup(t)99 a := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1"})100 b := open(t, m, Request{FrequencyHz: rfKMGH, Program: 4, Priority: Recording, Who: "Grit movie"})101 pa, pb := wait(t, a), wait(t, b)102 if a.Tuner() != b.Tuner() || srv.Streams() != 1 {103 t.Fatalf("tuners %d and %d, %d streams; want one shared", a.Tuner(), b.Tuner(), srv.Streams())104 }105 if audio, _ := AudioTracks(pa); len(audio) != 2 || audio[1].Lang != "spa" || audio[1].Map != "0:a:1" {106 t.Errorf("7.1's audio = %+v", audio)107 }108 if len(pb.Audio()) != 1 {109 t.Errorf("7.2's audio = %+v", pb.Audio())110 }111 if got := read(t, a); len(got) != 1 || got[0].Number != 3 {112 t.Errorf("7.1's stream carries %+v, want program 3 alone", got)113 }114 if got := read(t, b); len(got) != 1 || got[0].Number != 4 {115 t.Errorf("7.2's stream carries %+v, want program 4 alone", got)116 }117 tuners, err := m.Tuners(context.Background())118 if err != nil {119 t.Fatal(err)120 }121 on := tuners[a.Tuner()]122 if on.FrequencyHz != rfKMGH || len(on.Users) != 2 || on.Top != Recording || !on.Locked || on.Packets == 0 || on.Device.StrengthPct == nil {123 t.Errorf("tuner state = %+v", on)124 }125}126127// TestPriorities: with every tuner taken, a viewer can't have one from128// another viewer, but a recording can: the one tuned last, with the129// least to lose. The viewer it ends learns why.130func TestPriorities(t *testing.T) {131 m, _ := setup(t)132 a := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1", Owner: "den"})133 wait(t, a)134 b := open(t, m, Request{FrequencyHz: rfKUSA, Program: 3, Priority: Watching, Who: "9.1", Owner: "kitchen"})135 wait(t, b)136137 _, err := m.Open(context.Background(), Request{FrequencyHz: rfKTVD, Program: 1, Priority: Watching, Who: "20.1", Owner: "attic"})138 var busy *BusyError139 if !errors.As(err, &busy) || len(busy.Users) != 2 {140 t.Fatalf("a third viewer got %v, want every tuner in use by the other two", err)141 }142143 r := open(t, m, Request{FrequencyHz: rfKTVD, Program: 1, Priority: Recording, Who: "The News"})144 wait(t, r)145 <-b.Done()146 var pre *PreemptedError147 if !errors.As(b.Err(), &pre) || pre.By != "recording The News" {148 t.Fatalf("the kitchen's viewer ended with %v", b.Err())149 }150 if a.Err() != nil {151 t.Fatalf("the den's viewer ended too: %v", a.Err())152 }153}154155// TestOwnerRetunes: a viewer changing channel gets their own tuner back156// even when every tuner is busy.157func TestOwnerRetunes(t *testing.T) {158 m, srv := setup(t)159 a := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1", Owner: "den"})160 b := open(t, m, Request{FrequencyHz: rfKUSA, Program: 3, Priority: Recording, Who: "The News"})161 wait(t, a)162 wait(t, b)163 c := open(t, m, Request{FrequencyHz: rfKTVD, Program: 1, Priority: Watching, Who: "20.1", Owner: "den"})164 wait(t, c)165 if a.Err() == nil || b.Err() != nil || c.Tuner() != a.Tuner() {166 t.Fatalf("den's old tune %v, the recording %v, new tune on tuner %d", a.Err(), b.Err(), c.Tuner())167 }168 eventually(t, "two streams", func() bool { return srv.Streams() == 2 })169}170171// TestMeasuringYields: measuring takes only a free tuner, and a viewer172// takes it from measuring.173func TestMeasuringYields(t *testing.T) {174 m, _ := setup(t)175 a := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1"})176 meas := open(t, m, Request{FrequencyHz: rfKUSA, Priority: Measuring, Who: "RF 9"})177 wait(t, a)178 wait(t, meas)179 if !meas.Locked() {180 t.Fatal("measuring an RF channel with a multiplex has no lock")181 }182 if _, err := m.Open(context.Background(), Request{FrequencyHz: rfKTVD, Priority: Measuring, Who: "RF 26"}); err == nil {183 t.Fatal("measuring took a viewer's tuner")184 }185 v := open(t, m, Request{FrequencyHz: rfKTVD, Program: 1, Priority: Watching, Who: "20.1"})186 wait(t, v)187 <-meas.Done()188 if a.Err() != nil {189 t.Fatalf("the first viewer ended: %v", a.Err())190 }191}192193// TestNoLock: on an RF channel with nothing to lock, Wait gives up when194// its context does, and the device's reading of the tuner is there.195func TestNoLock(t *testing.T) {196 m, _ := setup(t)197 s := open(t, m, Request{FrequencyHz: rfNone, Program: 1, Priority: Watching, Who: "RF 20"})198 ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)199 defer cancel()200 if _, err := s.Wait(ctx); !errors.Is(err, context.DeadlineExceeded) {201 t.Fatalf("wait = %v, want the deadline", err)202 }203 if s.Locked() {204 t.Fatal("locked with nothing to lock")205 }206 tuners, err := m.Tuners(context.Background())207 if err != nil {208 t.Fatal(err)209 }210 st := tuners[s.Tuner()]211 if st.FrequencyHz != rfNone || st.Locked || st.Device.StrengthPct == nil || *st.Device.StrengthPct != 40 {212 t.Fatalf("tuner state = %+v", st)213 }214}215216// TestNoProgram: a program the RF channel doesn't carry fails once its217// tables are in.218func TestNoProgram(t *testing.T) {219 m, _ := setup(t)220 s := open(t, m, Request{FrequencyHz: rfKUSA, Program: 9, Priority: Watching, Who: "9.9"})221 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)222 defer cancel()223 if _, err := s.Wait(ctx); !errors.Is(err, ErrNoProgram) {224 t.Fatalf("wait = %v, want ErrNoProgram", err)225 }226}227228// TestLinger: a tuner stays on its RF channel for a moment after its last229// user, so coming back is instant, then lets go.230func TestLinger(t *testing.T) {231 m, srv := setup(t)232 m.Linger = 200 * time.Millisecond233 a := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1"})234 wait(t, a)235 a.Close()236 b := open(t, m, Request{FrequencyHz: rfKMGH, Program: 4, Priority: Watching, Who: "7.2"})237 wait(t, b)238 if srv.Streams() != 1 || b.Tuner() != a.Tuner() {239 t.Fatalf("%d streams after coming back, want the lingering one", srv.Streams())240 }241 b.Close()242 eventually(t, "the tuner to let go", func() bool { return srv.Streams() == 0 })243}244245// TestBusyElsewhere: a tuner someone else took is skipped from then on.246func TestBusyElsewhere(t *testing.T) {247 m, srv := setup(t)248 srv.Busy(0)249 a := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1"})250 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)251 defer cancel()252 var he *hdhr.Error253 if _, err := a.Wait(ctx); !errors.As(err, &he) || !he.Busy() {254 t.Fatalf("wait on a tuner taken elsewhere = %v", err)255 }256 eventually(t, "the failed tune to end", func() bool { m.mu.Lock(); defer m.mu.Unlock(); return len(m.muxes) == 0 })257 b := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1"})258 wait(t, b)259 if b.Tuner() != 1 {260 t.Fatalf("tuned %d, want the free one", b.Tuner())261 }262 tuners, _ := m.Tuners(ctx)263 if !tuners[0].Elsewhere || tuners[0].FrequencyHz != hdhrtest.OtherFrequency {264 t.Fatalf("tuner 0 = %+v, want it shown in use elsewhere", tuners[0])265 }266}267268// TestSlowUser: a user that stops reading is dropped, and doesn't hold up269// the others.270func TestSlowUser(t *testing.T) {271 m, _ := setup(t, 40<<20)272 slow := open(t, m, Request{FrequencyHz: rfKMGH, Program: 3, Priority: Watching, Who: "7.1"})273 fast := open(t, m, Request{FrequencyHz: rfKMGH, Program: 4, Priority: Watching, Who: "7.2"})274 wait(t, slow)275 wait(t, fast)276 var mu sync.Mutex277 n := 0278 ctx, cancel := context.WithCancel(context.Background())279 defer cancel()280 go func() {281 _ = fast.Copy(ctx, writerFunc(func(p []byte) (int, error) {282 mu.Lock()283 n += len(p)284 mu.Unlock()285 return len(p), nil286 }))287 }()288 select {289 case <-slow.Done():290 if !errors.Is(slow.Err(), errTooSlow) {291 t.Fatalf("slow user ended with %v", slow.Err())292 }293 case <-time.After(20 * time.Second):294 t.Fatal("a user that never reads wasn't dropped")295 }296 mu.Lock()297 defer mu.Unlock()298 if n == 0 || fast.Err() != nil {299 t.Fatalf("the reading user got %d bytes, ended %v", n, fast.Err())300 }301}302303type writerFunc func([]byte) (int, error)304305func (f writerFunc) Write(p []byte) (int, error) { return f(p) }