shep/tests/test_file.py

231 lines
6.5 KiB
Python
Raw Normal View History

2022-01-31 16:53:33 +01:00
# standard imports
import unittest
import tempfile
import os
# local imports
from shep.persist import PersistedState
from shep.store.file import SimpleFileStoreFactory
from shep.error import (
StateExists,
StateInvalid,
StateItemExists,
2022-01-31 16:53:33 +01:00
)
class TestStateReport(unittest.TestCase):
def setUp(self):
self.d = tempfile.mkdtemp()
self.factory = SimpleFileStoreFactory(self.d)
self.states = PersistedState(self.factory.add, 4)
self.states.add('foo')
self.states.add('bar')
self.states.add('baz')
def test_add(self):
2022-02-01 08:01:57 +01:00
self.states.put('abcd', state=self.states.FOO, contents='baz')
2022-01-31 16:53:33 +01:00
fp = os.path.join(self.d, 'FOO', 'abcd')
f = open(fp, 'r')
v = f.read()
f.close()
2022-02-01 08:01:57 +01:00
self.assertEqual(v, 'baz')
2022-01-31 16:53:33 +01:00
def test_dup(self):
2022-02-01 08:01:57 +01:00
self.states.put('abcd', state=self.states.FOO)
with self.assertRaises(StateItemExists):
2022-02-01 08:01:57 +01:00
self.states.put('abcd', state=self.states.FOO)
with self.assertRaises(StateItemExists): #FileExistsError):
self.states.put('abcd', state=self.states.FOO)
2022-01-31 16:53:33 +01:00
2022-02-01 10:55:12 +01:00
def test_list(self):
self.states.put('abcd', state=self.states.FOO)
self.states.put('xx!', state=self.states.FOO)
self.states.put('1234', state=self.states.BAR)
keys = self.states.list(self.states.FOO)
self.assertIn('abcd', keys)
self.assertIn('xx!', keys)
self.assertNotIn('1234', keys)
self.states.alias('xyzzy', self.states.BAR | self.states.FOO)
self.states.put('yyy', state=self.states.XYZZY)
keys = self.states.list(self.states.XYZZY)
self.assertIn('yyy', keys)
self.assertNotIn('1234', keys)
self.assertNotIn('xx!', keys)
def test_move(self):
self.states.put('abcd', state=self.states.FOO, contents='foo')
self.states.move('abcd', self.states.BAR)
2022-02-01 10:55:12 +01:00
fp = os.path.join(self.d, 'BAR', 'abcd')
f = open(fp, 'r')
v = f.read()
f.close()
fp = os.path.join(self.d, 'FOO', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
2022-03-11 13:01:56 +01:00
def test_change(self):
self.states.alias('inky', self.states.FOO | self.states.BAR)
self.states.put('abcd', state=self.states.FOO, contents='foo')
self.states.change('abcd', self.states.BAR, 0)
fp = os.path.join(self.d, 'INKY', 'abcd')
f = open(fp, 'r')
v = f.read()
f.close()
fp = os.path.join(self.d, 'FOO', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
fp = os.path.join(self.d, 'BAR', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
self.states.change('abcd', 0, self.states.BAR)
fp = os.path.join(self.d, 'FOO', 'abcd')
f = open(fp, 'r')
v = f.read()
f.close()
fp = os.path.join(self.d, 'INKY', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
fp = os.path.join(self.d, 'BAR', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
def test_set(self):
self.states.alias('xyzzy', self.states.FOO | self.states.BAR)
self.states.put('abcd', state=self.states.FOO, contents='foo')
self.states.set('abcd', self.states.BAR)
fp = os.path.join(self.d, 'XYZZY', 'abcd')
f = open(fp, 'r')
v = f.read()
f.close()
fp = os.path.join(self.d, 'FOO', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
fp = os.path.join(self.d, 'BAR', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
self.states.unset('abcd', self.states.FOO)
fp = os.path.join(self.d, 'BAR', 'abcd')
f = open(fp, 'r')
v = f.read()
f.close()
fp = os.path.join(self.d, 'FOO', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
fp = os.path.join(self.d, 'XYZZY', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
2022-02-01 10:55:12 +01:00
2022-03-17 20:16:33 +01:00
def test_sync_one(self):
2022-02-01 15:38:24 +01:00
self.states.put('abcd', state=self.states.FOO, contents='foo')
self.states.put('xxx', state=self.states.FOO)
self.states.put('yyy', state=self.states.FOO)
fp = os.path.join(self.d, 'FOO', 'yyy')
f = open(fp, 'w')
f.write('')
f.close()
fp = os.path.join(self.d, 'FOO', 'zzzz')
f = open(fp, 'w')
f.write('xyzzy')
f.close()
self.states.sync(self.states.FOO)
self.assertEqual(self.states.get('yyy'), None)
self.assertEqual(self.states.get('zzzz'), 'xyzzy')
2022-02-02 08:24:43 +01:00
2022-03-17 20:16:33 +01:00
def test_sync_all(self):
self.states.put('abcd', state=self.states.FOO)
self.states.put('xxx', state=self.states.BAR)
fp = os.path.join(self.d, 'FOO', 'abcd')
f = open(fp, 'w')
f.write('foofoo')
f.close()
fp = os.path.join(self.d, 'BAR', 'zzzz')
f = open(fp, 'w')
f.write('barbar')
f.close()
self.states.sync()
self.assertEqual(self.states.get('abcd'), None)
self.assertEqual(self.states.get('zzzz'), 'barbar')
2022-02-02 08:24:43 +01:00
def test_path(self):
self.states.put('yyy', state=self.states.FOO)
d = os.path.join(self.d, 'FOO')
self.assertEqual(self.states.path(self.states.FOO), d)
d = os.path.join(self.d, 'FOO', 'BAR')
self.assertEqual(self.states.path(self.states.FOO, key='BAR'), d)
2022-02-02 10:27:57 +01:00
def test_next(self):
self.states.put('abcd')
self.states.next('abcd')
self.assertEqual(self.states.state('abcd'), self.states.FOO)
2022-02-01 15:38:24 +01:00
2022-02-02 10:27:57 +01:00
self.states.next('abcd')
self.assertEqual(self.states.state('abcd'), self.states.BAR)
self.states.next('abcd')
self.assertEqual(self.states.state('abcd'), self.states.BAZ)
2022-02-02 10:27:57 +01:00
with self.assertRaises(StateInvalid):
self.states.next('abcd')
fp = os.path.join(self.d, 'FOO', 'abcd')
with self.assertRaises(FileNotFoundError):
os.stat(fp)
fp = os.path.join(self.d, 'BAZ', 'abcd')
2022-02-02 10:27:57 +01:00
os.stat(fp)
2022-02-01 15:38:24 +01:00
def test_replace(self):
self.states.put('abcd')
self.states.replace('abcd', 'foo')
self.assertEqual(self.states.get('abcd'), 'foo')
fp = os.path.join(self.d, 'NEW', 'abcd')
f = open(fp, 'r')
r = f.read()
f.close()
self.assertEqual(r, 'foo')
2022-01-31 16:53:33 +01:00
if __name__ == '__main__':
unittest.main()