Testing a schedule cache across feed revisions
Include feed and revision in cache identity to isolate updated schedule data; demonstrate failed loads and pruning in memory.
The stale-value failure
A cache keyed only by a stop identifier can return an old schedule after a feed changes. The same stop name may appear in multiple feeds, and a single feed can publish a new revision with different times. Reusing an entry without accounting for those identities risks serving the wrong value.
The supplied cache uses a tuple of feed, revision and stop as its key. This makes each combination distinct: a new revision gets its own entry, while another feed using the same revision and stop remains isolated. Feed revision information is available in the GTFS schedule context, but the example treats its identifiers as caller-supplied strings [1].
Observe reuse and isolation
On a cache miss, get calls the supplied loader and stores its output as a tuple. A later request for the same feed, revision and stop returns the stored tuple without calling the loader again. The test uses a loader that raises if called, so it directly checks that the existing value is reused.
The test then requests revision r2 for the same stop and confirms the new time is stored separately. It requests the same revision and stop under a different feed and gets an independent value. These assertions demonstrate key identity and basic reuse for this synchronous, in-memory implementation.
"""Python 3.10+; demonstrate revision-scoped keys and pruning."""
import unittest
class FeedCache:
def __init__(self):
self.values = {}
def get(self, feed, revision, stop, loader):
key = (feed, revision, stop)
if key not in self.values:
self.values[key] = tuple(loader())
return self.values[key]
def prune(self, feed, keep_revision):
self.values = {key: value for key, value in self.values.items()
if key[0] != feed or key[1] == keep_revision}
class RevisionTests(unittest.TestCase):
def test_revision_and_feed_are_both_part_of_identity(self):
cache = FeedCache()
self.assertEqual(cache.get("bus", "r1", "S", lambda: ["10:00"]), ("10:00",))
def must_not_reload():
raise AssertionError("cache miss for an existing key")
self.assertEqual(cache.get("bus", "r1", "S", must_not_reload), ("10:00",))
self.assertEqual(cache.get("bus", "r2", "S", lambda: ["10:05"]), ("10:05",))
self.assertEqual(cache.get("rail", "r1", "S", lambda: ["11:00"]), ("11:00",))
cache.prune("bus", "r2")
self.assertNotIn(("bus", "r1", "S"), cache.values)
self.assertIn(("rail", "r1", "S"), cache.values)
def test_failed_load_is_not_cached(self):
cache = FeedCache()
def broken():
raise OSError("download failed")
with self.assertRaises(OSError):
cache.get("bus", "r1", "S", broken)
self.assertEqual(cache.values, {})
if __name__ == "__main__":
unittest.main()
Prune after choosing the active revision
The prune method removes entries for one feed unless they match the revision to keep. In the test, pruning bus entries to r2 removes its r1 value while preserving the rail feed’s r1 entry. This avoids deleting another feed’s data just because its revision label happens to be the same.
The example assumes the caller has already selected which revision should be retained. A production feed update may need to validate and load a candidate snapshot before switching the active revision, then coordinate the pointer change and cleanup. The in-memory method does not implement an atomic active-snapshot swap, synchronization, expiration or persistence.
Failed loads and cache boundaries
If the loader raises, the assignment to values does not happen, so the failed result is not cached. The second test checks that an OSError leaves the dictionary empty. A later call can retry the loader rather than receiving a saved failure. This behavior is useful when a transient failure should not poison the cache, although repeated callers could also trigger repeated failures.
Values are tuples of the loader’s results, which provides a shallow immutable container. The example’s values are strings, but a tuple does not make mutable objects nested inside it immutable. This is not an HTTP cache or a distributed cache: it has no request headers, freshness rules, process sharing or cross-machine consistency guarantees.
Run the local checks
Download the example as how-can-you-test-a-cache-after-a-feed-version-changes.py and run python3 how-can-you-test-a-cache-after-a-feed-version-changes.py. The implementation and tests use only the Python standard library and require Python 3.10 or newer. A successful run reports Ran 2 tests and OK.
The two local test methods cover revision and feed isolation, reuse, pruning, and a failed load. They do not establish correctness under concurrent access, behavior across process restarts, or safety of a real feed-activation workflow.