diff --git a/ui/medashare/cli.py b/ui/medashare/cli.py index 402ccf7..3cf8801 100644 --- a/ui/medashare/cli.py +++ b/ui/medashare/cli.py @@ -32,6 +32,7 @@ from edgold.ed448 import EDDSA448 from unittest import mock from .hostid import hostuuid +from .tags import TagCache from . import orm from .btv import _TestCases as bttestcase, validate_file @@ -748,121 +749,6 @@ def enumeratedir(_dir, created_by_ref): created_by_ref) for x in sorted(os.listdir(_dir)) if not os.path.isdir(os.path.join(_dir, x)) ] -class TagCache: - '''Takes tuples, and stores them in a cache where only - the most count recent ones are kept.''' - - def __init__(self, tags=(), count=10): - self._cache = dict((x, None) for x in tags) - self._count = count - self._modified = False - - @property - def modified(self): - '''Return if the cache has been modified since the last - time store has been called.''' - - return self._modified - - def add(self, tag): - '''Add a tag (tuple) to the cache, possibly dropping ann older - tag if necessary.''' - - self._modified = True - - try: - del self._cache[tag] - except KeyError: - pass - - self._cache[tag] = None - - while len(self._cache) > self._count: - del self._cache[next(iter(self._cache.keys()))] - - def tags(self): - '''Returns the sorted list of tags in the cache.''' - - return sorted(self._cache.keys()) - - @classmethod - def load(cls, fname): - '''Load the cache from fname. Must be previously have saved - using the store method.''' - - try: - with open(fname, 'rb') as fp: - cache = _asn1coder.loads(fp.read()) - except (FileNotFoundError, IndexError): - # IndexError when file exists, but is invalid - return cls() - - # fix up - cache['tags'] = [ tuple(x) for x in cache['tags'] ] - - return cls(**cache) - - def store(self, fname): - '''Store the cache to fname. The modified property will - be cleared after this.''' - - self._modified = False - - cache = dict(tags=list(self._cache.keys()), count=self._count) - - with open(fname, 'wb') as fp: - fp.write(_asn1coder.dumps(cache)) - -class _TestTagCache(unittest.TestCase): - def setUp(self): - d = pathlib.Path(tempfile.mkdtemp()).resolve() - self.basetempdir = d - - self.tempdir = d / 'subdir' - - self.tempdir.mkdir() - - def tearDown(self): - shutil.rmtree(self.basetempdir) - self.tempdir = None - - def test_cache(self): - # test basic functionality - tc = TagCache(count=2) - - self.assertFalse(tc.modified) - - tc.add(('foo', 'foo')) - - self.assertTrue(tc.modified) - - tc.add(('bar', 'bar')) - - self.assertEqual(tc.tags(), [ ('bar', 'bar'), ('foo', 'foo') ]) - - tc.add(('foo', 'foo')) - - tc.add(('baz', 'baz')) - - tc.add(('foo', 'foo')) - - self.assertEqual(tc.tags(), [ ('baz', 'baz'), ('foo', 'foo') ]) - - cachefile = self.tempdir / 'somecache' - tc.store(cachefile) - - self.assertFalse(tc.modified) - - ntc = TagCache.load(cachefile) - - self.assertFalse(ntc.modified) - - self.assertEqual(tc.tags(), [ ('baz', 'baz'), ('foo', 'foo') ]) - - ntc.add(('whee', 'whee')) - - self.assertEqual(ntc.tags(), [ ('foo', 'foo'), ('whee', 'whee') ]) - def _get_paths(options): fnames = ( '.medashare_identity.pasn1', @@ -2282,6 +2168,7 @@ class _TestCases(unittest.TestCase): # setup object store storefname = self.tempdir / 'storefname' identfname = self.tempdir / 'identfname' + cachefname = self.tempdir / 'cachefname' # setup path mapping def expandusermock(arg): @@ -2289,6 +2176,8 @@ class _TestCases(unittest.TestCase): return storefname elif arg == '~/.medashare_identity.pasn1': return identfname + elif arg == '~/.medashare_cache.pasn1': + return cachefname # setup test fname testfname = os.path.join(self.tempdir, 'test.txt') diff --git a/ui/medashare/tags.py b/ui/medashare/tags.py new file mode 100644 index 0000000..6c2592a --- /dev/null +++ b/ui/medashare/tags.py @@ -0,0 +1,121 @@ +import pathlib +import shutil +import tempfile +import unittest +from .utils import _asn1coder + +class TagCache: + '''Takes tuples, and stores them in a cache where only + the most count recent ones are kept.''' + + def __init__(self, tags=(), count=10): + self._cache = dict((x, None) for x in tags) + self._count = count + self._modified = False + + @property + def modified(self): + '''Return if the cache has been modified since the last + time store has been called.''' + + return self._modified + + def add(self, tag): + '''Add a tag (tuple) to the cache, possibly dropping ann older + tag if necessary.''' + + self._modified = True + + try: + del self._cache[tag] + except KeyError: + pass + + self._cache[tag] = None + + while len(self._cache) > self._count: + del self._cache[next(iter(self._cache.keys()))] + + def tags(self): + '''Returns the sorted list of tags in the cache.''' + + return sorted(self._cache.keys()) + + @classmethod + def load(cls, fname): + '''Load the cache from fname. Must be previously have saved + using the store method.''' + + try: + with open(fname, 'rb') as fp: + cache = _asn1coder.loads(fp.read()) + except (FileNotFoundError, IndexError): + # IndexError when file exists, but is invalid + return cls() + + # fix up + cache['tags'] = [ tuple(x) for x in cache['tags'] ] + + return cls(**cache) + + def store(self, fname): + '''Store the cache to fname. The modified property will + be cleared after this.''' + + self._modified = False + + cache = dict(tags=list(self._cache.keys()), count=self._count) + + with open(fname, 'wb') as fp: + fp.write(_asn1coder.dumps(cache)) + +class _TestTagCache(unittest.TestCase): + def setUp(self): + d = pathlib.Path(tempfile.mkdtemp()).resolve() + self.basetempdir = d + + self.tempdir = d / 'subdir' + + self.tempdir.mkdir() + + def tearDown(self): + shutil.rmtree(self.basetempdir) + self.tempdir = None + + def test_cache(self): + # test basic functionality + tc = TagCache(count=2) + + self.assertFalse(tc.modified) + + tc.add(('foo', 'foo')) + + self.assertTrue(tc.modified) + + tc.add(('bar', 'bar')) + + self.assertEqual(tc.tags(), [ ('bar', 'bar'), ('foo', 'foo') ]) + + tc.add(('foo', 'foo')) + + tc.add(('baz', 'baz')) + + tc.add(('foo', 'foo')) + + self.assertEqual(tc.tags(), [ ('baz', 'baz'), ('foo', 'foo') ]) + + cachefile = self.tempdir / 'somecache' + tc.store(cachefile) + + self.assertFalse(tc.modified) + + ntc = TagCache.load(cachefile) + + self.assertFalse(ntc.modified) + + self.assertEqual(tc.tags(), [ ('baz', 'baz'), ('foo', 'foo') ]) + + ntc.add(('whee', 'whee')) + + self.assertEqual(ntc.tags(), [ ('foo', 'foo'), ('whee', 'whee') ]) + diff --git a/ui/medashare/tests.py b/ui/medashare/tests.py index ae5602a..08f2c2a 100644 --- a/ui/medashare/tests.py +++ b/ui/medashare/tests.py @@ -2,6 +2,7 @@ from .btv import _TestCases as btv_test_cases from .btv.bencode import _TestCases as bencode_test_cases from .mdb import _TestJSONEncoder from .cli import _TestCononicalCoder, _TestCases as cli_test_cases -from .cli import _TestMigrations, _TestTagCache +from .cli import _TestMigrations +from .tags import _TestTagCache from .mtree import Test from .server import _TestCases, _TestPostConfig