<s> """<STR_LIT>""" <EOL> import logging <EOL> import os . path as op <EOL> from itertools import product <EOL> import numpy as np <EOL> from numpy . testing import assert_array_equal as ae <EOL> import responses <EOL> from pytest import raises , yield_fixture <EOL> from . . datasets import ( download_file , <EOL> download_test_data , <EOL> download_sample_data , <EOL> _check_md5_of_url , <EOL> _BASE_URL , <EOL> _validate_output_dir , <EOL> ) <EOL> from phy . utils . testing import captured_logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> _URL = '<STR_LIT>' <EOL> _DATA = np . linspace ( <NUM_LIT:0.> , <NUM_LIT:1.> , <NUM_LIT> ) . astype ( np . float32 ) <EOL> _CHECKSUM = '<STR_LIT>' <EOL> def _add_mock_response ( url , body , file_type = '<STR_LIT>' ) : <EOL> content_type = ( '<STR_LIT>' <EOL> if file_type == '<STR_LIT>' else '<STR_LIT>' ) <EOL> responses . add ( responses . GET , url , <EOL> body = body , <EOL> status = <NUM_LIT:200> , <EOL> content_type = content_type , <EOL> ) <EOL> @ yield_fixture <EOL> def mock_url ( ) : <EOL> _add_mock_response ( _URL , _DATA . tostring ( ) ) <EOL> _add_mock_response ( _URL + '<STR_LIT>' , _CHECKSUM + '<STR_LIT:U+0020>' + op . basename ( _URL ) ) <EOL> yield _URL <EOL> responses . reset ( ) <EOL> @ yield_fixture ( params = product ( ( True , False ) , repeat = <NUM_LIT:4> ) ) <EOL> def mock_urls ( request ) : <EOL> data = _DATA . tostring ( ) <EOL> checksum = _CHECKSUM <EOL> url_data = _URL <EOL> url_checksum = _URL + '<STR_LIT>' <EOL> if not request . param [ <NUM_LIT:0> ] : <EOL> url_data = url_data [ : - <NUM_LIT:1> ] <EOL> if not request . param [ <NUM_LIT:1> ] : <EOL> data = data [ : - <NUM_LIT:1> ] <EOL> if not request . param [ <NUM_LIT:2> ] : <EOL> url_checksum = url_checksum [ : - <NUM_LIT:1> ] <EOL> if not request . param [ <NUM_LIT:3> ] : <EOL> checksum = checksum [ : - <NUM_LIT:1> ] <EOL> _add_mock_response ( url_data , data ) <EOL> _add_mock_response ( url_checksum , checksum ) <EOL> yield request . param , url_data , url_checksum <EOL> responses . reset ( ) <EOL> def _dl ( path ) : <EOL> assert path <EOL> download_file ( _URL , path ) <EOL> with open ( path , '<STR_LIT:rb>' ) as f : <EOL> data = f . read ( ) <EOL> return data <EOL> def _check ( data ) : <EOL> ae ( np . fromstring ( data , np . float32 ) , _DATA ) <EOL> def test_validate_output_dir ( chdir_tempdir ) : <EOL> _validate_output_dir ( None ) <EOL> _validate_output_dir ( op . join ( chdir_tempdir , '<STR_LIT>' ) ) <EOL> assert op . exists ( op . join ( chdir_tempdir , '<STR_LIT>' ) ) <EOL> @ responses . activate <EOL> def test_check_md5_of_url ( tempdir , mock_url ) : <EOL> output_path = op . join ( tempdir , '<STR_LIT:data>' ) <EOL> download_file ( _URL , output_path ) <EOL> assert _check_md5_of_url ( output_path , _URL ) <EOL> @ responses . activate <EOL> def test_download_not_found ( tempdir ) : <EOL> path = op . join ( tempdir , '<STR_LIT:test>' ) <EOL> with raises ( Exception ) : <EOL> download_file ( _URL + '<STR_LIT>' , path ) <EOL> @ responses . activate <EOL> def test_download_already_exists_invalid ( tempdir , mock_url ) : <EOL> with captured_logging ( ) as buf : <EOL> path = op . join ( tempdir , '<STR_LIT:test>' ) <EOL> open ( path , '<STR_LIT:a>' ) . close ( ) <EOL> _check ( _dl ( path ) ) <EOL> assert '<STR_LIT>' in buf . getvalue ( ) <EOL> @ responses . activate <EOL> def test_download_already_exists_valid ( tempdir , mock_url ) : <EOL> with captured_logging ( ) as buf : <EOL> path = op . join ( tempdir , '<STR_LIT:test>' ) <EOL> with open ( path , '<STR_LIT>' ) as f : <EOL> f . write ( _DATA . tostring ( ) ) <EOL> _check ( _dl ( path ) ) <EOL> assert '<STR_LIT>' in buf . getvalue ( ) <EOL> @ responses . activate <EOL> def test_download_file ( tempdir , mock_urls ) : <EOL> path = op . join ( tempdir , '<STR_LIT:test>' ) <EOL> param , url_data , url_checksum = mock_urls <EOL> data_here , data_valid , checksum_here , checksum_valid = param <EOL> assert_succeeds = ( data_here and data_valid and <EOL> ( ( checksum_here == checksum_valid ) or <EOL> ( not ( checksum_here ) and checksum_valid ) ) ) <EOL> download_succeeds = ( assert_succeeds or ( data_here and <EOL> ( not ( data_valid ) and not ( checksum_here ) ) ) ) <EOL> if download_succeeds : <EOL> data = _dl ( path ) <EOL> else : <EOL> with raises ( Exception ) : <EOL> data = _dl ( path ) <EOL> if assert_succeeds : <EOL> _check ( data ) <EOL> @ responses . activate <EOL> def test_download_sample_data ( tempdir ) : <EOL> name = '<STR_LIT>' <EOL> url = _BASE_URL [ '<STR_LIT>' ] + name <EOL> _add_mock_response ( url , _DATA . tostring ( ) ) <EOL> _add_mock_response ( url + '<STR_LIT>' , _CHECKSUM ) <EOL> download_sample_data ( name , tempdir ) <EOL> with open ( op . join ( tempdir , name ) , '<STR_LIT:rb>' ) as f : <EOL> data = f . read ( ) <EOL> ae ( np . fromstring ( data , np . float32 ) , _DATA ) <EOL> download_sample_data ( name + '<STR_LIT:_>' , tempdir ) <EOL> responses . reset ( ) <EOL> @ responses . activate <EOL> def test_dat_file ( tempdir ) : <EOL> data = np . random . randint ( size = ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> low = - <NUM_LIT:100> , high = <NUM_LIT:100> ) . astype ( np . int16 ) <EOL> fn = '<STR_LIT>' <EOL> _add_mock_response ( _BASE_URL [ '<STR_LIT>' ] + '<STR_LIT>' + fn , <EOL> data . tostring ( ) ) <EOL> path = download_test_data ( fn , tempdir ) <EOL> with open ( path , '<STR_LIT:rb>' ) as f : <EOL> arr = np . fromfile ( f , dtype = np . int16 ) . reshape ( ( - <NUM_LIT:1> , <NUM_LIT:4> ) ) <EOL> assert arr . shape == ( <NUM_LIT> , <NUM_LIT:4> ) <EOL> assert download_test_data ( fn , tempdir ) == path <EOL> responses . reset ( ) </s>
