1import datetime 2import threading 3import time 4from datetime import timedelta 5from typing import List, Union, cast 6from unittest.mock import patch 7 8import pytest 9from django.core.cache import caches 10from pytest_django.fixtures import SettingsWrapper 11from pytest_mock import MockerFixture 12 13import django_redis.cache 14from django_redis.cache import RedisCache 15from django_redis.client import ShardClient, herd 16from django_redis.serializers.json import JSONSerializer 17from django_redis.serializers.msgpack import MSGPackSerializer 18 19herd.CACHE_HERD_TIMEOUT = 2 20 21 22class TestDjangoRedisCache: 23 def test_setnx(self, cache: RedisCache): 24 # we should ensure there is no test_key_nx in redis 25 cache.delete("test_key_nx") 26 res = cache.get("test_key_nx") 27 assert res is None 28 29 res = cache.set("test_key_nx", 1, nx=True) 30 assert bool(res) is True 31 # test that second set will have 32 res = cache.set("test_key_nx", 2, nx=True) 33 assert res is False 34 res = cache.get("test_key_nx") 35 assert res == 1 36 37 cache.delete("test_key_nx") 38 res = cache.get("test_key_nx") 39 assert res is None 40 41 def test_setnx_timeout(self, cache: RedisCache): 42 # test that timeout still works for nx=True 43 res = cache.set("test_key_nx", 1, timeout=2, nx=True) 44 assert res is True 45 time.sleep(3) 46 res = cache.get("test_key_nx") 47 assert res is None 48 49 # test that timeout will not affect key, if it was there 50 cache.set("test_key_nx", 1) 51 res = cache.set("test_key_nx", 2, timeout=2, nx=True) 52 assert res is False 53 time.sleep(3) 54 res = cache.get("test_key_nx") 55 assert res == 1 56 57 cache.delete("test_key_nx") 58 res = cache.get("test_key_nx") 59 assert res is None 60 61 def test_unicode_keys(self, cache: RedisCache): 62 cache.set("ключ", "value") 63 res = cache.get("ключ") 64 assert res == "value" 65 66 def test_save_and_integer(self, cache: RedisCache): 67 cache.set("test_key", 2) 68 res = cache.get("test_key", "Foo") 69 70 assert isinstance(res, int) 71 assert res == 2 72 73 def test_save_string(self, cache: RedisCache): 74 cache.set("test_key", "hello" * 1000) 75 res = cache.get("test_key") 76 77 assert isinstance(res, str) 78 assert res == "hello" * 1000 79 80 cache.set("test_key", "2") 81 res = cache.get("test_key") 82 83 assert isinstance(res, str) 84 assert res == "2" 85 86 def test_save_unicode(self, cache: RedisCache): 87 cache.set("test_key", "heló") 88 res = cache.get("test_key") 89 90 assert isinstance(res, str) 91 assert res == "heló" 92 93 def test_save_dict(self, cache: RedisCache): 94 if isinstance(cache.client._serializer, (JSONSerializer, MSGPackSerializer)): 95 # JSONSerializer and MSGPackSerializer use the isoformat for 96 # datetimes. 97 now_dt: Union[str, datetime.datetime] = datetime.datetime.now().isoformat() 98 else: 99 now_dt = datetime.datetime.now() 100 101 test_dict = {"id": 1, "date": now_dt, "name": "Foo"} 102 103 cache.set("test_key", test_dict) 104 res = cache.get("test_key") 105 106 assert isinstance(res, dict) 107 assert res["id"] == 1 108 assert res["name"] == "Foo" 109 assert res["date"] == now_dt 110 111 def test_save_float(self, cache: RedisCache): 112 float_val = 1.345620002 113 114 cache.set("test_key", float_val) 115 res = cache.get("test_key") 116 117 assert isinstance(res, float) 118 assert res == float_val 119 120 def test_timeout(self, cache: RedisCache): 121 cache.set("test_key", 222, timeout=3) 122 time.sleep(4) 123 124 res = cache.get("test_key") 125 assert res is None 126 127 def test_timeout_0(self, cache: RedisCache): 128 cache.set("test_key", 222, timeout=0) 129 res = cache.get("test_key") 130 assert res is None 131 132 def test_timeout_parameter_as_positional_argument(self, cache: RedisCache): 133 cache.set("test_key", 222, -1) 134 res = cache.get("test_key") 135 assert res is None 136 137 cache.set("test_key", 222, 1) 138 res1 = cache.get("test_key") 139 time.sleep(2) 140 res2 = cache.get("test_key") 141 assert res1 == 222 142 assert res2 is None 143 144 # nx=True should not overwrite expire of key already in db 145 cache.set("test_key", 222, None) 146 cache.set("test_key", 222, -1, nx=True) 147 res = cache.get("test_key") 148 assert res == 222 149 150 def test_timeout_negative(self, cache: RedisCache): 151 cache.set("test_key", 222, timeout=-1) 152 res = cache.get("test_key") 153 assert res is None 154 155 cache.set("test_key", 222, timeout=None) 156 cache.set("test_key", 222, timeout=-1) 157 res = cache.get("test_key") 158 assert res is None 159 160 # nx=True should not overwrite expire of key already in db 161 cache.set("test_key", 222, timeout=None) 162 cache.set("test_key", 222, timeout=-1, nx=True) 163 res = cache.get("test_key") 164 assert res == 222 165 166 def test_timeout_tiny(self, cache: RedisCache): 167 cache.set("test_key", 222, timeout=0.00001) 168 res = cache.get("test_key") 169 assert res in (None, 222) 170 171 def test_set_add(self, cache: RedisCache): 172 cache.set("add_key", "Initial value") 173 res = cache.add("add_key", "New value") 174 assert res is False 175 176 res = cache.get("add_key") 177 assert res == "Initial value" 178 res = cache.add("other_key", "New value") 179 assert res is True 180 181 def test_get_many(self, cache: RedisCache): 182 cache.set("a", 1) 183 cache.set("b", 2) 184 cache.set("c", 3) 185 186 res = cache.get_many(["a", "b", "c"]) 187 assert res == {"a": 1, "b": 2, "c": 3} 188 189 def test_get_many_unicode(self, cache: RedisCache): 190 cache.set("a", "1") 191 cache.set("b", "2") 192 cache.set("c", "3") 193 194 res = cache.get_many(["a", "b", "c"]) 195 assert res == {"a": "1", "b": "2", "c": "3"} 196 197 def test_set_many(self, cache: RedisCache): 198 cache.set_many({"a": 1, "b": 2, "c": 3}) 199 res = cache.get_many(["a", "b", "c"]) 200 assert res == {"a": 1, "b": 2, "c": 3} 201 202 def test_set_call_empty_pipeline(self, cache: RedisCache, mocker: MockerFixture): 203 if isinstance(cache.client, ShardClient): 204 pytest.skip("ShardClient doesn't support get_client") 205 206 pipeline = cache.client.get_client(write=True).pipeline() 207 key = "key" 208 value = "value" 209 210 mocked_set = mocker.patch.object(pipeline, "set") 211 cache.set(key, value, client=pipeline) 212 213 if isinstance(cache.client, herd.HerdClient): 214 default_timeout = cache.client._backend.default_timeout 215 herd_timeout = (default_timeout + herd.CACHE_HERD_TIMEOUT) * 1000 216 herd_pack_value = cache.client._pack(value, default_timeout) 217 mocked_set.assert_called_once_with( 218 cache.client.make_key(key, version=None), 219 cache.client.encode(herd_pack_value), 220 nx=False, 221 px=herd_timeout, 222 xx=False, 223 ) 224 else: 225 mocked_set.assert_called_once_with( 226 cache.client.make_key(key, version=None), 227 cache.client.encode(value), 228 nx=False, 229 px=cache.client._backend.default_timeout * 1000, 230 xx=False, 231 ) 232 233 def test_delete(self, cache: RedisCache): 234 cache.set_many({"a": 1, "b": 2, "c": 3}) 235 res = cache.delete("a") 236 assert bool(res) is True 237 238 res = cache.get_many(["a", "b", "c"]) 239 assert res == {"b": 2, "c": 3} 240 241 res = cache.delete("a") 242 assert bool(res) is False 243 244 @patch("django_redis.cache.DJANGO_VERSION", (3, 1, 0, "final", 0)) 245 def test_delete_return_value_type_new31(self, cache: RedisCache): 246 """delete() returns a boolean instead of int since django version 3.1""" 247 cache.set("a", 1) 248 res = cache.delete("a") 249 assert isinstance(res, bool) 250 assert res is True 251 res = cache.delete("b") 252 assert isinstance(res, bool) 253 assert res is False 254 255 @patch("django_redis.cache.DJANGO_VERSION", new=(3, 0, 1, "final", 0)) 256 def test_delete_return_value_type_before31(self, cache: RedisCache): 257 """delete() returns a int before django version 3.1""" 258 cache.set("a", 1) 259 res = cache.delete("a") 260 assert isinstance(res, int) 261 assert res == 1 262 res = cache.delete("b") 263 assert isinstance(res, int) 264 assert res == 0 265 266 def test_delete_many(self, cache: RedisCache): 267 cache.set_many({"a": 1, "b": 2, "c": 3}) 268 res = cache.delete_many(["a", "b"]) 269 assert bool(res) is True 270 271 res = cache.get_many(["a", "b", "c"]) 272 assert res == {"c": 3} 273 274 res = cache.delete_many(["a", "b"]) 275 assert bool(res) is False 276 277 def test_delete_many_generator(self, cache: RedisCache): 278 cache.set_many({"a": 1, "b": 2, "c": 3}) 279 res = cache.delete_many(key for key in ["a", "b"]) 280 assert bool(res) is True 281 282 res = cache.get_many(["a", "b", "c"]) 283 assert res == {"c": 3} 284 285 res = cache.delete_many(["a", "b"]) 286 assert bool(res) is False 287 288 def test_delete_many_empty_generator(self, cache: RedisCache): 289 res = cache.delete_many(key for key in cast(List[str], [])) 290 assert bool(res) is False 291 292 def test_incr(self, cache: RedisCache): 293 if isinstance(cache.client, herd.HerdClient): 294 pytest.skip("HerdClient doesn't support incr") 295 296 cache.set("num", 1) 297 298 cache.incr("num") 299 res = cache.get("num") 300 assert res == 2 301 302 cache.incr("num", 10) 303 res = cache.get("num") 304 assert res == 12 305 306 # max 64 bit signed int 307 cache.set("num", 9223372036854775807) 308 309 cache.incr("num") 310 res = cache.get("num") 311 assert res == 9223372036854775808 312 313 cache.incr("num", 2) 314 res = cache.get("num") 315 assert res == 9223372036854775810 316 317 cache.set("num", 3) 318 319 cache.incr("num", 2) 320 res = cache.get("num") 321 assert res == 5 322 323 def test_incr_no_timeout(self, cache: RedisCache): 324 if isinstance(cache.client, herd.HerdClient): 325 pytest.skip("HerdClient doesn't support incr") 326 327 cache.set("num", 1, timeout=None) 328 329 cache.incr("num") 330 res = cache.get("num") 331 assert res == 2 332 333 cache.incr("num", 10) 334 res = cache.get("num") 335 assert res == 12 336 337 # max 64 bit signed int 338 cache.set("num", 9223372036854775807, timeout=None) 339 340 cache.incr("num") 341 res = cache.get("num") 342 assert res == 9223372036854775808 343 344 cache.incr("num", 2) 345 res = cache.get("num") 346 assert res == 9223372036854775810 347 348 cache.set("num", 3, timeout=None) 349 350 cache.incr("num", 2) 351 res = cache.get("num") 352 assert res == 5 353 354 def test_incr_error(self, cache: RedisCache): 355 if isinstance(cache.client, herd.HerdClient): 356 pytest.skip("HerdClient doesn't support incr") 357 358 with pytest.raises(ValueError): 359 # key does not exist 360 cache.incr("numnum") 361 362 def test_incr_ignore_check(self, cache: RedisCache): 363 if isinstance(cache.client, ShardClient): 364 pytest.skip("ShardClient doesn't support argument ignore_key_check to incr") 365 if isinstance(cache.client, herd.HerdClient): 366 pytest.skip("HerdClient doesn't support incr") 367 368 # key exists check will be skipped and the value will be incremented by 369 # '1' which is the default delta 370 cache.incr("num", ignore_key_check=True) 371 res = cache.get("num") 372 assert res == 1 373 cache.delete("num") 374 375 # since key doesnt exist it is set to the delta value, 10 in this case 376 cache.incr("num", 10, ignore_key_check=True) 377 res = cache.get("num") 378 assert res == 10 379 cache.delete("num") 380 381 # following are just regression checks to make sure it still works as 382 # expected with incr max 64 bit signed int 383 cache.set("num", 9223372036854775807) 384 385 cache.incr("num", ignore_key_check=True) 386 res = cache.get("num") 387 assert res == 9223372036854775808 388 389 cache.incr("num", 2, ignore_key_check=True) 390 res = cache.get("num") 391 assert res == 9223372036854775810 392 393 cache.set("num", 3) 394 395 cache.incr("num", 2, ignore_key_check=True) 396 res = cache.get("num") 397 assert res == 5 398 399 def test_get_set_bool(self, cache: RedisCache): 400 cache.set("bool", True) 401 res = cache.get("bool") 402 403 assert isinstance(res, bool) 404 assert res is True 405 406 cache.set("bool", False) 407 res = cache.get("bool") 408 409 assert isinstance(res, bool) 410 assert res is False 411 412 def test_decr(self, cache: RedisCache): 413 if isinstance(cache.client, herd.HerdClient): 414 pytest.skip("HerdClient doesn't support decr") 415 416 cache.set("num", 20) 417 418 cache.decr("num") 419 res = cache.get("num") 420 assert res == 19 421 422 cache.decr("num", 20) 423 res = cache.get("num") 424 assert res == -1 425 426 cache.decr("num", 2) 427 res = cache.get("num") 428 assert res == -3 429 430 cache.set("num", 20) 431 432 cache.decr("num") 433 res = cache.get("num") 434 assert res == 19 435 436 # max 64 bit signed int + 1 437 cache.set("num", 9223372036854775808) 438 439 cache.decr("num") 440 res = cache.get("num") 441 assert res == 9223372036854775807 442 443 cache.decr("num", 2) 444 res = cache.get("num") 445 assert res == 9223372036854775805 446 447 def test_version(self, cache: RedisCache): 448 cache.set("keytest", 2, version=2) 449 res = cache.get("keytest") 450 assert res is None 451 452 res = cache.get("keytest", version=2) 453 assert res == 2 454 455 def test_incr_version(self, cache: RedisCache): 456 cache.set("keytest", 2) 457 cache.incr_version("keytest") 458 459 res = cache.get("keytest") 460 assert res is None 461 462 res = cache.get("keytest", version=2) 463 assert res == 2 464 465 def test_ttl_incr_version_no_timeout(self, cache: RedisCache): 466 cache.set("my_key", "hello world!", timeout=None) 467 468 cache.incr_version("my_key") 469 470 my_value = cache.get("my_key", version=2) 471 472 assert my_value == "hello world!" 473 474 def test_delete_pattern(self, cache: RedisCache): 475 for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]: 476 cache.set(key, "foo") 477 478 res = cache.delete_pattern("*foo-a*") 479 assert bool(res) is True 480 481 keys = cache.keys("foo*") 482 assert set(keys) == {"foo-bb", "foo-bc"} 483 484 res = cache.delete_pattern("*foo-a*") 485 assert bool(res) is False 486 487 @patch("django_redis.cache.RedisCache.client") 488 def test_delete_pattern_with_custom_count(self, client_mock, cache: RedisCache): 489 for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]: 490 cache.set(key, "foo") 491 492 cache.delete_pattern("*foo-a*", itersize=2) 493 494 client_mock.delete_pattern.assert_called_once_with("*foo-a*", itersize=2) 495 496 @patch("django_redis.cache.RedisCache.client") 497 def test_delete_pattern_with_settings_default_scan_count( 498 self, client_mock, cache: RedisCache 499 ): 500 for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]: 501 cache.set(key, "foo") 502 expected_count = django_redis.cache.DJANGO_REDIS_SCAN_ITERSIZE 503 504 cache.delete_pattern("*foo-a*") 505 506 client_mock.delete_pattern.assert_called_once_with( 507 "*foo-a*", itersize=expected_count 508 ) 509 510 def test_close(self, cache: RedisCache, settings: SettingsWrapper): 511 settings.DJANGO_REDIS_CLOSE_CONNECTION = True 512 cache.set("f", "1") 513 cache.close() 514 515 def test_close_client(self, cache: RedisCache, mocker: MockerFixture): 516 mock = mocker.patch.object(cache.client, "close") 517 cache.close() 518 assert mock.called 519 520 def test_ttl(self, cache: RedisCache): 521 cache.set("foo", "bar", 10) 522 ttl = cache.ttl("foo") 523 524 if isinstance(cache.client, herd.HerdClient): 525 assert pytest.approx(ttl) == 12 526 else: 527 assert pytest.approx(ttl) == 10 528 529 # Test ttl None 530 cache.set("foo", "foo", timeout=None) 531 ttl = cache.ttl("foo") 532 assert ttl is None 533 534 # Test ttl with expired key 535 cache.set("foo", "foo", timeout=-1) 536 ttl = cache.ttl("foo") 537 assert ttl == 0 538 539 # Test ttl with not existent key 540 ttl = cache.ttl("not-existent-key") 541 assert ttl == 0 542 543 def test_pttl(self, cache: RedisCache): 544 545 # Test pttl 546 cache.set("foo", "bar", 10) 547 ttl = cache.pttl("foo") 548 549 # delta is set to 10 as precision error causes tests to fail 550 if isinstance(cache.client, herd.HerdClient): 551 assert pytest.approx(ttl, 10) == 12000 552 else: 553 assert pytest.approx(ttl, 10) == 10000 554 555 # Test pttl with float value 556 cache.set("foo", "bar", 5.5) 557 ttl = cache.pttl("foo") 558 559 if isinstance(cache.client, herd.HerdClient): 560 assert pytest.approx(ttl, 10) == 7500 561 else: 562 assert pytest.approx(ttl, 10) == 5500 563 564 # Test pttl None 565 cache.set("foo", "foo", timeout=None) 566 ttl = cache.pttl("foo") 567 assert ttl is None 568 569 # Test pttl with expired key 570 cache.set("foo", "foo", timeout=-1) 571 ttl = cache.pttl("foo") 572 assert ttl == 0 573 574 # Test pttl with not existent key 575 ttl = cache.pttl("not-existent-key") 576 assert ttl == 0 577 578 def test_persist(self, cache: RedisCache): 579 cache.set("foo", "bar", timeout=20) 580 assert cache.persist("foo") is True 581 582 ttl = cache.ttl("foo") 583 assert ttl is None 584 assert cache.persist("not-existent-key") is False 585 586 def test_expire(self, cache: RedisCache): 587 cache.set("foo", "bar", timeout=None) 588 assert cache.expire("foo", 20) is True 589 ttl = cache.ttl("foo") 590 assert pytest.approx(ttl) == 20 591 assert cache.expire("not-existent-key", 20) is False 592 593 def test_pexpire(self, cache: RedisCache): 594 cache.set("foo", "bar", timeout=None) 595 assert cache.pexpire("foo", 20500) is True 596 ttl = cache.pttl("foo") 597 # delta is set to 10 as precision error causes tests to fail 598 assert pytest.approx(ttl, 10) == 20500 599 assert cache.pexpire("not-existent-key", 20500) is False 600 601 def test_pexpire_at(self, cache: RedisCache): 602 603 # Test settings expiration time 1 hour ahead by datetime. 604 cache.set("foo", "bar", timeout=None) 605 expiration_time = datetime.datetime.now() + timedelta(hours=1) 606 assert cache.pexpire_at("foo", expiration_time) is True 607 ttl = cache.pttl("foo") 608 assert pytest.approx(ttl, 10) == timedelta(hours=1).total_seconds() 609 610 # Test settings expiration time 1 hour ahead by Unix timestamp. 611 cache.set("foo", "bar", timeout=None) 612 expiration_time = datetime.datetime.now() + timedelta(hours=2) 613 assert cache.pexpire_at("foo", int(expiration_time.timestamp() * 1000)) is True 614 ttl = cache.pttl("foo") 615 assert pytest.approx(ttl, 10) == timedelta(hours=2).total_seconds() * 1000 616 617 # Test settings expiration time 1 hour in past, which effectively 618 # deletes the key. 619 expiration_time = datetime.datetime.now() - timedelta(hours=2) 620 assert cache.pexpire_at("foo", expiration_time) is True 621 value = cache.get("foo") 622 assert value is None 623 624 expiration_time = datetime.datetime.now() + timedelta(hours=2) 625 assert cache.pexpire_at("not-existent-key", expiration_time) is False 626 627 def test_expire_at(self, cache: RedisCache): 628 629 # Test settings expiration time 1 hour ahead by datetime. 630 cache.set("foo", "bar", timeout=None) 631 expiration_time = datetime.datetime.now() + timedelta(hours=1) 632 assert cache.expire_at("foo", expiration_time) is True 633 ttl = cache.ttl("foo") 634 assert pytest.approx(ttl, 1) == timedelta(hours=1).total_seconds() 635 636 # Test settings expiration time 1 hour ahead by Unix timestamp. 637 cache.set("foo", "bar", timeout=None) 638 expiration_time = datetime.datetime.now() + timedelta(hours=2) 639 assert cache.expire_at("foo", int(expiration_time.timestamp())) is True 640 ttl = cache.ttl("foo") 641 assert pytest.approx(ttl, 1) == timedelta(hours=1).total_seconds() * 2 642 643 # Test settings expiration time 1 hour in past, which effectively 644 # deletes the key. 645 expiration_time = datetime.datetime.now() - timedelta(hours=2) 646 assert cache.expire_at("foo", expiration_time) is True 647 value = cache.get("foo") 648 assert value is None 649 650 expiration_time = datetime.datetime.now() + timedelta(hours=2) 651 assert cache.expire_at("not-existent-key", expiration_time) is False 652 653 def test_lock(self, cache: RedisCache): 654 lock = cache.lock("foobar") 655 lock.acquire(blocking=True) 656 657 assert cache.has_key("foobar") 658 lock.release() 659 assert not cache.has_key("foobar") 660 661 def test_lock_released_by_thread(self, cache: RedisCache): 662 lock = cache.lock("foobar", thread_local=False) 663 lock.acquire(blocking=True) 664 665 def release_lock(lock_): 666 lock_.release() 667 668 t = threading.Thread(target=release_lock, args=[lock]) 669 t.start() 670 t.join() 671 672 assert not cache.has_key("foobar") 673 674 def test_iter_keys(self, cache: RedisCache): 675 if isinstance(cache.client, ShardClient): 676 pytest.skip("ShardClient doesn't support iter_keys") 677 678 cache.set("foo1", 1) 679 cache.set("foo2", 1) 680 cache.set("foo3", 1) 681 682 # Test simple result 683 result = set(cache.iter_keys("foo*")) 684 assert result == {"foo1", "foo2", "foo3"} 685 686 def test_iter_keys_itersize(self, cache: RedisCache): 687 if isinstance(cache.client, ShardClient): 688 pytest.skip("ShardClient doesn't support iter_keys") 689 690 cache.set("foo1", 1) 691 cache.set("foo2", 1) 692 cache.set("foo3", 1) 693 694 # Test limited result 695 result = list(cache.iter_keys("foo*", itersize=2)) 696 assert len(result) == 3 697 698 def test_iter_keys_generator(self, cache: RedisCache): 699 if isinstance(cache.client, ShardClient): 700 pytest.skip("ShardClient doesn't support iter_keys") 701 702 cache.set("foo1", 1) 703 cache.set("foo2", 1) 704 cache.set("foo3", 1) 705 706 # Test generator object 707 result = cache.iter_keys("foo*") 708 next_value = next(result) 709 assert next_value is not None 710 711 def test_primary_replica_switching(self, cache: RedisCache): 712 if isinstance(cache.client, ShardClient): 713 pytest.skip("ShardClient doesn't support get_client") 714 715 cache = cast(RedisCache, caches["sample"]) 716 client = cache.client 717 client._server = ["foo", "bar"] 718 client._clients = ["Foo", "Bar"] 719 720 assert client.get_client(write=True) == "Foo" 721 assert client.get_client(write=False) == "Bar" 722 723 def test_touch_zero_timeout(self, cache: RedisCache): 724 cache.set("test_key", 222, timeout=10) 725 726 assert cache.touch("test_key", 0) is True 727 res = cache.get("test_key") 728 assert res is None 729 730 def test_touch_positive_timeout(self, cache: RedisCache): 731 cache.set("test_key", 222, timeout=10) 732 733 assert cache.touch("test_key", 2) is True 734 assert cache.get("test_key") == 222 735 time.sleep(3) 736 assert cache.get("test_key") is None 737 738 def test_touch_negative_timeout(self, cache: RedisCache): 739 cache.set("test_key", 222, timeout=10) 740 741 assert cache.touch("test_key", -1) is True 742 res = cache.get("test_key") 743 assert res is None 744 745 def test_touch_missed_key(self, cache: RedisCache): 746 assert cache.touch("test_key_does_not_exist", 1) is False 747 748 def test_touch_forever(self, cache: RedisCache): 749 cache.set("test_key", "foo", timeout=1) 750 result = cache.touch("test_key", None) 751 assert result is True 752 assert cache.ttl("test_key") is None 753 time.sleep(2) 754 assert cache.get("test_key") == "foo" 755 756 def test_touch_forever_nonexistent(self, cache: RedisCache): 757 result = cache.touch("test_key_does_not_exist", None) 758 assert result is False 759 760 def test_touch_default_timeout(self, cache: RedisCache): 761 cache.set("test_key", "foo", timeout=1) 762 result = cache.touch("test_key") 763 assert result is True 764 time.sleep(2) 765 assert cache.get("test_key") == "foo" 766 767 def test_clear(self, cache: RedisCache): 768 cache.set("foo", "bar") 769 value_from_cache = cache.get("foo") 770 assert value_from_cache == "bar" 771 cache.clear() 772 value_from_cache_after_clear = cache.get("foo") 773 assert value_from_cache_after_clear is None 774