chaintool/chainqueue
4
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Add purge queue entry method

Browse Source
This commit is contained in:
nolash
2021-06-01 13:47:55 +02:00
parent bdd678fb3f
commit 3e487d0702
2 changed files with 33 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
tests/test_fs.py
View File

@@ -54,7 +54,7 @@ class HexDirTest(unittest.TestCase):
tx_hash = os.urandom(32)
tx_content = os.urandom(128)
self.q.add(tx_hash, tx_content)
self.q.move(tx_hash, 'new', 'ready')
self.q.move(tx_hash, 'ready', from_state='new')
f = open(os.path.join(self.q.path_state['ready'], tx_hash.hex()), 'rb')
r = f.read()
@@ -62,5 +62,21 @@ class HexDirTest(unittest.TestCase):
self.assertEqual(r, b'\x00' * 8)
def test_purge(self):
tx_hash = os.urandom(32)
tx_content = os.urandom(128)
self.q.add(tx_hash, tx_content)
self.q.move(tx_hash, 'ready', from_state='new')
self.q.purge(tx_hash, 'ready')
with self.assertRaises(FileNotFoundError):
entry_path = os.path.join(self.q.path_state['ready'], tx_hash.hex())
os.stat(entry_path)
with self.assertRaises(FileNotFoundError):
entry_path = os.path.join(self.q.index_path, tx_hash.hex())
os.stat(entry_path)
if __name__ == '__main__':
unittest.main()