| 749 | |
| 750 | |
| 751 | def open_urlresource(url, *args, **kw): |
| 752 | import urllib.request, urllib.parse |
| 753 | from .os_helper import unlink |
| 754 | try: |
| 755 | import gzip |
| 756 | except ImportError: |
| 757 | gzip = None |
| 758 | |
| 759 | check = kw.pop('check', None) |
| 760 | |
| 761 | filename = urllib.parse.urlparse(url)[2].split('/')[-1] # '/': it's URL! |
| 762 | |
| 763 | fn = os.path.join(TEST_DATA_DIR, filename) |
| 764 | |
| 765 | def check_valid_file(fn): |
| 766 | f = open(fn, *args, **kw) |
| 767 | if check is None: |
| 768 | return f |
| 769 | elif check(f): |
| 770 | f.seek(0) |
| 771 | return f |
| 772 | f.close() |
| 773 | |
| 774 | if os.path.exists(fn): |
| 775 | f = check_valid_file(fn) |
| 776 | if f is not None: |
| 777 | return f |
| 778 | unlink(fn) |
| 779 | |
| 780 | # Verify the requirement before downloading the file |
| 781 | requires('urlfetch') |
| 782 | |
| 783 | if verbose: |
| 784 | print('\tfetching %s ...' % url, file=get_original_stdout()) |
| 785 | opener = urllib.request.build_opener() |
| 786 | if gzip: |
| 787 | opener.addheaders.append(('Accept-Encoding', 'gzip')) |
| 788 | f = opener.open(url, timeout=INTERNET_TIMEOUT) |
| 789 | if gzip and f.headers.get('Content-Encoding') == 'gzip': |
| 790 | f = gzip.GzipFile(fileobj=f) |
| 791 | try: |
| 792 | with open(fn, "wb") as out: |
| 793 | s = f.read() |
| 794 | while s: |
| 795 | out.write(s) |
| 796 | s = f.read() |
| 797 | finally: |
| 798 | f.close() |
| 799 | |
| 800 | f = check_valid_file(fn) |
| 801 | if f is not None: |
| 802 | return f |
| 803 | raise TestFailed('invalid resource %r' % fn) |
| 804 | |
| 805 | |
| 806 | @contextlib.contextmanager |