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 UrlTestCase(unittest
.TestCase
):
37 def set_logger(cls
, log
):
40 def _common_checks(self
, job
):
41 if job
.status
!= "COMPLETED":
44 # assert job.bytes_downloaded == job.bytes_total
45 assert job
.stop_time
> 0
46 assert job
.start_time
> 0
47 assert job
.stop_time
>= job
.start_time
49 def _display_result(self
, url_downl
):
50 UrlTestCase
.log
.debug("URL download result: {}".format(url_downl
))
52 def test_file_download(self
):
55 1. Successful download
56 2. Model attributes (Process percent, detail, status)
58 url_downl
= downloader
.UrlDownloader(TEST_URL
)
60 assert os
.path
.isfile(url_downl
.filepath
)
62 self
._display
_result
(url_downl
)
63 assert url_downl
.meta
.status
== downloader
.DownloadStatus
.COMPLETED
64 # assert url_downl.job.progress_percent == 100
65 assert "success" in url_downl
.meta
.detail
66 self
._common
_checks
(url_downl
.meta
)
68 def test_file_not_exists(self
):
71 1. 404 download with retries
72 2. Model attributes (Process percent, detail, status)
74 url_downl
= downloader
.UrlDownloader(TEST_URL
+ ".blah")
77 self
._display
_result
(url_downl
)
78 assert not os
.path
.isfile(url_downl
.filepath
)
79 assert url_downl
.meta
.status
== downloader
.DownloadStatus
.FAILED
80 assert "Max retries" in url_downl
.meta
.detail
or "404" in url_downl
.meta
.detail
82 self
._common
_checks
(url_downl
.meta
)
84 def test_cancellation(self
):
87 1. Cancel for a download and clean up of the downloaded file.
88 2. Model attributes (Process percent, detail, status)
90 url
= "http://speedtest.ftp.otenet.gr/files/test10Mb.db"
91 url_dwld
= downloader
.UrlDownloader(url
)
92 loop
= asyncio
.get_event_loop()
93 fut
= loop
.run_in_executor(None, url_dwld
.download
)
97 url_dwld
.cancel_download()
101 yield from asyncio
.sleep(2)
103 yield from asyncio
.sleep(2)
105 loop
.run_until_complete(sleep())
107 self
._display
_result
(url_dwld
)
108 assert url_dwld
.meta
.status
== downloader
.DownloadStatus
.CANCELLED
109 assert url_dwld
.meta
.bytes_downloaded
== url_dwld
.meta
.bytes_downloaded
110 assert "cancel" in url_dwld
.meta
.detail
111 self
._common
_checks
(url_dwld
.meta
)
113 def test_auth_url(self
):
114 url_downl
= downloader
.UrlDownloader(
115 'https://api.github.com/user')
118 self
._display
_result
(url_downl
)
125 def main(argv
=sys
.argv
[1:]):
126 logging
.basicConfig(format
='TEST %(message)s')
128 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
129 parser
= argparse
.ArgumentParser()
130 parser
.add_argument('-v', '--verbose', action
='store_true')
131 parser
.add_argument('-n', '--no-runner', action
='store_true')
133 args
, unknown
= parser
.parse_known_args(argv
)
137 # Set the global logging level
138 log
= logging
.getLogger()
139 log
.setLevel(logging
.DEBUG
if args
.verbose
else logging
.ERROR
)
140 UrlTestCase
.set_logger(log
)
142 # The unittest framework requires a program name, so use the name of this
143 # file instead (we do not want to have to pass a fake program name to main
144 # when this is called from the interpreter).
145 unittest
.main(argv
=[__file__
] + unknown
+ ["-v"], testRunner
=runner
)
147 if __name__
== '__main__':