-
Notifications
You must be signed in to change notification settings - Fork 13
/
exercise-assets.py
145 lines (128 loc) · 7.2 KB
/
exercise-assets.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
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
# coding: utf-8
import os
import sys
import time
import mux_python
from mux_python.rest import NotFoundException
import logger
# Exercises all asset operations.
# Authentication Setup
configuration = mux_python.Configuration()
configuration.username = os.environ['MUX_TOKEN_ID']
configuration.password = os.environ['MUX_TOKEN_SECRET']
# API Client Initialization
assets_api = mux_python.AssetsApi(mux_python.ApiClient(configuration))
playback_ids_api = mux_python.PlaybackIDApi(mux_python.ApiClient(configuration))
# ========== create-asset ==========
add_captions = mux_python.CreateTrackRequest(url="https://tears-of-steel-subtitles.s3.amazonaws.com/tears-fr.vtt", type="text", text_type="subtitles", language_code="fr", closed_captions=False, name="French")
input_settings = [mux_python.InputSettings(url='https://storage.googleapis.com/muxdemofiles/mux-video-intro.mp4'), add_captions]
create_asset_request = mux_python.CreateAssetRequest(input=input_settings)
create_asset_response = assets_api.create_asset(create_asset_request)
logger.print_debug("Created Asset: " + str(create_asset_response))
assert create_asset_response != None
assert create_asset_response.data.id != None
print("create-asset OK ✅")
# ========== list-assets ==========
list_assets_response = assets_api.list_assets()
logger.print_debug ("Got List Asset Response: " + str(list_assets_response))
assert list_assets_response != None
assert list_assets_response.data[0].id == create_asset_response.data.id
print("list-assets OK ✅")
# Wait for the asset to become ready...
if create_asset_response.data.status != 'ready':
print(" waiting for asset to become ready...")
while True:
# ========== get-asset ==========
asset_response = assets_api.get_asset(create_asset_response.data.id)
assert asset_response != None
assert asset_response.data.id == create_asset_response.data.id
if asset_response.data.status != 'ready':
logger.print_debug("Asset still not ready. Status was: " + asset_response.data.status)
time.sleep(1)
else:
# ========== get-asset-input-info ==========
logger.print_debug("Asset Ready. Checking input info.")
get_asset_input_info_response = assets_api.get_asset_input_info(create_asset_response.data.id)
logger.print_debug("Got Asset Input Info: " + str(get_asset_input_info_response))
assert get_asset_input_info_response != None
assert get_asset_input_info_response.data != None
break
print("get-asset OK ✅")
print("get-asset-input-info OK ✅")
# ========== clipping ==========
clip_input_settings = [mux_python.InputSettings(url='mux://assets/' + create_asset_response.data.id, start_time=0, end_time=5)]
clip_request = mux_python.CreateAssetRequest(input=clip_input_settings)
clip_response = assets_api.create_asset(clip_request)
assert clip_response != None
assert clip_response.data.id != None
print("clipping OK ✅")
# ========== create-asset-playback-id ==========
create_playback_id_request = mux_python.CreatePlaybackIDRequest(policy=mux_python.PlaybackPolicy.PUBLIC)
create_asset_playback_id_response = assets_api.create_asset_playback_id(create_asset_response.data.id, create_playback_id_request)
logger.print_debug("Added Playback ID: " + str(create_asset_playback_id_response))
assert create_asset_playback_id_response != None
assert create_asset_playback_id_response.data != None
print("create-asset-playback-id OK ✅")
# ========== get-asset-playback-id ==========
playback_id_response = assets_api.get_asset_playback_id(create_asset_response.data.id, create_asset_playback_id_response.data.id)
logger.print_debug("Got Playback ID: " + str(playback_id_response))
assert playback_id_response != None
assert playback_id_response.data != None
assert playback_id_response.data.id == create_asset_playback_id_response.data.id
print("get-asset-playback-id OK ✅")
# ========== get-asset-or-livestream-id ==========
playback_id = playback_id_response.data.id
get_playback_id_resp = playback_ids_api.get_asset_or_livestream_id(playback_id).data
assert get_playback_id_resp.object.id == create_asset_response.data.id
assert get_playback_id_resp.object.type == "asset"
print("get-asset-or-livestream-id OK ✅")
# ========== update-asset-mp4-support ==========
update_asset_mp4_support_request = mux_python.UpdateAssetMP4SupportRequest(mp4_support="standard")
asset_response_mp4 = assets_api.update_asset_mp4_support(create_asset_response.data.id, update_asset_mp4_support_request)
logger.print_debug("Added MP4 support: " + str(asset_response_mp4))
assert asset_response_mp4 != None
assert asset_response_mp4.data != None
assert asset_response_mp4.data.id == create_asset_response.data.id
assert asset_response_mp4.data.mp4_support == "standard"
print("update-asset-mp4-support OK ✅")
# ========== update-asset-master-access ==========
update_asset_master_access_request = mux_python.UpdateAssetMasterAccessRequest(master_access="temporary")
asset_response_master = assets_api.update_asset_master_access(create_asset_response.data.id, update_asset_master_access_request)
logger.print_debug("Added Master Acccess: " + str(asset_response_master))
assert asset_response_master != None
assert asset_response_master.data != None
assert asset_response_master.data.id == create_asset_response.data.id
assert asset_response_master.data.master_access == "temporary"
print("update-asset-master-access OK ✅")
# ========== create-asset-track ==========
add_captions = mux_python.CreateTrackRequest(url="https://tears-of-steel-subtitles.s3.amazonaws.com/tears-en.vtt", type="text", text_type="subtitles", language_code="en", closed_captions=False, name="English")
caption_track = assets_api.create_asset_track(create_asset_response.data.id, add_captions)
assert caption_track != None
assert caption_track.data != None
assert caption_track.data.id != None
assert caption_track.data.name == 'English'
asset_with_2_captions = assets_api.get_asset(create_asset_response.data.id)
assert len(asset_with_2_captions.data.tracks) == 4 # Audio, Video, French that we ingested with the asset, and the English we added here!
print("create-asset-track OK ✅")
# ========== delete-asset-track ==========
time.sleep(5)
assets_api.delete_asset_track(create_asset_response.data.id, caption_track.data.id)
asset_no_captions = assets_api.get_asset(create_asset_response.data.id)
assert len(asset_no_captions.data.tracks) == 3 # Audio, Video, French that we ingested with the asset
print("delete-asset-track OK ✅")
# ========== delete-asset-playback-id ==========
assets_api.delete_asset_playback_id(create_asset_response.data.id, create_asset_playback_id_response.data.id)
logger.print_debug("Deleted Playback ID (void response) ")
deleted_playback_id_get_asset_response = assets_api.get_asset(create_asset_response.data.id)
assert deleted_playback_id_get_asset_response.data.playback_ids == None
print("delete-asset-playback-id OK ✅")
# ========== delete-asset ==========
assets_api.delete_asset(create_asset_response.data.id)
logger.print_debug("Deleted Asset (void response) ")
try:
assets_api.get_asset(create_asset_response.data.id)
print("Should have 404'd when getting deleted asset ❌ ")
sys.exit(1)
except NotFoundException as e:
assert e != None
print("delete-asset OK ✅")