1package ts_test23import (4 "bytes"5 "encoding/json"6 "errors"7 "os"8 "os/exec"9 "path/filepath"10 "reflect"11 "slices"12 "testing"1314 "airwaves/internal/ts"15 "airwaves/internal/ts/tstest"16)1718func filter(t *testing.T, in []byte, program int) []byte {19 t.Helper()20 var out bytes.Buffer21 f := ts.NewFilter(&out, program)22 for _, p := range packets(in) {23 if err := f.Packet(p); err != nil {24 t.Fatal(err)25 }26 }27 if err := f.Flush(); err != nil {28 t.Fatal(err)29 }30 return out.Bytes()31}3233func TestFilter(t *testing.T) {34 in := denver.Packets(500)35 out := filter(t, in, 7)36 if len(out) == 0 || len(out)%ts.PacketSize != 0 {37 t.Fatalf("%d bytes out", len(out))38 }39 pkts := packets(out)40 // It starts clean: our PAT, then the PMT from its first packet.41 if ts.PID(pkts[0]) != 0 || ts.PID(pkts[1]) != 0x40 || pkts[1][1]&0x40 == 0 {42 t.Errorf("starts with PIDs %#x, %#x", ts.PID(pkts[0]), ts.PID(pkts[1]))43 }44 allowed := []int{0, 0x40, 0x41, 0x44, 0x45}45 var c ts.Counter46 for _, p := range pkts {47 if !slices.Contains(allowed, ts.PID(p)) {48 t.Fatalf("PID %#x in program 7's stream", ts.PID(p))49 }50 c.Packet(p)51 }52 if c.Errors() != 0 {53 t.Errorf("%d continuity errors in the output", c.Errors())54 }55 d := demux(out)56 progs := d.Programs()57 want, _ := demux(in).Program(7)58 if len(progs) != 1 || !reflect.DeepEqual(progs[0], want) || !d.Ready() {59 t.Errorf("output programs = %+v, want only %+v", progs, want)60 }61 if tsid, _ := d.TSID(); tsid != 467 || d.Channels() != nil {62 t.Errorf("TSID %d, channels %v", tsid, d.Channels())63 }64 // Every packet of the program's streams after the start went out.65 start := slices.IndexFunc(packets(in), func(p []byte) bool { return ts.PID(p) == 0x40 && p[1]&0x40 != 0 })66 n := 067 for _, p := range packets(in)[start:] {68 if slices.Contains(allowed[1:], ts.PID(p)) {69 n++70 }71 }72 if got := len(pkts) - countPID(pkts, 0); got != n {73 t.Errorf("%d packets of the program out, want %d", got, n)74 }75}7677func countPID(pkts [][]byte, pid int) int {78 n := 079 for _, p := range pkts {80 if ts.PID(p) == pid {81 n++82 }83 }84 return n85}8687// TestFilterWaits sends nothing until the program's PMT has been read.88func TestFilterWaits(t *testing.T) {89 var noTables []byte90 for _, p := range packets(denver.Packets(200)) {91 if pid := ts.PID(p); pid != 0 && pid != 0x30 && pid != 0x40 && pid != 0x50 && pid != 0x1ffb {92 noTables = append(noTables, p...)93 }94 }95 if out := filter(t, noTables, 6); len(out) != 0 {96 t.Errorf("%d bytes out before the tables", len(out))97 }98 if out := filter(t, denver.Packets(200), 99); len(out) != 0 {99 t.Errorf("%d bytes out for a program that isn't there", len(out))100 }101 out := filter(t, append(noTables, denver.Packets(100)...), 6)102 if len(out) == 0 || ts.PID(out) != 0 {103 t.Error("no output once the tables came")104 }105}106107// TestFilterFollows keeps up with a PMT that moves.108func TestFilterFollows(t *testing.T) {109 moved := denver110 moved.Programs = slices.Clone(denver.Programs)111 moved.Programs[0].PMTPID = 0x60112 pat := patSection(467, 1, [][2]int{{6, 0x60}, {7, 0x40}, {8, 0x50}})113 in := slices.Concat(denver.Packets(100), tstest.Packetize(0, pat), moved.Packets(100))114 out := filter(t, in, 6)115 pkts := packets(out)116 if countPID(pkts, 0x30) == 0 || countPID(pkts, 0x60) == 0 {117 t.Fatalf("PMTs on %d and %d packets", countPID(pkts, 0x30), countPID(pkts, 0x60))118 }119 // After the move, the output's PAT points at the new PMT.120 d := demux(out)121 if p, ok := d.Program(6); !ok || p.PMTPID != 0x60 || len(d.Programs()) != 1 {122 t.Errorf("program 6 = %+v", p)123 }124}125126type failing struct{ n int }127128func (f *failing) Write(p []byte) (int, error) {129 f.n++130 return 0, errors.New("pipe closed")131}132133func TestFilterWriteError(t *testing.T) {134 w := &failing{}135 f := ts.NewFilter(w, 6)136 var err error137 for _, p := range packets(denver.Packets(300)) {138 if err = f.Packet(p); err != nil {139 break140 }141 }142 if err == nil || f.Flush() == nil || f.Packet(denver.Packets(1)) == nil || w.n != 1 {143 t.Errorf("err %v after %d writes; want it kept, and no more writes", err, w.n)144 }145}146147// TestFilterFFprobe checks that ffprobe sees one program in the output.148func TestFilterFFprobe(t *testing.T) {149 ffprobe, err := exec.LookPath("ffprobe")150 if err != nil {151 t.Skip("no ffprobe")152 }153 path := filepath.Join(t.TempDir(), "out.ts")154 if err := os.WriteFile(path, filter(t, denver.Packets(2000), 6), 0o644); err != nil {155 t.Fatal(err)156 }157 raw, err := exec.Command(ffprobe, "-v", "quiet", "-show_programs", "-of", "json", path).Output()158 if err != nil {159 t.Fatalf("ffprobe: %v", err)160 }161 var probe struct {162 Programs []struct {163 ProgramNum int `json:"program_num"`164 PMTPID int `json:"pmt_pid"`165 NbStreams int `json:"nb_streams"`166 } `json:"programs"`167 }168 if err := json.Unmarshal(raw, &probe); err != nil {169 t.Fatal(err)170 }171 if len(probe.Programs) != 1 || probe.Programs[0].ProgramNum != 6 || probe.Programs[0].PMTPID != 0x30 {172 t.Errorf("ffprobe found %+v", probe.Programs)173 }174}