Browse Source

add support for module ordering via config...

when a dictionary is missing a key, make sure it gets deleted when
test setting it...
main
John-Mark Gurney 4 years ago
parent
commit
6889d7a27f
3 changed files with 65 additions and 4 deletions
  1. +55
    -3
      casimport/__init__.py
  2. +1
    -1
      casimport/default.conf
  3. +9
    -0
      fixtures/ordertest.conf

+ 55
- 3
casimport/__init__.py View File

@@ -22,6 +22,7 @@
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF # OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE. # SUCH DAMAGE.


import configparser
import contextlib import contextlib
import filecmp import filecmp
import functools import functools
@@ -75,11 +76,19 @@ def tempset(obj, key, value):
''' '''


try: try:
oldvalue = obj[key]
dodelitem = False
if key in obj:
oldvalue = obj[key]
else:
dodelitem = True

obj[key] = value obj[key] = value
yield yield
finally: finally:
obj[key] = oldvalue
if not dodelitem:
obj[key] = oldvalue
else:
del obj[key]


@contextlib.contextmanager @contextlib.contextmanager
def tempattrset(obj, key, value): def tempattrset(obj, key, value):
@@ -119,6 +128,10 @@ def urlfetch(url):
return req.read() return req.read()


class HTTPSCAS(object): class HTTPSCAS(object):
@classmethod
def fromconfig(cls, conf):
return cls()

def fetch_data(self, url): def fetch_data(self, url):
if url.scheme != 'https': if url.scheme != 'https':
raise ValueError('cannot handle scheme %s' % raise ValueError('cannot handle scheme %s' %
@@ -131,6 +144,10 @@ class IPFSCAS(object):
gwhost = 'gateway.ipfs.io' gwhost = 'gateway.ipfs.io'
gwhost = 'cloudflare-ipfs.com' gwhost = 'cloudflare-ipfs.com'


@classmethod
def fromconfig(cls, conf):
return cls()

def make_url(self, url): def make_url(self, url):
return urllib.parse.urlunparse(('https', self.gwhost, return urllib.parse.urlunparse(('https', self.gwhost,
'/ipfs/' + url.netloc) + ('', ) * 3) '/ipfs/' + url.netloc) + ('', ) * 3)
@@ -152,6 +169,10 @@ class FileDirCAS(object):
self._path.mkdir(exist_ok=True) self._path.mkdir(exist_ok=True)
self._hashes = {} self._hashes = {}


@classmethod
def fromconfig(cls, conf):
return cls(os.path.expanduser(conf['path']))

def refresh_dir(self): def refresh_dir(self):
'''Internal method to refresh the internal cache of '''Internal method to refresh the internal cache of
hashes.''' hashes.'''
@@ -353,17 +374,33 @@ class CASFinder(MetaPathFinder, Loader):


exec(data, module.__dict__) exec(data, module.__dict__)


_supportedmodules = {
'https': HTTPSCAS.fromconfig,
'ipfs': IPFSCAS.fromconfig,
'cache': FileDirCAS.fromconfig,
}

def defaultinit(casf): def defaultinit(casf):
basedir = pathlib.Path.home() / '.casimport' basedir = pathlib.Path.home() / '.casimport'
basedir.mkdir(exist_ok=True) basedir.mkdir(exist_ok=True)


conffile = basedir / 'casimport.conf'
conffile = pathlib.Path(os.environ.get('CASIMPORT_CONF', basedir / 'casimport.conf'))
if not conffile.exists(): if not conffile.exists():
import casimport import casimport
with importlib.resources.path(casimport, with importlib.resources.path(casimport,
'default.conf') as defconf: 'default.conf') as defconf:
shutil.copy(defconf, conffile) shutil.copy(defconf, conffile)


cp = configparser.ConfigParser()
cp.read(conffile)

modorder = [ x.strip() for x in
cp['casimport']['module_prio'].split(',') ]

for i in modorder:
modfun = _supportedmodules[cp[i]['type']]
casf.register(modfun(cp[i]))

cachedir = basedir / 'cache' cachedir = basedir / 'cache'
cachedir.mkdir(parents=True, exist_ok=True) cachedir.mkdir(parents=True, exist_ok=True)


@@ -484,6 +521,21 @@ class Test(unittest.TestCase):
# and when disconnected as second time, nothing happens # and when disconnected as second time, nothing happens
f.disconnect() f.disconnect()


def test_conforder(self):
conf = self.fixtures / 'ordertest.conf'

# that w/ a custom config
with tempset(os.environ, 'CASIMPORT_CONF', str(conf)), \
CASFinder() as f:
# that is gets loaded
defaultinit(f)

# and that the first loader is the HTTPSCAS
self.assertIsInstance(f._loaders[0], HTTPSCAS)

# and that the second loader is the IPFSCAS
self.assertIsInstance(f._loaders[1], IPFSCAS)

def test_defaultinit(self): def test_defaultinit(self):
temphome = self.tempdir / 'home' temphome = self.tempdir / 'home'
temphome.mkdir() temphome.mkdir()


+ 1
- 1
casimport/default.conf View File

@@ -11,7 +11,7 @@ module_prio = homecache, ipfsgw, https
# Default cache is in the user's home directory. # Default cache is in the user's home directory.


type = cache type = cache
dir = ~/.casimport/cache
path = ~/.casimport/cache


#size = 10MB #size = 10MB




+ 9
- 0
fixtures/ordertest.conf View File

@@ -0,0 +1,9 @@
[casimport]
module_prio = https, ipfsgw

[ipfsgw]
type = ipfs
gateway = https://gateway.ipfs.io/ipfs/

[https]
type = https

Loading…
Cancel
Save