Function test_persistent_storage_disk_prune

tests/test_persistent_store.py:1377–1588

General testing of a Persistent Store prune calls.

(tmpdir)

    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"