-
Notifications
You must be signed in to change notification settings - Fork 1
/
rtsp_test.go
128 lines (111 loc) · 2.71 KB
/
rtsp_test.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
package rtsp
import (
"bufio"
"bytes"
"os"
"path/filepath"
"testing"
"gotest.tools/v3/assert"
)
func TestResponse(t *testing.T) {
files, err := filepath.Glob("testdata/*.response")
assert.NilError(t, err)
for _, file := range files {
t.Run(file, func(t *testing.T) {
// decode it
f, err := os.Open(file)
assert.NilError(t, err)
defer f.Close()
res, err := ReadResponse(bufio.NewReader(f))
assert.NilError(t, err)
// encode it
var buf bytes.Buffer
err = res.Write(&buf)
assert.NilError(t, err)
// decode it again
res2, err := ReadResponse(bufio.NewReader(&buf))
assert.NilError(t, err)
// compare to original
assert.DeepEqual(t, res, res2)
})
}
}
func TestRequest(t *testing.T) {
files, err := filepath.Glob("testdata/*.request")
assert.NilError(t, err)
for _, file := range files {
t.Run(file, func(t *testing.T) {
// decode it
f, err := os.Open(file)
assert.NilError(t, err)
defer f.Close()
req, err := ReadRequest(bufio.NewReader(f))
assert.NilError(t, err)
// encode it
var buf bytes.Buffer
err = req.Write(&buf)
assert.NilError(t, err)
// decode it again
req2, err := ReadRequest(bufio.NewReader(&buf))
assert.NilError(t, err)
// compare to original
assert.DeepEqual(t, req, req2)
})
}
}
func TestFrame(t *testing.T) {
f := Frame{
Channel: 1,
Data: []byte("hello world"),
}
// encode it
var buf bytes.Buffer
err := f.Write(&buf)
assert.NilError(t, err)
// make sure it's recognized as a frame
br := bufio.NewReader(&buf)
ok, err := IsFrame(br)
assert.NilError(t, err)
assert.Assert(t, ok)
// decode it
f2, err := ReadFrame(br)
assert.NilError(t, err)
assert.DeepEqual(t, f, f2)
}
func TestNewResponse(t *testing.T) {
res, err := NewResponse(StatusNotFound, []byte("Hello world"))
assert.NilError(t, err)
res.Header.Set("CSeq", "100")
res.Header.Set("foo", "bar")
// encode it
var buf bytes.Buffer
err = res.Write(&buf)
assert.NilError(t, err)
// decode it
res2, err := ReadResponse(bufio.NewReader(&buf))
assert.NilError(t, err)
assert.DeepEqual(t, res, res2)
}
func TestNewRequest(t *testing.T) {
req, err := NewRequest(MethodOptions, "rtsp://someurl", []byte("what"))
assert.NilError(t, err)
req.Header.Set("CSeq", "1")
req.Header.Set("Authorize", "secret")
// encode it
var buf bytes.Buffer
err = req.Write(&buf)
assert.NilError(t, err)
// decode it
req2, err := ReadRequest(bufio.NewReader(&buf))
assert.NilError(t, err)
assert.DeepEqual(t, req, req2)
}
func TestHeaderClone(t *testing.T) {
h := Header{}
h.Set("foo", "bar")
h.Add("foo", "baz")
h2 := h.Clone()
h2.Set("foo", "test")
assert.DeepEqual(t, h["foo"], []string{"bar", "baz"})
assert.DeepEqual(t, h2["foo"], []string{"test"})
}