# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2025, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
#    this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

import time
import os
import sys
import zlib
import pytest
import shutil
import json
import gzip
from unittest import mock
from datetime import datetime, timedelta, timezone
from apprise import exception
from apprise.asset import AppriseAsset
from apprise.persistent_store import (
    CacheJSONEncoder, CacheObject, PersistentStore, PersistentStoreMode)

# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)

# Attachment Directory
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')


def test_persistent_storage_asset(tmpdir):
    """
    Tests the Apprise Asset Object when setting the Persistent Store
    """

    asset = AppriseAsset(storage_path=str(tmpdir))
    assert asset.storage_path == str(tmpdir)
    assert asset.storage_mode is PersistentStoreMode.AUTO

    # If there is no storage path, we're always set to memory
    asset = AppriseAsset(
        storage_path=None, storage_mode=PersistentStoreMode.MEMORY)
    assert asset.storage_path is None
    assert asset.storage_mode is PersistentStoreMode.MEMORY


def test_disabled_persistent_storage(tmpdir):
    """
    Persistent Storage General Testing

    """
    # Create ourselves an attachment object set in Memory Mode only
    pc = PersistentStore(
        namespace='abc', path=str(tmpdir), mode=PersistentStoreMode.MEMORY)
    assert pc.read() is None
    assert pc.read('mykey') is None
    with pytest.raises(AttributeError):
        # Invalid key specified
        pc.read('!invalid')
    assert pc.write('data') is False
    assert pc.get('key') is None
    assert pc.set('key', 'value')
    assert pc.get('key') == 'value'

    assert pc.set('key2', 'value')
    pc.clear('key', 'key-not-previously-set')
    assert pc.get('key2') == 'value'
    assert pc.get('key') is None

    # Set it again
    assert pc.set('key', 'another-value')
    # Clears all
    pc.clear()
    assert pc.get('key2') is None
    assert pc.get('key') is None
    # A second call to clear on an already empty cache set
    pc.clear()

    # No dirty flag is set as ther is nothing to write to disk
    pc.set('not-persistent', 'value', persistent=False)
    del pc['not-persistent']
    with pytest.raises(KeyError):
        # Can't delete it twice
        del pc['not-persistent']

    # A Persistent key
    pc.set('persistent', 'value')
    # Removes it and sets/clears the dirty flag
    del pc['persistent']

    # After all of the above, nothing was done to the directory
    assert len(os.listdir(str(tmpdir))) == 0

    with pytest.raises(AttributeError):
        # invalid persistent store specified
        PersistentStore(
            namespace='abc', path=str(tmpdir), mode='garbage')


def test_persistent_storage_init(tmpdir):
    """
    Test storage initialization
    """
    with pytest.raises(AttributeError):
        PersistentStore(namespace="", path=str(tmpdir))
    with pytest.raises(AttributeError):
        PersistentStore(namespace=None, path=str(tmpdir))

    with pytest.raises(AttributeError):
        PersistentStore(namespace="_", path=str(tmpdir))
    with pytest.raises(AttributeError):
        PersistentStore(namespace=".", path=str(tmpdir))
    with pytest.raises(AttributeError):
        PersistentStore(namespace="-", path=str(tmpdir))

    with pytest.raises(AttributeError):
        PersistentStore(namespace="_abc", path=str(tmpdir))
    with pytest.raises(AttributeError):
        PersistentStore(namespace=".abc", path=str(tmpdir))
    with pytest.raises(AttributeError):
        PersistentStore(namespace="-abc", path=str(tmpdir))

    with pytest.raises(AttributeError):
        PersistentStore(namespace="%", path=str(tmpdir))


def test_persistent_storage_general(tmpdir):
    """
    Persistent Storage General Testing

    """
    namespace = 'abc'
    # Create ourselves an attachment object
    pc = PersistentStore()

    # Default mode when a path is not provided
    assert pc.mode == PersistentStoreMode.MEMORY

    assert pc.size() == 0
    assert pc.files() == []
    assert pc.files(exclude=True, lazy=False) == []
    assert pc.files(exclude=False, lazy=False) == []
    pc.set('key', 'value')
    # There is no disk size utilized
    assert pc.size() == 0
    assert pc.files(exclude=True, lazy=False) == []
    assert pc.files(exclude=False, lazy=False) == []

    # Create ourselves an attachment object
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir))

    # Default mode when a path is provided
    assert pc.mode == PersistentStoreMode.AUTO

    # Get our path associated with our Persistent Store
    assert pc.path == os.path.join(str(tmpdir), 'abc')

    # Expiry testing
    assert pc.set('key', 'value', datetime.now() + timedelta(hours=1))
    # i min in the future
    assert pc.set('key', 'value', 60)

    with pytest.raises(AttributeError):
        assert pc.set('key', 'value', 'invalid')

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir))

    # Our key is still valid and we load it from disk
    assert pc.get('key') == 'value'
    assert pc['key'] == 'value'

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir))
    assert pc.keys()
    # Second call after already initialized skips over initialization
    assert pc.keys()

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir))

    with pytest.raises(KeyError):
        # The below
        pc['unassigned_key']


