4 # Copyright 2016 RIFT.IO Inc
6 # Licensed under the Apache License, Version 2.0 (the "License");
7 # you may not use this file except in compliance with the License.
8 # You may obtain a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS,
14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
28 import rift
.downloader
as downloader
30 TEST_URL
= "https://raw.githubusercontent.com/RIFTIO/RIFT.ware/master/rift-shell"
32 class TestCase(unittest
.TestCase
):
36 def _common_checks(self
, job
):
37 if job
.status
!= "COMPLETED":
40 # assert job.bytes_downloaded == job.bytes_total
41 assert job
.stop_time
> 0
42 assert job
.start_time
> 0
43 assert job
.stop_time
>= job
.start_time
45 def test_file_download(self
):
48 1. Successful download
49 2. Model attributes (Process percent, detail, status)
51 url_downl
= downloader
.UrlDownloader(TEST_URL
)
53 assert os
.path
.isfile(url_downl
.filename
)
56 assert url_downl
.meta
.status
== downloader
.DownloadStatus
.COMPLETED
57 # assert url_downl.job.progress_percent == 100
58 assert "success" in url_downl
.meta
.detail
59 self
._common
_checks
(url_downl
.meta
)
61 def test_file_not_exists(self
):
64 1. 404 download with retries
65 2. Model attributes (Process percent, detail, status)
67 url_downl
= downloader
.UrlDownloader(TEST_URL
+ ".blah")
70 assert not os
.path
.isfile(url_downl
.filename
)
71 assert url_downl
.meta
.status
== downloader
.DownloadStatus
.FAILED
72 assert "Max retries" in url_downl
.meta
.detail
or "404" in url_downl
.meta
.detail
74 self
._common
_checks
(url_downl
.meta
)
76 def test_cancellation(self
):
79 1. Cancel for a download and clean up of the downloaded file.
80 2. Model attributes (Process percent, detail, status)
82 url
= "http://speedtest.ftp.otenet.gr/files/test1Mb.db"
83 url_dwld
= downloader
.UrlDownloader(url
)
84 loop
= asyncio
.get_event_loop()
85 fut
= loop
.run_in_executor(None, url_dwld
.download
)
89 url_dwld
.cancel_download()
93 yield from asyncio
.sleep(2)
95 yield from asyncio
.sleep(2)
97 loop
.run_until_complete(sleep())
99 assert url_dwld
.meta
.status
== downloader
.DownloadStatus
.CANCELLED
100 assert url_dwld
.meta
.bytes_downloaded
== url_dwld
.meta
.bytes_downloaded
101 assert "cancel" in url_dwld
.meta
.detail
102 self
._common
_checks
(url_dwld
.meta
)
104 def test_auth_url(self
):
105 url_downl
= downloader
.UrlDownloader(
106 'https://api.github.com/user')
115 def main(argv
=sys
.argv
[1:]):
116 logging
.basicConfig(format
='TEST %(message)s')
118 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
119 parser
= argparse
.ArgumentParser()
120 parser
.add_argument('-v', '--verbose', action
='store_true')
121 parser
.add_argument('-n', '--no-runner', action
='store_true')
123 args
, unknown
= parser
.parse_known_args(argv
)
127 # Set the global logging level
128 logging
.getLogger().setLevel(logging
.DEBUG
if args
.verbose
else logging
.ERROR
)
130 # The unittest framework requires a program name, so use the name of this
131 # file instead (we do not want to have to pass a fake program name to main
132 # when this is called from the interpreter).
133 unittest
.main(argv
=[__file__
] + unknown
+ ["-v"], testRunner
=runner
)
135 if __name__
== '__main__':