Skip to content

Commit

Permalink
refine cases
Browse files Browse the repository at this point in the history
  • Loading branch information
@baiyubin2020
baiyubin2020 committed Aug 11, 2016
1 parent a3b49fa commit b64b0ba
Show file tree
Hide file tree
Showing 2 changed files with 56 additions and 54 deletions.
82 changes: 28 additions & 54 deletions tests/test_live_channel.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,50 +52,38 @@ def _assert_list_result(self,
self.assertEqual(len(result.channels), return_count)

def test_create_live_channel(self):
auth = oss2.Auth(OSS_ID, OSS_SECRET)
bucket_name = random_string(63).lower()
bucket = oss2.Bucket(auth, OSS_ENDPOINT, bucket_name)
bucket.create_bucket(oss2.BUCKET_ACL_PRIVATE)

channel_name = 'test-create-channel'
playlist_name = 'test.m3u8'

channel_target = oss2.models.LiveChannelInfoTarget(playlist_name = playlist_name)
channel_info = oss2.models.LiveChannelInfo(target = channel_target)
create_result = bucket.create_live_channel(channel_name, channel_info)
create_result = self.bucket.create_live_channel(channel_name, channel_info)

self.assertEqual(create_result.play_url,
self._get_play_url(bucket_name, channel_name, playlist_name))
self._get_play_url(self.bucket.bucket_name, channel_name, playlist_name))
self.assertEqual(create_result.publish_url,
self._get_publish_url(bucket_name, channel_name))
self._get_publish_url(self.bucket.bucket_name, channel_name))

bucket.delete_live_channel(channel_name)
bucket.delete_bucket()
self.assertRaises(oss2.exceptions.NoSuchBucket, bucket.delete_bucket)
self.bucket.delete_live_channel(channel_name)

def test_get_live_channel(self):
auth = oss2.Auth(OSS_ID, OSS_SECRET)
bucket = oss2.Bucket(auth, OSS_ENDPOINT, random_string(63).lower())
bucket.create_bucket(oss2.BUCKET_ACL_PRIVATE)

channel_name = 'test-get-channel'

self.assertRaises(NoSuchLiveChannel, bucket.get_live_channel, channel_name)
self.assertRaises(NoSuchLiveChannel, self.bucket.get_live_channel, channel_name)

channel_target = oss2.models.LiveChannelInfoTarget(playlist_name = 'test.m3u8')
channel_info = oss2.models.LiveChannelInfo(target = channel_target)
bucket.create_live_channel(channel_name, channel_info)
self.bucket.create_live_channel(channel_name, channel_info)

get_result = bucket.get_live_channel(channel_name)
get_result = self.bucket.get_live_channel(channel_name)
self.assertEqual(get_result.description, channel_info.description)
self.assertEqual(get_result.status, channel_info.status)
self.assertEqual(get_result.target.type, channel_target.type)
self.assertEqual(get_result.target.frag_duration, str(channel_target.frag_duration))
self.assertEqual(get_result.target.frag_count, str(channel_target.frag_count))
self.assertEqual(get_result.target.playlist_name, channel_target.playlist_name)

bucket.delete_live_channel(channel_name)
bucket.delete_bucket()
self.assertRaises(oss2.exceptions.NoSuchBucket, bucket.delete_bucket)
self.bucket.delete_live_channel(channel_name)

def test_list_live_channel(self):
auth = oss2.Auth(OSS_ID, OSS_SECRET)
Expand Down Expand Up @@ -170,52 +158,43 @@ def test_list_live_channel(self):
self.assertRaises(oss2.exceptions.NoSuchBucket, bucket.delete_bucket)

def test_get_live_channel_stat(self):
auth = oss2.Auth(OSS_ID, OSS_SECRET)
bucket = oss2.Bucket(auth, OSS_ENDPOINT, random_string(63).lower())
bucket.create_bucket(oss2.BUCKET_ACL_PRIVATE)

channel_name = 'test-get-channel-stat'
_playlist_name = 'test.m3u8'
channel_target = oss2.models.LiveChannelInfoTarget(playlist_name = _playlist_name)
playlist_name = 'test.m3u8'

channel_target = oss2.models.LiveChannelInfoTarget(playlist_name = playlist_name)
channel_info = oss2.models.LiveChannelInfo(target = channel_target)
bucket.create_live_channel(channel_name, channel_info)
self.bucket.create_live_channel(channel_name, channel_info)

