Mercurial > code > home > repos > rdfdb
changeset 86:5b6e90a708ce
some weak file_vs_uri coverage
author | drewp@bigasterisk.com |
---|---|
date | Mon, 04 Apr 2022 23:00:42 -0700 |
parents | 7d894e977615 |
children | a96f4eb95ef0 |
files | rdfdb/file_vs_uri.py rdfdb/file_vs_uri_test.py rdfdb/service.py |
diffstat | 3 files changed, 52 insertions(+), 16 deletions(-) [+] |
line wrap: on
line diff
--- a/rdfdb/file_vs_uri.py Mon Apr 04 22:08:41 2022 -0700 +++ b/rdfdb/file_vs_uri.py Mon Apr 04 23:00:42 2022 -0700 @@ -2,36 +2,45 @@ note that if there are ambiguities in dirUriMap, you'll get undefined results that depend on python's items() order. """ -import os +from pathlib import Path from typing import Dict from rdflib import URIRef -DirUriMap = Dict[bytes, URIRef] +DirUriMap = Dict[Path, URIRef] + + +def _pathStartsWith(fullPath: Path, possiblePrefix: Path) -> bool: + return str(fullPath).startswith(str(possiblePrefix)) -def uriFromFile(dirUriMap: DirUriMap, filename: bytes) -> URIRef: - assert filename.endswith(b'.n3'), filename - for d, prefix in list(dirUriMap.items()): - if filename.startswith(d): - return URIRef(prefix + filename[len(d):-len(b'.n3')].decode('ascii')) +def uriFromFile(dirUriMap: DirUriMap, filename: Path) -> URIRef: + if not str(filename).endswith('.n3'): + raise ValueError(filename) + for d, prefix in dirUriMap.items(): + if _pathStartsWith(filename, d): + return URIRef(prefix + str(filename)[len(str(d)):-len(b'.n3')]) raise ValueError("filename %s doesn't start with any of %s" % (filename, list(dirUriMap.keys()))) -def fileForUri(dirUriMap: DirUriMap, ctx: URIRef) -> bytes: - assert isinstance(ctx, URIRef), ctx +def fileForUri(dirUriMap: DirUriMap, ctx: URIRef) -> Path: for d, prefix in dirUriMap.items(): if ctx.startswith(prefix): - return d + ctx[len(prefix):].encode('utf8') + b'.n3' + middle = ctx[len(prefix):] + if middle.endswith('/'): + raise NotImplementedError(f'no file mapped to {ctx=}') + return Path(str(d) + middle + '.n3') raise ValueError("don't know what filename to use for %s" % ctx) -def correctToTopdirPrefix(dirUriMap: DirUriMap, inFile: bytes) -> bytes: - if not any(inFile.startswith(prefix) for prefix in dirUriMap): +def correctToTopdirPrefix(dirUriMap: DirUriMap, inFile: Path) -> Path: + # can't remember what this is for, so it's a no-op to hopefully flush out the bugs + return inFile + if not any(str(inFile).startswith(str(prefix)) for prefix in dirUriMap): for prefix in dirUriMap: - prefixAbs = os.path.abspath(prefix) - if inFile.startswith(prefixAbs): - inFile = prefix + inFile[len(prefixAbs):] + prefixAbs = prefix.absolute() + if _pathStartsWith(inFile, prefixAbs): + inFile = Path(str(prefix) + str(inFile)[len(str(prefixAbs)):]) break else: raise ValueError("can't correct %s to start with one of %s" % (inFile, list(dirUriMap.keys())))
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/rdfdb/file_vs_uri_test.py Mon Apr 04 23:00:42 2022 -0700 @@ -0,0 +1,25 @@ +import os +from pathlib import Path + +import pytest +from rdflib import URIRef + +from rdfdb.file_vs_uri import correctToTopdirPrefix, fileForUri, uriFromFile + +demoMap = {Path('/tmp'): URIRef('http://example.com/myroot')} + + +def test_uriFromFile(): + assert uriFromFile(demoMap, Path('/tmp/foo/bar.n3')) == URIRef('http://example.com/myroot/foo/bar') + + +def test_fileForUri(): + assert fileForUri(demoMap, URIRef('http://example.com/myroot/where/to/save/this')) == Path('/tmp/where/to/save/this.n3') + with pytest.raises(NotImplementedError): + fileForUri(demoMap, URIRef('http://example.com/myroot/where/to/save/this/')) + + +def test_correctToTopdirPrefix(): + os.chdir('/tmp') + assert correctToTopdirPrefix(demoMap, Path("/tmp/foo.n3")) == Path("/tmp/foo.n3") + # todo: something to do with relative paths
--- a/rdfdb/service.py Mon Apr 04 22:08:41 2022 -0700 +++ b/rdfdb/service.py Mon Apr 04 23:00:42 2022 -0700 @@ -80,7 +80,7 @@ def dirChange(self, watch, path: FilePath, mask): if mask & IN_CREATE: - if path.path.endswith((b'~', b'.swp', b'swx', b'.rdfdb-temp')): + if cast(str, path.path).endswith((b'~', b'.swp', b'swx', b'.rdfdb-temp')): return log.debug("%s created; consider adding a watch", path) @@ -374,5 +374,7 @@ db=db)) log.info("serving on %s" % port) reactor.run() + + if __name__ == '__main__': main() \ No newline at end of file