dtas.git  about / heads / tags
duct tape audio suite for *nix
blob 09eceee1f59c664ed1ada839c7b7127ddeee705a 6630 bytes (raw)
$ git show HEAD:test/test_player_integration.rb	# shows this blob on the CLI

  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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
 
# Copyright (C) all contributors <dtas-all@nongnu.org>
# License: GPL-3.0+ <https://www.gnu.org/licenses/gpl-3.0.txt>
# frozen_string_literal: true
require './test/player_integration'
class TestPlayerIntegration < Testcase
  include PlayerIntegration

  def test_cmd_rate
    env = ENV.to_hash.merge(@fmt.to_env)
    cmd = "sox -n $SOXFMT - synth 3 pinknoise | #@cmd"
    pid = spawn(env, cmd)
    t = Time.now
    _, _ = Process.waitpid2(pid)
    elapsed = Time.now - t
    assert_in_delta 3.0, elapsed, 0.5
  end if ENV["MATH_IS_HARD"] # ensure our @cmd timing is accurate

  def test_sink_close_after_play
    s = client_socket
    @cmd = "cat >/dev/null"
    default_pid = default_sink_pid(s)
    Tempfile.open('junk') do |junk|
      pink = "sox -n $SOXFMT - synth 0.0001 pinknoise | tee -i #{junk.path}"
      s.req_ok("enq-cmd \"#{pink}\"")
      wait_files_not_empty(junk)
    end
    wait_files_not_empty(default_pid)
    pid = read_pid_file(default_pid)
    wait_pid_dead(pid)
  end

  def test_sink_killed_during_play
    s = client_socket
    default_pid = default_sink_pid(s)
    cmd = Tempfile.new(%w(sox-cmd .pid))
    pink = "echo $$ > #{cmd.path}; sox -n $SOXFMT - synth 100 pinknoise"
    s.req_ok("enq-cmd \"#{pink}\"")
    wait_files_not_empty(cmd, default_pid)
    pid = read_pid_file(default_pid)
    Process.kill(:KILL, pid)
    cmd_pid = read_pid_file(cmd)
    wait_pid_dead(cmd_pid)
  end

  def test_sink_activate
    s = client_socket
    res = s.req("sink ls")
    assert_equal "default", res

    # setup two outputs

    # make the default sink trickle
    default_pid = Tempfile.new(%w(dtas-test .pid))
    pf = "echo $$ >> #{default_pid.path}; "
    s.req_ok("sink ed default command='#{pf}#@cmd'")

    # make a sleepy sink trickle, too
    sleepy_pid = Tempfile.new(%w(dtas-test .pid))
    pf = "echo $$ >> #{sleepy_pid.path};"
    s.req_ok("sink ed sleepy command='#{pf}#@cmd' active=true")

    # ensure both sinks were created
    res = s.req("sink ls")
    assert_equal "default sleepy", res

    # generate pinknoise
    pinknoise = "sox -n -r 44100 -c 2 -t s32 - synth 0 pinknoise"
    s.req_ok("enq-cmd \"#{pinknoise}\"")

    # wait for sinks to start
    wait_files_not_empty(sleepy_pid, default_pid)

    # deactivate sleepy sink and ensure it's gone
    sleepy = File.read(sleepy_pid).to_i
    assert_operator sleepy, :>, 0
    Process.kill(0, sleepy)
    s.req_ok("sink ed sleepy active=false")
    wait_pid_dead(sleepy)

    # ensure default sink is still alive
    default = File.read(default_pid).to_i
    assert_operator default, :>, 0
    Process.kill(0, default)

    # restart sleepy sink
    sleepy_pid.sync = true
    sleepy_pid.seek(0)
    sleepy_pid.truncate(0)
    s.req_ok("sink ed sleepy active=true")

    # wait for sleepy sink
    wait_files_not_empty(sleepy_pid)

    # check sleepy restarted
    sleepy = File.read(sleepy_pid).to_i
    assert_operator sleepy, :>, 0
    Process.kill(0, sleepy)

    # stop playing current track
    s.req_ok("skip")

    wait_pid_dead(sleepy)
    wait_pid_dead(default)
  end

  def test_env_change
    s = client_socket
    tmp = Tempfile.new(%w(env .txt))
    s.req_ok("sink ed default active=true command='cat >/dev/null'")

    s.req_ok("env FOO=BAR")
    s.req_ok(["enq-cmd", "echo $FOO | tee #{tmp.path}"])
    wait_files_not_empty(tmp)
    assert_equal "BAR\n", tmp.read

    tmp.rewind
    tmp.truncate(0)
    s.req_ok("env FOO#")
    s.req_ok(["enq-cmd", "echo -$FOO- | tee #{tmp.path}"])
    wait_files_not_empty(tmp)
    assert_equal "--\n", tmp.read
  end

  def test_sink_env
    s = client_socket
    tmp = Tempfile.new(%w(env .txt))
    s.req_ok("sink ed default active=true " \
             "command='echo -$FOO- > #{tmp.path}; cat >/dev/null'")

    s.req_ok("sink ed default env.FOO=BAR")
    s.req_ok(["enq-cmd", "echo HI"])
    wait_files_not_empty(tmp)
    assert_equal "-BAR-\n", tmp.read

    tmp.rewind
    tmp.truncate(0)
    s.req_ok("sink ed default env#FOO")

    Timeout.timeout(5) do
      begin
        yaml = s.req("current")
        cur = DTAS.yaml_load(yaml)
      end while cur["sinks"] && sleep(0.01)
    end

    s.req_ok(["enq-cmd", "echo HI"])
    wait_files_not_empty(tmp)
    assert_equal "--\n", tmp.read
  end

  def test_enq_head
    s = client_socket
    default_sink_pid(s)
    dump = Tempfile.new(%W(d .sox))
    s.req_ok "sink ed dump active=true command='sox $SOXFMT - #{dump.path}'"
    noise, len = tmp_noise
    s.req_ok("enq-head #{noise.path}")
    s.req_ok("enq-head #{noise.path} 4")
    s.req_ok("enq-head #{noise.path} 3")
    dethrottle_decoder(s)
    expect = Tempfile.new(%W(expect .sox))

    c = "sox #{noise.path} -t sox '|sox #{noise.path} -p trim 3' " \
            "-t sox '|sox #{noise.path} -p trim 4' #{expect.path}"
    assert system(c)
    Timeout.timeout(len) do
      begin
        yaml = s.req("current")
        cur = DTAS.yaml_load(yaml)
      end while cur["sinks"] && sleep(0.01)
    end
    assert(system("cmp", dump.path, expect.path),
           "files don't match #{dump.path} != #{expect.path}")
  end

  def test_cd_pwd
    s = client_socket
    pwd = Dir.pwd

    assert_equal pwd, s.req("pwd")

    s.req_ok("cd /")

    assert_equal "/", s.req("pwd")

    err = s.req("cd /this-better-be-totally-non-existent-on-any-system-#{rand}")
    assert_match(%r{\AERR }, err, err)

    assert_equal "/", s.req("pwd")
  end

  def test_state_file
    state = Tempfile.new(%w(state .yml))
    state_path = state.path
    state.close!
    s = client_socket
    s.req_ok(%W(state dump #{state_path}))
    hash = DTAS.yaml_load(IO.binread(state_path))
    assert_equal @sock_path, hash["socket"]
    assert_equal "default", hash["sinks"][0]["name"]

    assert_equal "", IO.binread(@state_tmp.path)
    s.req_ok(%W(state dump))
    orig = DTAS.yaml_load(IO.binread(@state_tmp.path))
    assert_equal orig, hash
  ensure
    File.unlink(state_path)
  end

  def test_source_ed
    s = client_socket
    assert_equal "sox ff av splitfx", s.req("source ls")
    s.req_ok("source ed av tryorder=-1")
    assert_equal "av sox ff splitfx", s.req("source ls")
    s.req_ok("source ed av tryorder=")
    assert_equal "sox ff av splitfx", s.req("source ls")

    s.req_ok("source ed sox command=true")
    sox = DTAS.yaml_load(s.req("source cat sox"))
    assert_equal "true", sox["command"]

    s.req_ok("source ed sox command=")
    sox = DTAS.yaml_load(s.req("source cat sox"))
    assert_equal DTAS::Source::Sox::SOX_DEFAULTS["command"], sox["command"]
  end

  def test_tl_tracks
    s = client_socket
    assert_match(%r{\A\d+ }, s.req('tl tracks'))
  end
end

git clone git://80x24.org/dtas.git
git clone https://80x24.org/dtas.git