<s> from __future__ import absolute_import <EOL> import importlib <EOL> import os <EOL> import sys <EOL> from zipfile import ZipFile <EOL> from bd2k . util . files import mkdir_p <EOL> from io import BytesIO <EOL> from mock import MagicMock , patch <EOL> from toil . resource import ModuleDescriptor , Resource , ResourceException <EOL> from toil . test import ToilTest <EOL> class ResourceTest ( ToilTest ) : <EOL> """<STR_LIT>""" <EOL> def testStandAlone ( self ) : <EOL> self . _testExternal ( moduleName = '<STR_LIT>' , pyFiles = ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) ) <EOL> def testPackage ( self ) : <EOL> self . _testExternal ( moduleName = '<STR_LIT>' , pyFiles = ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) ) <EOL> def testStandAloneInPackage ( self ) : <EOL> self . assertRaises ( ResourceException , <EOL> self . _testExternal , <EOL> moduleName = '<STR_LIT>' , <EOL> pyFiles = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def _testExternal ( self , moduleName , pyFiles ) : <EOL> dirPath = self . _createTempDir ( ) <EOL> pycFiles = set ( pyFile + '<STR_LIT:c>' for pyFile in pyFiles ) <EOL> for relPath in pyFiles : <EOL> path = os . path . join ( dirPath , relPath ) <EOL> mkdir_p ( os . path . dirname ( path ) ) <EOL> with open ( path , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> sys . path . append ( dirPath ) <EOL> try : <EOL> userScript = importlib . import_module ( moduleName ) <EOL> try : <EOL> self . _test ( userScript . __name__ , expectedContents = pycFiles ) <EOL> finally : <EOL> del userScript <EOL> del sys . modules [ moduleName ] <EOL> self . assertFalse ( moduleName in sys . modules ) <EOL> finally : <EOL> sys . path . remove ( dirPath ) <EOL> def testBuiltIn ( self ) : <EOL> module_name = ModuleDescriptor . __module__ <EOL> self . assertEquals ( module_name , '<STR_LIT>' ) <EOL> self . _test ( module_name , shouldBelongToToil = True ) <EOL> def _test ( self , module_name , shouldBelongToToil = False , expectedContents = None ) : <EOL> module = ModuleDescriptor . forModule ( module_name ) <EOL> self . assertEqual ( module . belongsToToil , shouldBelongToToil ) <EOL> self . assertEquals ( module . name , module_name ) <EOL> if shouldBelongToToil : <EOL> self . assertTrue ( module . dirPath . endswith ( '<STR_LIT>' ) ) <EOL> self . assertIs ( module . localize ( ) , module ) <EOL> self . assertIs ( module . globalize ( ) , module ) <EOL> jobStore = MagicMock ( ) <EOL> url = '<STR_LIT>' <EOL> jobStore . getSharedPublicUrl . return_value = url <EOL> resource = module . saveAsResourceTo ( jobStore ) <EOL> jobStore . getSharedPublicUrl . assert_called_once_with ( resource . pathHash ) <EOL> jobStore . writeSharedFileStream . assert_called_once_with ( resource . pathHash , <EOL> isProtected = False ) <EOL> file_handle = jobStore . writeSharedFileStream . return_value . __enter__ . return_value <EOL> zipFile = file_handle . write . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertTrue ( zipFile . startswith ( '<STR_LIT>' ) ) <EOL> if expectedContents is not None : <EOL> with ZipFile ( BytesIO ( zipFile ) ) as _zipFile : <EOL> self . assertEqual ( set ( _zipFile . namelist ( ) ) , expectedContents ) <EOL> self . assertEquals ( resource . url , url ) <EOL> Resource . prepareSystem ( ) <EOL> resource . register ( ) <EOL> localResource = Resource . lookup ( module . _resourcePath ) <EOL> self . assertEquals ( resource , localResource ) <EOL> self . assertIsNot ( resource , localResource ) <EOL> mock_urlopen = MagicMock ( ) <EOL> mock_urlopen . return_value . read . return_value = zipFile <EOL> with patch ( '<STR_LIT>' , mock_urlopen ) : <EOL> localModule = module . localize ( ) <EOL> self . assertEquals ( module . name , localModule . name ) <EOL> self . assertNotEquals ( module . dirPath , localModule . dirPath ) <EOL> self . assertEquals ( localModule . globalize ( ) , module ) </s>
