|
4 | 4 |
|
5 | 5 | # pylint: disable-msg=protected-access |
6 | 6 |
|
7 | | -import os.path |
8 | 7 | import logging |
| 8 | +import os.path |
9 | 9 | from datetime import datetime |
| 10 | +from shutil import rmtree, copytree |
| 11 | + |
| 12 | +import pyppmsconf |
10 | 13 | import pytest |
11 | 14 | import requests.exceptions |
12 | | -from shutil import rmtree |
13 | 15 |
|
14 | | -import pyppmsconf |
15 | 16 | from pyppms import ppms |
16 | 17 |
|
17 | | - |
18 | 18 | # TODO: system ID is hard-coded here, so this will fail on any other instance! |
19 | 19 | __SYS_ID__ = 69 |
20 | 20 |
|
@@ -441,6 +441,15 @@ def test_get_systems_matching(ppms_connection, system_details_raw): |
441 | 441 | assert sys_ids == [] |
442 | 442 |
|
443 | 443 |
|
| 444 | +def test_get_systems_matching__raises(ppms_connection, system_details_raw): |
| 445 | + """Test get_systems_matching() with a wrong parameter type.""" |
| 446 | + with pytest.raises(TypeError): |
| 447 | + ppms_connection.get_systems_matching("foo", "wrong-type") |
| 448 | + |
| 449 | + with pytest.raises(TypeError): |
| 450 | + ppms_connection.get_systems_matching("foo", "") |
| 451 | + |
| 452 | + |
444 | 453 | ############ system / user permissions ############ |
445 | 454 |
|
446 | 455 |
|
@@ -654,49 +663,167 @@ def test_get_running_sheet_fail(ppms_connection): |
654 | 663 | assert ppms_connection.get_running_sheet("2", date=day) == [] |
655 | 664 |
|
656 | 665 |
|
657 | | -############ deprecated methods ############ |
| 666 | +############ cache ############ |
658 | 667 |
|
659 | 668 |
|
660 | | -def test__get_system_with_name(ppms_connection, system_details_raw): |
661 | | - """Test the (deprecated) _get_system_with_name() method.""" |
662 | | - logd("Testing with a well-known system name") |
663 | | - name = system_details_raw["Name"] |
664 | | - sys_id = ppms_connection._get_system_with_name(name) |
665 | | - print(f"_get_system_with_name: {sys_id}") |
666 | | - assert sys_id == int(system_details_raw["System id"]) |
| 669 | +def test_flush_cache(ppms_connection, caplog, tmp_path): |
| 670 | + """Test flushing the on-disk PyPPMS cache. |
667 | 671 |
|
668 | | - logd("Testing with a non-existing system name") |
669 | | - sys_id = ppms_connection._get_system_with_name("invalid-system-name") |
670 | | - assert sys_id == -1 |
| 672 | + - Make sure the temporary test-directory exists but doesn't contain a cache yet. |
| 673 | + - Copy over one of the cache directories provided with the tests. |
| 674 | + - Make sure the test-directory *does* contain a cache now. |
| 675 | + - Update the connection object's `cache_path` to point to the test location. |
| 676 | + - Trigger the `flush_cache()` method. |
| 677 | + - Verify the cache has been removed from the test-directory. |
| 678 | + """ |
| 679 | + orig_cache_path = os.path.join(pyppmsconf.CACHE_PATH, "stage_1") |
| 680 | + fresh_cache_path = tmp_path / "pyppms_cache" |
| 681 | + |
| 682 | + assert os.path.exists(tmp_path) |
| 683 | + assert os.path.exists(orig_cache_path) |
| 684 | + |
| 685 | + assert not os.path.exists(fresh_cache_path) |
| 686 | + copytree(orig_cache_path, fresh_cache_path) |
| 687 | + assert os.path.exists(fresh_cache_path) |
| 688 | + _logger.info("Cache path created: %s", fresh_cache_path) |
| 689 | + |
| 690 | + ppms_connection.cache_path = fresh_cache_path |
| 691 | + _logger.info("Updated connection cache path: %s", fresh_cache_path) |
| 692 | + ppms_connection.flush_cache() |
| 693 | + _logger.info("Flushed connection cache path: %s", fresh_cache_path) |
| 694 | + assert not os.path.exists(fresh_cache_path) |
| 695 | + |
| 696 | + |
| 697 | +def test_flush_cache__keep_users(ppms_connection, caplog, tmp_path): |
| 698 | + """Test flushing the on-disk PyPPMS cache while keeping the user details. |
| 699 | +
|
| 700 | + - Make sure the temporary test-directory exists but doesn't contain a cache yet. |
| 701 | + - Create the cache directory there. |
| 702 | + - Update the connection object's `cache_path` to point to the test location. |
| 703 | + - Copy over the subdirs listed in `to_keep` and `to_flush` from the cache provided |
| 704 | + with the tests. |
| 705 | + - Make sure the copied directories exist at the test-cache location. |
| 706 | + - Trigger the `flush_cache(keep_users=True)` method. |
| 707 | + - Verify the subdirs in `to_keep` have been retained at the test-directory. |
| 708 | + - Verify the subdirs in `to_flush` have been removed from the test-directory. |
| 709 | + """ |
| 710 | + to_keep = ["getuser"] |
| 711 | + to_flush = ["auth", "getgroups", "getusers", "getbooking"] |
671 | 712 |
|
| 713 | + orig_cache_root = os.path.join(pyppmsconf.CACHE_PATH, "stage_0") |
| 714 | + fresh_cache_path = tmp_path / "pyppms_cache" |
672 | 715 |
|
673 | | -def test__get_machine_catalogue_from_system(ppms_connection, system_details_raw): |
674 | | - """Test the (deprecated) _get_machine_catalogue_from_system() method.""" |
675 | | - name = system_details_raw["Name"] |
| 716 | + assert os.path.exists(tmp_path) |
| 717 | + assert os.path.exists(orig_cache_root) |
676 | 718 |
|
677 | | - # define a list of categories we are interested in: |
678 | | - categories = ["VDI (Development)", "VDI (CAD)", "VDI (BigMemory)"] |
| 719 | + assert not os.path.exists(fresh_cache_path) |
| 720 | + fresh_cache_path.mkdir() |
| 721 | + assert os.path.exists(fresh_cache_path) |
| 722 | + _logger.info("Cache path created: %s", fresh_cache_path) |
679 | 723 |
|
680 | | - # ask to which one the given machine belongs: |
681 | | - cat = ppms_connection._get_machine_catalogue_from_system(name, categories) |
| 724 | + ppms_connection.cache_path = fresh_cache_path |
| 725 | + _logger.info("Updated connection cache path: %s", fresh_cache_path) |
682 | 726 |
|
683 | | - # expected be the first one: |
684 | | - assert cat == categories[0] |
| 727 | + for subdir in to_keep + to_flush: |
| 728 | + srcdir = os.path.join(orig_cache_root, subdir) |
| 729 | + tgt_path = fresh_cache_path / subdir |
| 730 | + assert not os.path.exists(tgt_path) |
| 731 | + copytree(srcdir, tgt_path) |
| 732 | + _logger.info("Copied [%s] to [%s]", subdir, tgt_path) |
| 733 | + assert os.path.exists(tgt_path) |
685 | 734 |
|
686 | | - # test when no category is found: |
687 | | - cat = ppms_connection._get_machine_catalogue_from_system(name, categories[1:]) |
688 | | - assert cat == "" |
| 735 | + ppms_connection.flush_cache(keep_users=True) |
689 | 736 |
|
690 | | - # test with a system name that doesn't exist |
691 | | - name = "_invalid_pyppms_system_name_" |
692 | | - cat = ppms_connection._get_machine_catalogue_from_system(name, categories) |
693 | | - assert cat == "" |
| 737 | + for subdir in to_keep: |
| 738 | + tgt_path = fresh_cache_path / subdir |
| 739 | + _logger.debug("Verifying directory has been KEPT: %s", tgt_path) |
| 740 | + assert os.path.exists(tgt_path) |
694 | 741 |
|
| 742 | + for subdir in to_flush: |
| 743 | + tgt_path = fresh_cache_path / subdir |
| 744 | + _logger.debug("Verifying directory has been FLUSHED: %s", tgt_path) |
| 745 | + assert not os.path.exists(tgt_path) |
695 | 746 |
|
696 | | -def test_not_implemented_errors(ppms_connection): |
697 | | - """Test methods raising a NotImplementedError.""" |
698 | | - with pytest.raises(NotImplementedError): |
699 | | - ppms_connection.get_bookable_ids("", "") |
700 | 747 |
|
701 | | - with pytest.raises(NotImplementedError): |
702 | | - ppms_connection.get_system(99) |
| 748 | +@pytest.mark.online |
| 749 | +def test_flush_cache__keep_users__request_new(ppms_connection, caplog, tmp_path): |
| 750 | + """Test flush_cache() with `keep_users=True` and request a new user after. |
| 751 | +
|
| 752 | + This test has a huge overlap to the `test_flush_cache__keep_users()` one |
| 753 | + (that works offline, unlike this one) with the main difference being that |
| 754 | + after flushing the cache, the cached details of a known user are removed and |
| 755 | + then exactly those details are requested. This will trigger an online |
| 756 | + request for that user while requests for previously existing (cached and |
| 757 | + preserved by the `keep_users` option set to `True`) users will be served |
| 758 | + directly from the cache. |
| 759 | +
|
| 760 | + - Make sure the temporary test-directory exists and has no cache inside. |
| 761 | + - Create the cache directory there. |
| 762 | + - Update the connection object's `cache_path` to point to the test location. |
| 763 | + - Copy over the subdirs listed in `to_keep` and `to_flush` from the cache |
| 764 | + provided with the tests. |
| 765 | + - Trigger the `flush_cache(keep_users=True)` method. |
| 766 | + - Simulate a new user in PPMS that is not yet cached locally: |
| 767 | + - Remove a specific file of previously cached user details from the |
| 768 | + `getuser` cache. |
| 769 | + - |
| 770 | + - Then copy back the `getusers` cache which contains the list of usernames |
| 771 | + known to PPMS, which will include the username whose details have been |
| 772 | + removed explicitly in the previous step. |
| 773 | + - Request details of the preserved user, verify they're being served from |
| 774 | + the cache by inspecting the log messages. |
| 775 | + - Request details of the "pseudo-new" user, verify they're triggering an |
| 776 | + on-line request to PUMAPI. |
| 777 | + """ |
| 778 | + to_keep = ["getuser"] |
| 779 | + to_flush = ["auth", "getgroups", "getusers", "getbooking"] |
| 780 | + |
| 781 | + orig_cache_root = os.path.join(pyppmsconf.CACHE_PATH, "stage_0") |
| 782 | + fresh_cache_path = tmp_path / "pyppms_cache" |
| 783 | + |
| 784 | + assert os.path.exists(orig_cache_root) |
| 785 | + |
| 786 | + assert not os.path.exists(fresh_cache_path) |
| 787 | + fresh_cache_path.mkdir() |
| 788 | + assert os.path.exists(fresh_cache_path) |
| 789 | + _logger.info("Cache path created: %s", fresh_cache_path) |
| 790 | + |
| 791 | + ppms_connection.cache_path = fresh_cache_path |
| 792 | + _logger.info("Updated connection cache path: %s", fresh_cache_path) |
| 793 | + |
| 794 | + for subdir in to_keep + to_flush: |
| 795 | + srcdir = os.path.join(orig_cache_root, subdir) |
| 796 | + tgt_path = fresh_cache_path / subdir |
| 797 | + assert not os.path.exists(tgt_path) |
| 798 | + copytree(srcdir, tgt_path) |
| 799 | + _logger.info("Copied [%s] to [%s]", subdir, tgt_path) |
| 800 | + assert os.path.exists(tgt_path) |
| 801 | + |
| 802 | + ppms_connection.flush_cache(keep_users=True) |
| 803 | + |
| 804 | + new_user_name = "pyppms-adm" # simulated "new" user |
| 805 | + old_user_name = "pyppms" # previously existing, cached user (preserved) |
| 806 | + |
| 807 | + _logger.info("Removing preserved user-cache for [%s]...", new_user_name) |
| 808 | + new_user_cache = fresh_cache_path / "getuser" / f"login--{new_user_name}.txt" |
| 809 | + old_user_cache = fresh_cache_path / "getuser" / f"login--{old_user_name}.txt" |
| 810 | + assert os.path.isfile(new_user_cache) |
| 811 | + os.unlink(new_user_cache) |
| 812 | + assert not os.path.exists(new_user_cache) |
| 813 | + assert os.path.exists(old_user_cache) |
| 814 | + |
| 815 | + _logger.info("Restoring cache of existing user names...") |
| 816 | + users_list = os.path.join(orig_cache_root, "getusers") |
| 817 | + tgt_path = fresh_cache_path / "getusers" |
| 818 | + copytree(users_list, tgt_path) |
| 819 | + assert os.path.exists(tgt_path) |
| 820 | + _logger.info("Restored user names cache to [%s].", tgt_path) |
| 821 | + |
| 822 | + _logger.info("Requesting details from PUMAPI for cached user [%s]", old_user_name) |
| 823 | + ppms_connection.get_user(old_user_name) |
| 824 | + assert "No cache hit" not in caplog.text # served from the cache |
| 825 | + |
| 826 | + _logger.info("Requesting details from PUMAPI for 'new' user [%s]", new_user_name) |
| 827 | + ppms_connection.get_user(new_user_name) |
| 828 | + assert "No cache hit" in caplog.text # requires an on-line request |
| 829 | + assert os.path.exists(new_user_cache) |
0 commit comments