/
test_pipeline.py
96 lines (80 loc) · 2.91 KB
/
test_pipeline.py
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
from dataclasses import dataclass, field
from pathlib import Path
import shutil
from unittest.mock import patch
from unittest.mock import patch
from pipeline import PipelineProgress
from transcode import TranscodingProgress
from transcribe import TranscriptionProgress
import common
import pipeline
import transcode
import transcribe
import annotate
fixtures = Path("fixtures")
def test_encode_pipeline_progress():
transcription_id = "abc"
p = PipelineProgress(
state="completed",
transcription = common.Transcription(
transcription_id=transcription_id,
path="/tmp/abc",
upload=common.UploadInfo(
filename="file.name",
content_type="audio/mp3",
size_bytes=100
)
)
)
progress = common.dataclass_to_event(p)
assert transcription_id in progress
@dataclass
class MockedStub:
transcriptions: dict = field(default_factory=dict)
pipeline_stub = MockedStub()
@patch('annotate.stub', new=pipeline_stub)
@patch('common.stub', new=pipeline_stub)
@patch('pipeline.stub', new=pipeline_stub)
@patch('transcode.stub', new=pipeline_stub)
@patch('transcribe.stub', new=pipeline_stub)
def test_pipeline(transcription_id = "abc"):
with common.tmpdir_scope() as tmp:
media_path = Path(tmp)
with patch('common.db', new=common.Store(media_path)):
from_file = fixtures / "one.mp3"
to_file = media_path / transcription_id
shutil.copyfile(from_file, to_file)
# create the transcription metadata
common.db.create(
common.Transcription(
transcription_id=transcription_id,
path=to_file,
upload=common.UploadInfo(
filename="file.name",
content_type="audio/mp3",
size_bytes=15
)
)
)
updates = list(
pipeline.pipeline(
transcription_id,
"en",
media_path=Path(tmp),
local_mode=True
)
)
assert len(updates) > 1
assert type(updates[0]) == PipelineProgress
assert updates[0].state == "transcoding"
assert type(updates[1]) == TranscodingProgress
assert updates[2].percent_done == 96
assert updates[5].percent_done is 100
assert updates[5].track.duration > 1.4
assert type(updates[-3]) == TranscriptionProgress
assert updates[-3].percent_done is 100
assert updates[-3].transcript['text'].strip() == "One."
assert type(updates[-2]) == PipelineProgress
assert updates[-2].state == "annotating"
assert type(updates[-1]) == PipelineProgress
assert updates[-1].state == "completed"