mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 00:53:45 +08:00
firestar5683
22a8575199
Keep official openpilot ancestry and dependency licenses. Store resources directly in Git, remove LFS setup requirements, and leave the downloadable big model outside the source tree.
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
import unittest
|
|
import pickle
|
|
from tinygrad.helpers import diskcache_get, diskcache_put, diskcache, diskcache_clear
|
|
|
|
def remote_get(table,q,k): q.put(diskcache_get(table, k))
|
|
def remote_put(table,k,v): diskcache_put(table, k, v)
|
|
|
|
class DiskCache(unittest.TestCase):
|
|
def test_putget(self):
|
|
table = "test_putget"
|
|
diskcache_put(table, "hello", "world")
|
|
self.assertEqual(diskcache_get(table, "hello"), "world")
|
|
diskcache_put(table, "hello", "world2")
|
|
self.assertEqual(diskcache_get(table, "hello"), "world2")
|
|
|
|
def test_putcomplex(self):
|
|
table = "test_putcomplex"
|
|
diskcache_put(table, "k", ("complex", 123, "object"))
|
|
ret = diskcache_get(table, "k")
|
|
self.assertEqual(ret, ("complex", 123, "object"))
|
|
|
|
def test_getotherprocess(self):
|
|
table = "test_getotherprocess"
|
|
from multiprocessing import Process, Queue
|
|
diskcache_put(table, "k", "getme")
|
|
q = Queue()
|
|
p = Process(target=remote_get, args=(table,q,"k"))
|
|
p.start()
|
|
p.join()
|
|
self.assertEqual(q.get(), "getme")
|
|
|
|
def test_putotherprocess(self):
|
|
table = "test_putotherprocess"
|
|
from multiprocessing import Process
|
|
p = Process(target=remote_put, args=(table,"k", "remote"))
|
|
p.start()
|
|
p.join()
|
|
self.assertEqual(diskcache_get(table, "k"), "remote")
|
|
|
|
def test_no_table(self):
|
|
self.assertIsNone(diskcache_get("faketable", "k"))
|
|
|
|
def test_ret(self):
|
|
table = "test_ret"
|
|
self.assertEqual(diskcache_put(table, "key", ("vvs",)), ("vvs",))
|
|
|
|
def test_non_str_key(self):
|
|
table = "test_non_str_key"
|
|
diskcache_put(table, 4, 5)
|
|
self.assertEqual(diskcache_get(table, 4), 5)
|
|
self.assertEqual(diskcache_get(table, "4"), 5)
|
|
|
|
def test_decorator(self):
|
|
calls = 0
|
|
@diskcache
|
|
def hello(x):
|
|
nonlocal calls
|
|
calls += 1
|
|
return "world"+x
|
|
self.assertEqual(hello("bob"), "worldbob")
|
|
self.assertEqual(hello("billy"), "worldbilly")
|
|
kcalls = calls
|
|
self.assertEqual(hello("bob"), "worldbob")
|
|
self.assertEqual(hello("billy"), "worldbilly")
|
|
self.assertEqual(kcalls, calls)
|
|
|
|
def test_dict_key(self):
|
|
table = "test_dict_key"
|
|
fancy_key = {"hello": "world", "goodbye": 7, "good": True, "pkl": pickle.dumps("cat")}
|
|
fancy_key2 = {"hello": "world", "goodbye": 8, "good": True, "pkl": pickle.dumps("cat")}
|
|
fancy_key3 = {"hello": "world", "goodbye": 8, "good": True, "pkl": pickle.dumps("dog")}
|
|
diskcache_put(table, fancy_key, 5)
|
|
self.assertEqual(diskcache_get(table, fancy_key), 5)
|
|
diskcache_put(table, fancy_key2, 8)
|
|
self.assertEqual(diskcache_get(table, fancy_key2), 8)
|
|
self.assertEqual(diskcache_get(table, fancy_key), 5)
|
|
self.assertEqual(diskcache_get(table, fancy_key3), None)
|
|
|
|
def test_table_name(self):
|
|
table = "test_gfx1010:xnack-"
|
|
diskcache_put(table, "key", "test")
|
|
self.assertEqual(diskcache_get(table, "key"), "test")
|
|
|
|
@unittest.skip("disabled by default because this drops cache table")
|
|
def test_clear_cache(self):
|
|
# clear cache to start
|
|
diskcache_clear()
|
|
tables = [f"test_clear_cache:{i}" for i in range(3)]
|
|
for table in tables:
|
|
# check no entries
|
|
self.assertIsNone(diskcache_get(table, "k"))
|
|
for table in tables:
|
|
diskcache_put(table, "k", "test")
|
|
# check insertion
|
|
self.assertEqual(diskcache_get(table, "k"), "test")
|
|
|
|
diskcache_clear()
|
|
for table in tables:
|
|
# check no entries again
|
|
self.assertIsNone(diskcache_get(table, "k"))
|
|
|
|
# calling multiple times is fine
|
|
diskcache_clear()
|
|
diskcache_clear()
|
|
diskcache_clear()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|