pigway-raspberry-pi-agent/tests/test_storage_cache.py

import sys,tempfile,unittest
from pathlib import Path
from types import SimpleNamespace
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'app'))
from storage_cache import StorageCache


class StorageTests(unittest.TestCase):
    def test_weekly_reads_and_restart_cache(self):
        with tempfile.TemporaryDirectory() as directory:
            now=[1000];reads=[]
            def reader(path):
                reads.append(path);return SimpleNamespace(total=1000,used=200,free=800)
            p=Path(directory)/'cache.json'
            cache=StorageCache(p,reader,lambda:now[0])
            self.assertEqual(cache.sample()['used'],200)
            for i in range(20):now[0]+=1;cache.sample()
            self.assertEqual(len(reads),1)
            restored=StorageCache(p,reader,lambda:now[0]);restored.sample()
            self.assertEqual(len(reads),1)
            now[0]=605800;restored.sample()
            self.assertEqual(len(reads),2)

    def test_failure_does_not_fabricate_zero_or_retry_each_second(self):
        with tempfile.TemporaryDirectory() as directory:
            now=[1000];reads=[]
            def fail(path):reads.append(path);raise OSError('unavailable')
            cache=StorageCache(Path(directory)/'cache.json',fail,lambda:now[0])
            self.assertIsNone(cache.sample());self.assertTrue(cache.error)
            now[0]+=5;self.assertIsNone(cache.sample())
            self.assertEqual(len(reads),1)

    def test_failed_read_cooldown_survives_restart(self):
        with tempfile.TemporaryDirectory() as directory:
            reads=[]
            def fail(path):reads.append(path);raise OSError('unavailable')
            p=Path(directory)/'cache.json'
            StorageCache(p,fail,lambda:1000).sample()
            restored=StorageCache(p,fail,lambda:1005)
            self.assertTrue(restored.error);self.assertIsNone(restored.sample())
            self.assertEqual(len(reads),1)