def test_persistent_storage_auto_mode(tmpdir):
    """
    Persistent Storage Auto Write Testing

    """
    namespace = 'abc'
    # Create ourselves an attachment object
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.AUTO)

    pc.write(b'test')
    with mock.patch('os.unlink', side_effect=FileNotFoundError()):
        assert pc.delete(all=True) is True

    # Create a temporary file we can delete
    with open(os.path.join(pc.path, pc.temp_dir, 'test.file'), 'wb') as fd:
        fd.write(b'data')

    # Delete just the temporary files
    assert pc.delete(temp=True) is True

    # Delete just the temporary files
    # Create a cache entry and delete it
    assert pc.set('key', 'value') is True
    pc.write(b'test')
    assert pc.delete(cache=True) is True
    # Verify our data entry wasn't removed
    assert pc.read() == b'test'
    # But our cache was
    assert pc.get('key') is None

    # A reverse of the above... create a cache an data variable and
    # Clear the data; make sure our cache is still there
    assert pc.set('key', 'value') is True
    pc.write(b'test', key='iokey') is True
    assert pc.delete('iokey') is True
    assert pc.get('key') == 'value'
    assert pc.read('iokey') is None


def test_persistent_storage_flush_mode(tmpdir):
    """
    Persistent Storage Forced Write Testing

    """
    namespace = 'abc'
    # Create ourselves an attachment object
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # Reference path
    path = os.path.join(str(tmpdir), namespace)

    assert pc.size() == 0
    assert list(pc.files()) == []

    # Key is not set yet
    assert pc.get('key') is None
    assert len(pc.keys()) == 0
    assert 'key' not in pc

    # Verify our data is set
    assert pc.set('key', 'value')
    assert len(pc.keys()) == 1
    assert 'key' in list(pc.keys())

    assert pc.size() > 0
    assert len(pc.files()) == 1

    # Second call uses Lazy cache
    # Just our cache file
    assert len(pc.files()) == 1

    # Setting the same value again uses a lazy mode and
    # bypasses all of the write overhead
    assert pc.set('key', 'value')

    path_content = os.listdir(path)
    # var, cache.psdata, and tmp
    assert len(path_content) == 3

    # Assignments (causes another disk write)
    pc['key'] = 'value2'

    # Setting the same value and explictly marking the field as not being
    # perisistent
    pc.set('key-xx', 'abc123', persistent=False)
    # Changing it's value doesn't alter the persistent flag
    pc['key-xx'] = 'def678'
    # Setting it twice
    pc['key-xx'] = 'def678'

    # Our retrievals
    assert pc['key-xx'] == 'def678'
    assert pc.get('key-xx') == 'def678'

    # But on the destruction of our object, it is not available again
    del pc
    # Create ourselves an attachment object
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    assert pc.get('key-xx') is None
    with pytest.raises(KeyError):
        pc['key-xx']

    # Now our key is set
    assert 'key' in pc
    assert pc.get('key') == 'value2'

    # A directory was created identified by the namespace
    assert len(os.listdir(str(tmpdir))) == 1
    assert namespace in os.listdir(str(tmpdir))

    path_content = os.listdir(path)
    assert len(path_content) == 4

    # Another write doesn't change the file count
    pc['key'] = 'value3'
    path_content = os.listdir(path)
    assert len(path_content) == 4

    # Our temporary directory used for all file handling in this namespace
    assert pc.temp_dir in path_content
    # Our cache file
    assert os.path.basename(pc.cache_file) in path_content

    path = os.path.join(pc.path, pc.temp_dir)
    path_content = os.listdir(path)

    # We always do our best to clean any temporary files up
    assert len(path_content) == 0

    # Destroy our object
    del pc

    # Re-initialize it
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # Our key is persistent and available right away
    assert pc.get('key') == 'value3'
    assert 'key' in pc

    # Remove our item
    del pc['key']
    assert pc.size() == 0
    assert 'key' not in pc

    assert pc.write('data') is True
    assert pc.read() == b'data'
    assert pc.write(b'data') is True
    assert pc.read() == b'data'

    assert pc.read('default') == b'data'
    assert pc.write('data2', key='mykey') is True
    assert pc.read('mykey') == b'data2'

    # We can selectively delete our key
    assert pc.delete('mykey')
    assert pc.read('mykey') is None
    # Other keys are not touched
    assert pc.read('default') == b'data'
    assert pc.read() == b'data'
    # Full purge
    assert pc.delete()
    assert pc.read('mykey') is None
    assert pc.read() is None

    # Practice with files
    with open(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'), 'rb') as fd:
        assert pc.write(fd, key='mykey', compress=False) is True

        # Read our content back
        fd.seek(0)
        assert pc.read('mykey', compress=False) == fd.read()

    with open(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'), 'rb') as fd:
        assert pc.write(fd, key='mykey', compress=True) is True

        # Read our content back; content will be compressed
        fd.seek(0)
        assert pc.read('mykey', compress=True) == fd.read()

    class Foobar:
        def read(*args, **kwargs):
            return 42

    foobar = Foobar()
    # read() returns a non string/bin
    with pytest.raises(exception.AppriseDiskIOError):
        pc.write(foobar, key='foobar', compress=True)
    assert pc.read('foobar') is None

    class Foobar:
        def read(*args, **kwargs):
            return 'good'

    foobar = Foobar()
    # read() returns a string so the below write works
    assert pc.write(foobar, key='foobar', compress=True)
    assert pc.read('foobar') == b'good'
    pc.delete()

    class Foobar:
        def read(*args, **kwargs):
            # Throw an exception
            raise TypeError()

    foobar = Foobar()
    # read() returns a non string/bin
    with pytest.raises(exception.AppriseDiskIOError):
        pc.write(foobar, key='foobar', compress=True)
    assert pc.read('foobar') is None

    # Set our max_file_size
    _prev_max_file_size = pc.max_file_size
    pc.max_file_size = 1
    assert pc.delete()

    assert pc.write('data') is False
    assert pc.read() is None

    # Restore setting
    pc.max_file_size = _prev_max_file_size

    # Reset
    pc.delete()

    assert pc.write('data')
    # Corrupt our data
    data = pc.read(compress=False)[:20] + pc.read(compress=False)[:10]
    pc.write(data, compress=False)

    # Now we'll get an exception reading back the corrupted data
    assert pc.read() is None

    # Keep in mind though the data is still there; operator should write
    # and read the way they expect to and things will work out fine
    # This test just proves that Apprise Peresistent storage still
    # gracefully handles bad data
    assert pc.read(compress=False) == data

    # No key exists also returns None
    assert pc.read('no-key-exists') is None

    pc.write(b'test')
    pc['key'] = 'value'
    with mock.patch('os.unlink', side_effect=FileNotFoundError()):
        assert pc.delete(all=True) is True
    with mock.patch('os.unlink', side_effect=OSError()):
        assert pc.delete(all=True) is False

    # Create a temporary file we can delete
    tmp_file = os.path.join(pc.path, pc.temp_dir, 'test.file')
    with open(tmp_file, 'wb') as fd:
        fd.write(b'data')

    assert pc.set('key', 'value') is True
    pc.write(b'test', key='iokey') is True
    # Delete just the temporary files
    assert pc.delete(temp=True) is True
    assert os.path.exists(tmp_file) is False
    # our other entries are untouched
    assert pc.get('key') == 'value'
    assert pc.read('iokey') == b'test'

    # Delete just the temporary files
    # Create a cache entry and delete it
    assert pc.set('key', 'value') is True
    pc.write(b'test')
    assert pc.delete(cache=True) is True
    # Verify our data entry wasn't removed
    assert pc.read() == b'test'
    # But our cache was
    assert pc.get('key') is None

    # A reverse of the above... create a cache an data variable and
    # Clear the data; make sure our cache is still there
    assert pc.set('key', 'value') is True
    pc.write(b'test', key='iokey') is True
    assert pc.delete('iokey') is True
    assert pc.get('key') == 'value'
    assert pc.read('iokey') is None

    # Create some custom files
    cust1_file = os.path.join(pc.path, 'test.file')
    cust2_file = os.path.join(pc.path, pc.data_dir, 'test.file')
    with open(cust1_file, 'wb') as fd:
        fd.write(b'data')
    with open(cust2_file, 'wb') as fd:
        fd.write(b'data')

    # Even after a full flush our files will exist
    assert pc.delete()
    assert os.path.exists(cust1_file) is True
    assert os.path.exists(cust2_file) is True

    # However, if we turn off validate, we do a full sweep because these
    # unknown files are lingering in our directory space
    assert pc.delete(validate=False)
    assert os.path.exists(cust1_file) is False
    assert os.path.exists(cust2_file) is False

    pc['key'] = 'value'
    pc['key2'] = 'value2'
    assert 'key' in pc
    assert 'key2' in pc
    pc.clear('key')
    assert 'key' not in pc
    assert 'key2' in pc

    # Set expired content
    pc.set(
        'expired', 'expired-content',
        expires=datetime.now() - timedelta(days=1))

    # It's actually there... but it's expired so our persistent
    # storage is behaving as it should
    assert 'expired' not in pc
    assert pc.get('expired') is None
    # Prune our content
    pc.prune()


def test_persistent_storage_corruption_handling(tmpdir):
    """
    Test corrupting handling of storage
    """

    # Namespace
    namespace = 'def456'

    # Initialize it
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    cache_file = pc.cache_file
    assert not os.path.isfile(cache_file)

    # Store our key
    pc['mykey'] = 42
    assert os.path.isfile(cache_file)

    with gzip.open(cache_file, 'rb') as f:
        # Read our content from disk
        json.loads(f.read().decode('utf-8'))

    # Remove object
    del pc

    # Corrupt the file
    with open(cache_file, 'wb') as f:
        f.write(b'{')

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # File is corrupted
    assert 'mykey' not in pc
    pc['mykey'] = 42
    del pc

    # File is corrected now
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    assert 'mykey' in pc

    # Corrupt the file again
    with gzip.open(cache_file, 'wb') as f:
        # Bad JSON File
        f.write(b'{')

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # File is corrupted
    assert 'mykey' not in pc
    pc['mykey'] = 42
    del pc

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # Test our force flush
    assert pc.flush(force=True) is True
    # double call
    assert pc.flush(force=True) is True

    # Zlib error handling as well during open
    with mock.patch('gzip.open', side_effect=OSError()):
        with pytest.raises(KeyError):
            pc['mykey'] = 43

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # Zlib error handling as well during open
    with mock.patch('gzip.open', side_effect=OSError()):
        # No keys can be returned
        assert not pc.keys()

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    with mock.patch('json.loads', side_effect=TypeError()):
        with mock.patch('os.unlink', side_effect=FileNotFoundError()):
            with pytest.raises(KeyError):
                pc['mykey'] = 44

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    with mock.patch('json.loads', side_effect=TypeError()):
        with mock.patch('os.unlink', side_effect=OSError()):
            with pytest.raises(KeyError):
                pc['mykey'] = 45

    pc['my-new-key'] = 43
    with mock.patch('gzip.open', side_effect=OSError()):
        # We will fail to flush our content to disk
        assert pc.flush(force=True) is False

    with mock.patch('json.dumps', side_effect=TypeError()):
        # We will fail to flush our content to disk
        assert pc.flush(force=True) is False

    with mock.patch('os.makedirs', side_effect=OSError()):
        pc = PersistentStore(
            namespace=namespace, path=str(tmpdir),
            mode=PersistentStoreMode.FLUSH)

        # Directory initialization failed so we fall back to memory mode
        assert pc.mode == PersistentStoreMode.MEMORY

    # Handle file updates
    pc = PersistentStore(
        namespace='file-time-refresh', path=str(tmpdir),
        mode=PersistentStoreMode.AUTO)

    pc['test'] = 'abcd'
    assert pc.write(b'data', key='abcd') is True
    assert pc.read('abcd', expires=True) == b'data'
    assert pc.write(b'data2', key='defg') is True
    assert pc.read('defg', expires=False) == b'data2'
    assert pc.write(b'data3', key='hijk') is True
    assert pc.read('hijk', expires=False) == b'data3'
    assert pc['test'] == 'abcd'

    with mock.patch('os.utime', side_effect=(OSError(), FileNotFoundError())):
        pc.flush()

    # directory initialization okay
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    assert 'mykey' not in pc
    pc['mykey'] = 42
    del pc

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)
    assert 'mykey' in pc

    # Remove the last entry
    del pc['mykey']
    with mock.patch('os.rename', side_effect=OSError()):
        with mock.patch('os.unlink', side_effect=OSError()):
            assert not pc.flush(force=True)

    # Create another entry
    pc['mykey'] = 42
    with mock.patch('tempfile.NamedTemporaryFile', side_effect=OSError()):
        assert not pc.flush(force=True)

        # Temporary file cleanup failure
        with mock.patch('tempfile._TemporaryFileWrapper.close',
                        side_effect=OSError()):
            assert not pc.flush(force=True)

    # Create another entry
    pc['mykey'] = 43
    mock_ntf = mock.MagicMock()
    mock_ntf.name = os.path.join(tmpdir, 'file')

    #
    # Recursion loop checking
    #
    with mock.patch(
            'tempfile.NamedTemporaryFile',
            side_effect=[FileNotFoundError(), FileNotFoundError(), mock_ntf]):
        # No way to have recursion loop
        assert not pc.flush(force=True, _recovery=True)

    with mock.patch(
            'tempfile.NamedTemporaryFile',
            side_effect=[FileNotFoundError(), FileNotFoundError(), mock_ntf]):
        # No way to have recursion loop
        assert not pc.flush(force=False, _recovery=True)

    with mock.patch(
            'tempfile.NamedTemporaryFile',
            side_effect=[FileNotFoundError(), FileNotFoundError(), mock_ntf]):
        # No way to have recursion loop
        assert not pc.flush(force=False, _recovery=False)

    with mock.patch(
            'tempfile.NamedTemporaryFile',
            side_effect=[FileNotFoundError(), FileNotFoundError(), mock_ntf]):
        # No way to have recursion loop
        assert not pc.flush(force=True, _recovery=False)

    with mock.patch('tempfile._TemporaryFileWrapper.close',
                    side_effect=(OSError(), None)):
        with mock.patch('os.unlink', side_effect=(OSError())):
            assert not pc.flush(force=True)

    with mock.patch(
            'tempfile._TemporaryFileWrapper.close', side_effect=OSError()):
        assert not pc.flush(force=True)

    with mock.patch(
            'tempfile._TemporaryFileWrapper.close',
            side_effect=(OSError(), None)):
        with mock.patch('os.unlink', side_effect=OSError()):
            assert not pc.flush(force=True)

    with mock.patch(
            'tempfile._TemporaryFileWrapper.close',
            side_effect=(OSError(), None)):
        with mock.patch('os.unlink', side_effect=FileNotFoundError()):
            assert not pc.flush(force=True)

    del pc

    # directory initialization okay
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # Allows us to play with encoding errors
    pc.encoding = 'ascii'

    # Handle write() calls
    with mock.patch('os.stat', side_effect=OSError()):
        # We fail to fetch the filesize of our old file causing us to fail
        assert pc.write('abcd') is False

    # ボールト translates to vault (no bad word here) :)
    data = "ボールト"

    # We'll have encoding issues
    assert pc.write(data) is False

    with mock.patch('gzip.open', side_effect=FileNotFoundError()):
        pc = PersistentStore(namespace=namespace, path=str(tmpdir))

        # recovery mode will kick in and even it will fail
        assert pc.write(b'key') is False

    with mock.patch('gzip.open', side_effect=OSError()):
        pc = PersistentStore(namespace=namespace, path=str(tmpdir))

        # Falls to default
        assert pc.get('key') is None

        pc = PersistentStore(namespace=namespace, path=str(tmpdir))
        with pytest.raises(KeyError):
            pc['key'] = 'value'

        pc = PersistentStore(namespace=namespace, path=str(tmpdir))
        with pytest.raises(KeyError):
            pc['key']

        pc = PersistentStore(namespace=namespace, path=str(tmpdir))
        with pytest.raises(KeyError):
            del pc['key']

        pc = PersistentStore(namespace=namespace, path=str(tmpdir))
        # Fails to set key
        assert pc.set('key', 'value') is False

        pc = PersistentStore(namespace=namespace, path=str(tmpdir))
        # Fails to clear
        assert pc.clear() is False

        pc = PersistentStore(namespace=namespace, path=str(tmpdir))
        # Fails to prune
        assert pc.prune() is False

    # Set some expired content
    pc.set(
        'key', 'value', persistent=False,
        expires=datetime.now() - timedelta(days=1))
    pc.set(
        'key2', 'value2', persistent=True,
        expires=datetime.now() - timedelta(days=1))

    # Set some un-expired content
    pc.set('key3', 'value3', persistent=True)
    pc.set('key4', 'value4', persistent=False)
    assert pc.prune() is True

    # Second call has no change made
    assert pc.prune() is False

    # Reset
    pc.delete()

    # directory initialization okay
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # Write some content that expires almost immediately
    pc.set(
        'key1', 'value', persistent=True,
        expires=datetime.now() + timedelta(seconds=1))
    pc.set(
        'key2', 'value', persistent=True,
        expires=datetime.now() + timedelta(seconds=1))
    pc.set(
        'key3', 'value', persistent=True,
        expires=datetime.now() + timedelta(seconds=1))
    pc.flush()

    # Wait out our expiry
    time.sleep(1.3)

    # now initialize our storage again
    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # This triggers our __load_cache() which reads in a value
    # determined to have already been expired
    assert 'key1' not in pc
    assert 'key2' not in pc
    assert 'key3' not in pc

    # Sweep
    pc.delete()
    pc.set('key', 'value')
    pc.set('key2', 'value2')
    pc.write('more-content')
    # Flush our content to disk
    pc.flush()

    # Ideally we'd use os.stat below, but it is called inside a list
    # comprehension block and mock doesn't appear to throw the exception
    # there.  So this is a bit of a cheat, but it works
    with mock.patch('builtins.sum', side_effect=OSError()):
        assert pc.size(exclude=True, lazy=False) == 0
        assert pc.size(exclude=False, lazy=False) == 0

    pc = PersistentStore(namespace=namespace, path=str(tmpdir))
    with mock.patch('glob.glob', side_effect=OSError()):
        assert pc.files(exclude=True, lazy=False) == []
        assert pc.files(exclude=False, lazy=False) == []

    pc = PersistentStore(
        namespace=namespace, path=str(tmpdir),
        mode=PersistentStoreMode.FLUSH)

    # Causes an initialization
    pc['abc'] = 1
    with mock.patch('os.unlink', side_effect=OSError()):
        # Now we can't set data
        with pytest.raises(KeyError):
            pc['new-key'] = 'value'
        # However keys that alrady exist don't get caught in check
        # and therefore won't throw
        pc['abc'] = 'value'

    #
    # Handles flush() when the queue is empty
    #
    pc.clear()
    with mock.patch('os.unlink', side_effect=OSError()):
        # We can't remove backup cache file
        assert pc.flush(force=True) is False

    with mock.patch('os.unlink', side_effect=FileNotFoundError()):
        # FileNotFound is not an issue
        assert pc.flush(force=True) is True

    with mock.patch('os.rename', side_effect=OSError()):
        # We can't create a backup
        assert pc.flush(force=True) is False

    with mock.patch('os.rename', side_effect=FileNotFoundError()):
        # FileNotFound is not an issue
        assert pc.flush(force=True) is True

    # Flush any previous cache and data
    pc.delete()

    #
    # Handles flush() cases where is data to write
    #

    # Create a key
    pc.set('abc', 'a-test-value')
    with mock.patch(
            'os.unlink', side_effect=(OSError(), None)):
        # We failed to move our content in place
        assert pc.flush(force=True) is False

    with mock.patch(
            'os.unlink', side_effect=(OSError(), FileNotFoundError())):
        # We failed to move our content in place
        assert pc.flush(force=True) is False

    with mock.patch(
            'os.unlink', side_effect=(OSError(), OSError())):
        # We failed to move our content in place
        assert pc.flush(force=True) is False


def test_persistent_custom_io(tmpdir):
    """
    Test reading and writing custom files
    """

    # Initialize it for memory only
    pc = PersistentStore(path=str(tmpdir))

    with pytest.raises(AttributeError):
        pc.open('!invalid#-Key')

    # We can't open the file as it does not exist
    with pytest.raises(FileNotFoundError):
        pc.open('valid-key')

    with pytest.raises(AttributeError):
        # Bad data
        pc.open(1234)

    with pytest.raises(FileNotFoundError):
        with pc.open('key') as fd:
            pass

    # Also can be caught using Apprise Exception Handling
    with pytest.raises(exception.AppriseFileNotFound):
        with pc.open('key') as fd:
            pass

    # Write some valid data
    with pc.open('new-key', 'wb') as fd:
        fd.write(b'data')

    with mock.patch("builtins.open", new_callable=mock.mock_open,
                    read_data="mocked file content") as mock_file:
        mock_file.side_effect = OSError
        with pytest.raises(exception.AppriseDiskIOError):
            with pc.open('new-key', compress=False) as fd:
                pass

    # Again but with compression this time
    with mock.patch("gzip.open", new_callable=mock.mock_open,
                    read_data="mocked file content") as mock_file:
        mock_file.side_effect = OSError
        with pytest.raises(exception.AppriseDiskIOError):
            with pc.open('new-key', compress=True) as fd:
                pass

    # Zlib error handling as well during open
    with mock.patch("gzip.open", new_callable=mock.mock_open,
                    read_data="mocked file content") as mock_file:
        mock_file.side_effect = zlib.error
        with pytest.raises(exception.AppriseDiskIOError):
            with pc.open('new-key', compress=True) as fd:
                pass

    # Writing
    with pytest.raises(AttributeError):
        pc.write(1234)

    with pytest.raises(AttributeError):
        pc.write(None)

    with pytest.raises(AttributeError):
        pc.write(True)

    pc = PersistentStore(str(tmpdir))
    with pc.open('key', 'wb') as fd:
        fd.write(b'test')
        fd.close()

    # Handle error capuring when failing to write to disk
    with mock.patch("gzip.open", new_callable=mock.mock_open,
                    read_data="mocked file content") as mock_file:
        mock_file.side_effect = zlib.error

        # We fail to write to disk
        assert pc.write(b'test') is False

        # We support other errors too
        mock_file.side_effect = OSError
        assert pc.write(b'test') is False

    with pytest.raises(AttributeError):
        pc.write(b'data', key='!invalid#-Key')

    pc.delete()
    with mock.patch('os.unlink', side_effect=OSError()):
        # Write our data and the __move() will fail under the hood
        assert pc.write(b'test') is False

    pc.delete()
    with mock.patch('os.rename', side_effect=OSError()):
        # Write our data and the __move() will fail under the hood
        assert pc.write(b'test') is False

    pc.delete()
    with mock.patch('os.unlink', side_effect=(OSError(), FileNotFoundError())):
        # Write our data and the __move() will fail under the hood
        assert pc.write(b'test') is False

    pc.delete()
    with mock.patch('os.unlink', side_effect=(OSError(), None)):
        # Write our data and the __move() will fail under the hood
        assert pc.write(b'test') is False

    pc.delete()
    with mock.patch('os.unlink', side_effect=(OSError(), OSError())):
        # Write our data and the __move() will fail under the hood
        assert pc.write(b'test') is False

    pc.delete()
    with mock.patch('os.rename', side_effect=(None, OSError(), None)):
        assert pc.write(b'test') is False

    with mock.patch('os.rename', side_effect=(None, OSError(), OSError())):
        assert pc.write(b'test') is False

    with mock.patch('os.rename', side_effect=(
            None, OSError(), FileNotFoundError())):
        assert pc.write(b'test') is False

    pc.delete()
    with mock.patch('os.rename', side_effect=(None, None, None, OSError())):
        # not enough reason to fail
        assert pc.write(b'test') is True

    with mock.patch('os.stat', side_effect=OSError()):
        with mock.patch('os.close', side_effect=(None, OSError())):
            assert pc.write(b'test') is False

    pc.delete()
    with mock.patch(
            'tempfile._TemporaryFileWrapper.close', side_effect=OSError()):
        assert pc.write(b'test') is False

    pc.delete()
    with mock.patch(
            'tempfile._TemporaryFileWrapper.close',
            side_effect=(OSError(), None)):
        with mock.patch('os.unlink', side_effect=OSError()):
            assert pc.write(b'test') is False

    pc.delete()
    with mock.patch(
            'tempfile._TemporaryFileWrapper.close',
            side_effect=(OSError(), None)):
        with mock.patch('os.unlink', side_effect=FileNotFoundError()):
            assert pc.write(b'test') is False


def test_persistent_storage_cache_object(tmpdir):
    """
    General testing of a CacheObject
    """
    # A cache object
    c = CacheObject(123)

    ref = datetime.now(tz=timezone.utc)
    expires = ref + timedelta(days=1)
    # Create a cache object that expires tomorrow
    c = CacheObject('abcd', expires=expires)
    assert c.expires == expires
    assert c.expires_sec > 86390.0 and c.expires_sec <= 86400.0
    assert bool(c) is True
    assert 'never' not in str(c)
    assert 'str:+:abcd' in str(c)

    #
    # Testing CacheObject.set()
    #
    c.set(123)
    assert 'never' not in str(c)
    assert 'int:+:123' in str(c)
    hash_value = c.hash()
    assert isinstance(hash_value, str)

    c.set(124)
    assert 'never' not in str(c)
    assert 'int:+:124' in str(c)
    assert c.hash() != hash_value

    c.set(123)
    # sha is the same again if we set the value back
    assert c.hash() == hash_value

    c.set(124)
    assert isinstance(c.hash(), str)
    assert c.value == 124
    assert bool(c) is True
    c.set(124, expires=False, persistent=False)
    assert bool(c) is True
    assert c.expires is None
    assert c.expires_sec is None
    c.set(124, expires=True)
    # we're expired now
    assert bool(c) is False

    #
    # Testing CacheObject equality (==)
    #
    a = CacheObject('abc')
    b = CacheObject('abc')

    assert a == b
    assert a == 'abc'
    assert b == 'abc'

    # Equality is no longer a thing
    b = CacheObject('abc', 30)
    assert a != b
    # however we can look at the value inside
    assert a == b.value

    b = CacheObject('abc', persistent=False)
    a = CacheObject('abc', persistent=True)
    # Persistent flag matters
    assert a != b
    # however we can look at the value inside
    assert a == b.value
    b = CacheObject('abc', persistent=True)
    assert a == b

    # Epoch
    EPOCH = datetime(1970, 1, 1)

    # test all of our supported types (also test time naive and aware times)
    for entry in ('string', 123, 1.2222, datetime.now(),
                  datetime.now(tz=timezone.utc), None, False, True, b'\0'):
        # Create a cache object that expires tomorrow
        c = CacheObject(entry, datetime.now() + timedelta(days=1))

        # Verify our content hasn't expired
        assert c

        # Verify we can dump our object
        result = json.loads(json.dumps(
            c, separators=(',', ':'), cls=CacheJSONEncoder))

        # Instantiate our object
        cc = CacheObject.instantiate(result)
        assert cc.json() == c.json()

    # Test our JSON Encoder against items we don't support
    with pytest.raises(TypeError):
        json.loads(json.dumps(
            object(), separators=(',', ':'), cls=CacheJSONEncoder))

    assert CacheObject.instantiate(None) is None
    assert CacheObject.instantiate({}) is None

    # Bad data
    assert CacheObject.instantiate({
        'v': 123,
        'x': datetime.now(),
        'c': 'int'}) is None

    # object type is not supported
    assert CacheObject.instantiate({
        'v': 123,
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': object}) is None

    obj = CacheObject.instantiate({
        'v': 123,
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': 'int'}, verify=False)
    assert isinstance(obj, CacheObject)
    assert obj.value == 123

    # no HASH and verify is set to true; our checksum will fail
    assert CacheObject.instantiate({
        'v': 123,
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': 'int'}, verify=True) is None

    # We can't instantiate our object if the expiry value is bad
    assert CacheObject.instantiate({
        'v': 123,
        'x': 'garbage',
        'c': 'int'}, verify=False) is None

    # We need a valid hash sum too
    assert CacheObject.instantiate({
        'v': 123,
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': 'int',
        # Expecting a valid sha string
        '!': 1.0}, verify=False) is None

    # Our Bytes Object with corruption
    assert CacheObject.instantiate({
        'v': 'garbage',
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': 'bytes'}, verify=False) is None

    obj = CacheObject.instantiate({
        'v': 'AA==',
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': 'bytes'}, verify=False)
    assert isinstance(obj, CacheObject)
    assert obj.value == b'\0'

    # Test our datetime objects
    obj = CacheObject.instantiate({
        'v': '2024-06-08T01:50:01.587267',
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': 'datetime'}, verify=False)
    assert isinstance(obj, CacheObject)
    assert obj.value == datetime(2024, 6, 8, 1, 50, 1, 587267)

    # A corrupt datetime object
    assert CacheObject.instantiate({
        'v': 'garbage',
        'x': (datetime.now() - EPOCH).total_seconds(),
        'c': 'datetime'}, verify=False) is None


@pytest.mark.skipif(
    sys.platform == "win32", reason="Unreliable results to be determined")
def test_persistent_storage_disk_prune(tmpdir):
    """
    General testing of a Persistent Store prune calls
    """

    # Persistent Storage Initialization
    pc = PersistentStore(
        path=str(tmpdir), namespace='t01', mode=PersistentStoreMode.FLUSH)
    # Store some data
    assert pc.write(b'data-t01') is True
    assert pc.set('key-t01', 'value')

    pc = PersistentStore(
        path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
    # Store some data
    assert pc.write(b'data-t02') is True
    assert pc.set('key-t02', 'value')

    # purne anything older then 30s
    results = PersistentStore.disk_prune(path=str(tmpdir), expires=30)
    # Nothing is older then 30s right now
    assert isinstance(results, dict)
    assert 't01' in results
    assert 't02' in results
    assert len(results['t01']) == 0
    assert len(results['t02']) == 0

    pc = PersistentStore(
        path=str(tmpdir), namespace='t01', mode=PersistentStoreMode.FLUSH)

    # Nothing is pruned
    assert pc.get('key-t01') == 'value'
    assert pc.read() == b'data-t01'

    # An expiry of zero gets everything
    # Note: This test randomly fails in Microsoft Windows for unknown reasons
    # When this is determined, this test can be opened back up
    results = PersistentStore.disk_prune(path=str(tmpdir), expires=0)
    # We match everything now
    assert isinstance(results, dict)
    assert 't01' in results
    assert 't02' in results
    assert len(results['t01']) == 2
    assert len(results['t02']) == 2

    # Content is still not removed however because no action was put in place
    pc = PersistentStore(
        path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
    # Nothing is pruned
    assert pc.get('key-t02') == 'value'
    assert pc.read() == b'data-t02'
    pc = PersistentStore(
        path=str(tmpdir), namespace='t01', mode=PersistentStoreMode.FLUSH)
    # Nothing is pruned
    assert pc.get('key-t01') == 'value'
    assert pc.read() == b'data-t01'

    with mock.patch('os.listdir', side_effect=OSError()):
        results = PersistentStore.disk_scan(
            namespace='t01', path=str(tmpdir), closest=True)
        assert isinstance(results, list)
        assert len(results) == 0

    with mock.patch('os.listdir', side_effect=FileNotFoundError()):
        results = PersistentStore.disk_scan(
            namespace='t01', path=str(tmpdir), closest=True)
        assert isinstance(results, list)
        assert len(results) == 0

        # Without closest flag
        results = PersistentStore.disk_scan(
            namespace='t01', path=str(tmpdir), closest=False)
        assert isinstance(results, list)
        assert len(results) == 0

    # Now we'll filter on specific namespaces
    results = PersistentStore.disk_prune(
        namespace='notfound', path=str(tmpdir), expires=0, action=True)

    # nothing matched, nothing found
    assert isinstance(results, dict)
    assert len(results) == 0

    results = PersistentStore.disk_prune(
        namespace=('t01', 'invalid', '-garbag!'),
        path=str(tmpdir), expires=0, action=True)

    # only t01 would be cleaned now
    assert isinstance(results, dict)
    assert len(results) == 1
    assert len(results['t01']) == 2

    # A second call will yield no results because the content has
    # already been cleaned up
    results = PersistentStore.disk_prune(
        namespace='t01',
        path=str(tmpdir), expires=0, action=True)
    assert isinstance(results, dict)
    assert len(results) == 0

    # t02 is still untouched
    pc = PersistentStore(
        path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
    # Nothing is pruned
    assert pc.get('key-t02') == 'value'
    assert pc.read() == b'data-t02'

    # t01 of course... it's gone
    pc = PersistentStore(
        path=str(tmpdir), namespace='t01', mode=PersistentStoreMode.FLUSH)
    # Nothing is pruned
    assert pc.get('key-t01') is None
    assert pc.read() is None

    with pytest.raises(AttributeError):
        # provide garbage in namespace field and we're going to have a problem
        PersistentStore.disk_prune(
            namespace=object, path=str(tmpdir), expires=0, action=True)

    # Error Handling
    with mock.patch('os.path.getmtime', side_effect=FileNotFoundError()):
        results = PersistentStore.disk_prune(
            namespace='t02', path=str(tmpdir), expires=0, action=True)
        assert isinstance(results, dict)
        assert len(results) == 1
        assert len(results['t02']) == 0

        # no files were removed, so our data is still accessible
        pc = PersistentStore(
            path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
        # Nothing is pruned
        assert pc.get('key-t02') == 'value'
        assert pc.read() == b'data-t02'

    with mock.patch('os.path.getmtime', side_effect=OSError()):
        results = PersistentStore.disk_prune(
            namespace='t02', path=str(tmpdir), expires=0, action=True)
        assert isinstance(results, dict)
        assert len(results) == 1
        assert len(results['t02']) == 0

        # no files were removed, so our data is still accessible
        pc = PersistentStore(
            path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
        # Nothing is pruned
        assert pc.get('key-t02') == 'value'
        assert pc.read() == b'data-t02'

    with mock.patch('os.unlink', side_effect=FileNotFoundError()):
        results = PersistentStore.disk_prune(
            namespace='t02', path=str(tmpdir), expires=0, action=True)
        assert isinstance(results, dict)
        assert len(results) == 1
        assert len(results['t02']) == 2

        # no files were removed, so our data is still accessible
        pc = PersistentStore(
            path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
        # Nothing is pruned
        assert pc.get('key-t02') == 'value'
        assert pc.read() == b'data-t02'

    with mock.patch('os.unlink', side_effect=OSError()):
        results = PersistentStore.disk_prune(
            namespace='t02', path=str(tmpdir), expires=0, action=True)
        assert isinstance(results, dict)
        assert len(results) == 1
        assert len(results['t02']) == 2

        # no files were removed, so our data is still accessible
        pc = PersistentStore(
            path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
        # Nothing is pruned
        assert pc.get('key-t02') == 'value'
        assert pc.read() == b'data-t02'

    with mock.patch('os.rmdir', side_effect=OSError()):
        results = PersistentStore.disk_prune(
            namespace='t02', path=str(tmpdir), expires=0, action=True)
        assert isinstance(results, dict)
        assert len(results) == 1
        assert len(results['t02']) == 2

        # no files were removed, so our data is still accessible
        pc = PersistentStore(
            path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
        # Nothing is pruned
        assert pc.get('key-t02') is None
        assert pc.read() is None


def test_persistent_storage_disk_changes(tmpdir):
    """
    General testing of a Persistent Store with underlining disk changes
    """

    # Create a garbage file in place of where the namespace should be
    tmpdir.join('t01').write('0' * 1024)

    # Persistent Storage Initialization where namespace directory now is
    # already occupied by a filename
    pc = PersistentStore(
        path=str(tmpdir), namespace='t01', mode=PersistentStoreMode.FLUSH)

    # Store some data and note that it isn't possible
    assert pc.write(b'data-t01') is False
    # We actually fell back to memory mode:
    assert pc.mode == PersistentStoreMode.MEMORY

    # Set's work
    assert pc.set('key-t01', 'value')

    # But upon reinitializtion (enforcing memory mode check) we will not have
    # the data available to us
    pc = PersistentStore(
        path=str(tmpdir), namespace='t01', mode=PersistentStoreMode.FLUSH)

    assert pc.get('key-t01') is None

    #
    # Test situation where the file structure changed after initialization
    #
    pc = PersistentStore(
        path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
    # Our mode stuck as t02 initialized correctly
    assert pc.mode == PersistentStoreMode.FLUSH
    assert os.path.isdir(pc.path)

    shutil.rmtree(pc.path)
    assert not os.path.isdir(pc.path)
    assert pc.set('key-t02', 'value')
    # The directory got re-created
    assert os.path.isdir(pc.path)

    # Same test but flag set to AUTO
    pc = PersistentStore(
        path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.AUTO)
    # Our mode stuck as t02 initialized correctly
    assert pc.mode == PersistentStoreMode.AUTO
    assert os.path.isdir(pc.path)

    shutil.rmtree(pc.path)
    assert not os.path.isdir(pc.path)
    assert pc.set('key-t02', 'value')
    # The directory is not recreated because of auto; it will occur on save
    assert not os.path.isdir(pc.path)
    path = pc.path
    del pc
    # It exists now
    assert os.path.isdir(path)

    pc = PersistentStore(
        path=str(tmpdir), namespace='t02', mode=PersistentStoreMode.FLUSH)
    # Content was not lost
    assert pc.get('key-t02') == 'value'

    # We'll remove a sub directory of it this time
    shutil.rmtree(os.path.join(pc.path, pc.temp_dir))

    # We will still successfully write our data
    assert pc.write(b'data-t02') is True
    assert os.path.isdir(pc.path)

    shutil.rmtree(pc.path)
    assert not os.path.isdir(pc.path)
    assert pc.set('key-t01', 'value')