get_stat_result = bucket.get_live_channel_stat(channel_name)
get_stat_result = self.bucket.get_live_channel_stat(channel_name)
self.assertEqual(get_stat_result.status, 'Idle')

bucket.delete_live_channel(channel_name)
bucket.delete_bucket()
self.assertRaises(oss2.exceptions.NoSuchBucket, bucket.delete_bucket)
self.bucket.delete_live_channel(channel_name)

def test_put_live_channel_status(self):
auth = oss2.Auth(OSS_ID, OSS_SECRET)
bucket = oss2.Bucket(auth, OSS_ENDPOINT, random_string(63).lower())
bucket.create_bucket(oss2.BUCKET_ACL_PRIVATE)

channel_name = 'test-put-channel-status'

channel_target = oss2.models.LiveChannelInfoTarget()
channel_info = oss2.models.LiveChannelInfo(target = channel_target)
bucket.create_live_channel(channel_name, channel_info)
self.bucket.create_live_channel(channel_name, channel_info)

get_result = bucket.get_live_channel(channel_name)
get_result = self.bucket.get_live_channel(channel_name)
self.assertEqual(get_result.status, 'enabled')

bucket.put_live_channel_status(channel_name, 'disabled')
self.bucket.put_live_channel_status(channel_name, 'disabled')

get_result = bucket.get_live_channel(channel_name)
get_result = self.bucket.get_live_channel(channel_name)
self.assertEqual(get_result.status, 'disabled')

bucket.put_live_channel_status(channel_name, 'enabled')
self.bucket.put_live_channel_status(channel_name, 'enabled')

get_result = bucket.get_live_channel(channel_name)
get_result = self.bucket.get_live_channel(channel_name)
self.assertEqual(get_result.status, 'enabled')

bucket.delete_live_channel(channel_name)
bucket.delete_bucket()
self.assertRaises(oss2.exceptions.NoSuchBucket, bucket.delete_bucket)
self.bucket.delete_live_channel(channel_name)

def test_get_live_channel_history(self):
channel_name = 'test-get-channel-history'

channel_target = oss2.models.LiveChannelInfoTarget()
channel_info = oss2.models.LiveChannelInfo(target = channel_target)
self.bucket.create_live_channel(channel_name, channel_info)
Expand All @@ -226,14 +205,11 @@ def test_get_live_channel_history(self):
self.bucket.delete_live_channel(channel_name)

def test_post_vod_playlist(self):
auth = oss2.Auth(OSS_ID, OSS_SECRET)
bucket = oss2.Bucket(auth, OSS_ENDPOINT, random_string(63).lower())
bucket.create_bucket(oss2.BUCKET_ACL_PRIVATE)

channel_name = 'test-post-vod-playlist'

channel_target = oss2.models.LiveChannelInfoTarget()
channel_info = oss2.models.LiveChannelInfo(target = channel_target)
bucket.create_live_channel(channel_name, channel_info)
self.bucket.create_live_channel(channel_name, channel_info)

# push rtmp stream here, then generate some ts files on oss.

Expand All @@ -243,15 +219,13 @@ def test_post_vod_playlist(self):

# throw exception because no ts file been generated.
self.assertRaises(oss2.exceptions.InvalidArgument,
bucket.post_vod_playlist,
self.bucket.post_vod_playlist,
channel_name,
playlist_name,
start_time = start_time,
end_time = end_time)

bucket.delete_live_channel(channel_name)
bucket.delete_bucket()
self.assertRaises(oss2.exceptions.NoSuchBucket, bucket.delete_bucket)
self.bucket.delete_live_channel(channel_name)

def test_sign_rtmp_url(self):
channel_name = 'test-sign-rtmp-url'
Expand Down
28 changes: 28 additions & 0 deletions tests/test_mock_object.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -501,6 +501,34 @@ def test_get_acl(self, do_request):
result = unittests.common.bucket().get_object_acl('fake-key')
self.assertEqual(result.acl, expected)

# for ci
def test_oss_utils_negative(self):
try:
oss2.utils.makedir_p('/')
self.assertTrue(False)
except:
pass

try:
oss2.utils.silently_remove('/')
self.assertTrue(False)
except:
pass

try:
oss2.utils.force_rename('/', '/')
self.assertTrue(False)
except:
pass

oss2.utils.makedir_p('xyz')
oss2.utils.makedir_p('zyz')
try:
oss2.utils.force_rename('xyz', 'zyx')
self.assertTrue(False)
except:
pass


if __name__ == '__main__':
unittest.main()

0 comments on commit b64b0ba

Please sign in to comment.