OneDrive & Outlook unit tests improvements

Author: vgrem

office365/outlook/mail/messages/message.py

@@ -17,6 +17,7 @@
 from office365.outlook.mail.messages.followup_flag import FollowupFlag
 from office365.outlook.mail.messages.internet_header import InternetMessageHeader
 from office365.outlook.mail.recipient import Recipient
+from office365.runtime.client_object import persist_property
 from office365.runtime.client_result import ClientResult
 from office365.runtime.client_value_collection import ClientValueCollection
 from office365.runtime.paths.resource_path import ResourcePath
@@ -229,10 +230,10 @@ def has_attachments(self):
         return self.properties.get("hasAttachments", None)
 
     @property
+    @persist_property()
     def attachments(self):
         # type: () -> AttachmentCollection
         """The fileAttachment and itemAttachment attachments for the message."""
-        self._persist_changes("attachments")
         return self.properties.setdefault(
             "attachments",
             AttachmentCollection(

office365/outlook/mail/messages/rules/rule.py

@@ -3,6 +3,7 @@
 from office365.entity import Entity
 from office365.outlook.mail.messages.rules.actions import MessageRuleActions
 from office365.outlook.mail.messages.rules.predicates import MessageRulePredicates
+from office365.runtime.client_object import persist_property
 
 
 class MessageRule(Entity):
@@ -12,6 +13,7 @@ class MessageRule(Entity):
     upon certain conditions."""
 
     @property
+    @persist_property()
     def actions(self):
         """Actions to be taken on a message when the corresponding conditions are fulfilled."""
         return self.properties.get("actions", MessageRuleActions())

office365/runtime/client_object.py

@@ -25,16 +25,34 @@
 """Property Type."""
 
 
+def persist_property(property_name=None):
+    # type: (Optional[str]) -> Callable[[Callable[..., P_T]], Callable[..., P_T]]
+    """Decorator to automatically track property access for persistence"""
+
+    def decorator(method):
+        # type: (Callable[..., P_T]) -> Callable[..., P_T]
+        def wrapper(self, *args, **kwargs):
+            # type: (ClientObject, *Any, **Any) -> P_T
+            name = property_name if property_name is not None else method.__name__
+            if name not in self._properties_to_persist:
+                self._properties_to_persist.append(name)
+            return method(self, *args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
 class ClientObject(Generic[T]):
     def __init__(self, context, resource_path=None, parent_collection=None):
         # type: (ClientRuntimeContext, Optional[ResourcePath], Optional[ClientObjectCollection]) -> None
         """Base client object which define named properties and relationships of an entity."""
-        self._properties = {}
-        self._properties_to_persist = []
+        self._properties = {}  # type: dict[str, Any]
+        self._properties_to_persist = []  # type: list[str]
         self._query_options = QueryOptions()
         self._parent_collection = parent_collection
         self._context = context
-        self._entity_type_name = None
+        self._entity_type_name = None  # type: Optional[str]
         self._resource_path = resource_path
 
     def clear_state(self):

tests/decorators.py

@@ -19,8 +19,8 @@ def _get_cached_permissions(client, client_id):
     return result.value
 
 
-def requires_app_permission(app_role):
-    # type: (str) -> Callable[[T], T]
+def requires_app_permission(*app_roles):
+    # type: (*str) -> Callable[[T], T]
     def decorator(test_method):
         # type: (T) -> T
         @wraps(test_method)
@@ -32,8 +32,9 @@ def wrapper(self, *args, **kwargs):
 
             permissions = _get_cached_permissions(client, test_client_id)
 
-            if not any(role.value == app_role for role in permissions):
-                self.skipTest(f"Required app permission '{app_role}' not granted")
+            if not any(role.value in app_roles for role in permissions):
+                required_roles = ", ".join(f"'{role}'" for role in app_roles)
+                self.skipTest(f"Required app permission '{required_roles}' not granted")
 
             return test_method(self, *args, **kwargs)
 

tests/onedrive/test_column.py

@@ -1,5 +1,6 @@
 from office365.onedrive.columns.definition import ColumnDefinition
 from tests import create_unique_name
+from tests.decorators import requires_delegated_permission
 from tests.graph_case import GraphTestCase
 
 
@@ -11,24 +12,33 @@ def setUpClass(cls):
         super(TestColumn, cls).setUpClass()
         cls.doclib = cls.client.sites.root.lists["Documents"]
 
-    def test1_get_list_columns(self):
+    @requires_delegated_permission(
+        "Sites.Read.All",
+        "Sites.Manage.All",
+        "Sites.FullControl.All",
+        "Sites.ReadWrite.All",
+    )
+    def test1_list_list_columns(self):
         columns = self.doclib.columns.get().execute_query()
         self.assertGreater(len(columns), 0)
 
-    def test2_create_text_column(self):
+    @requires_delegated_permission("Sites.Manage.All", "Sites.FullControl.All")
+    def test2_create_text_column_for_list(self):
         column_name = create_unique_name("TextColumn")
         column = self.doclib.columns.add_text(column_name).execute_query()
         self.assertIsNotNone(column.resource_path)
         self.__class__.list_columns.append(column)
 
-    def test3_create_lookup_column(self):
+    @requires_delegated_permission("Sites.Manage.All", "Sites.FullControl.All")
+    def test3_create_lookup_column_for_list(self):
         column_name = create_unique_name("LookupColumn")
         column = self.doclib.columns.add_lookup(
             column_name, self.doclib
         ).execute_query()
         self.assertIsNotNone(column.resource_path)
         self.__class__.list_columns.append(column)
 
+    @requires_delegated_permission("Sites.Manage.All", "Sites.FullControl.All")
     def test3_delete_list_columns(self):
         for col_to_del in self.__class__.list_columns:
             col_to_del.delete_object().execute_query()

tests/onedrive/test_content_type.py

@@ -4,6 +4,7 @@
 from office365.graph_client import GraphClient
 from office365.onedrive.contenttypes.content_type import ContentType
 from tests import test_client_id, test_password, test_tenant, test_username
+from tests.decorators import requires_delegated_permission
 
 
 class TestContentType(TestCase):
@@ -26,6 +27,7 @@ def test1_get_compatible_hub_content_types(self):
         )
         self.assertIsNotNone(cts.resource_path)
 
+    @requires_delegated_permission("Sites.Manage.All", "Sites.FullControl.All")
     def test2_create_site_content_type(self):
         name = "docSet" + uuid.uuid4().hex
         ct = self.client.sites.root.content_types.add(
@@ -34,19 +36,33 @@ def test2_create_site_content_type(self):
         self.assertIsNotNone(ct.resource_path)
         self.__class__.target_ct = ct
 
-    # def test3_publish_and_verify_if_published(self):
-    #    result = self.__class__.target_ct.publish().is_published().execute_query()
-    #    self.assertTrue(result.value)
+    @requires_delegated_permission("Sites.FullControl.All")
+    def test3_publish(self):
+        result = self.__class__.target_ct.publish().execute_query()
+        self.assertFalse(result.resource_path)
 
-    # def test4_unpublish(self):
-    #    result = self.__class__.target_ct.unpublish().is_published().execute_query()
-    #    self.assertFalse(result.value)
+    @requires_delegated_permission("Sites.FullControl.All")
+    def test4_is_published(self):
+        result = self.__class__.target_ct.is_published().execute_query()
+        self.assertTrue(result.value)
 
-    def test5_delete(self):
+    @requires_delegated_permission("Sites.FullControl.All")
+    def test5_unpublish(self):
+        result = self.__class__.target_ct.unpublish().is_published().execute_query()
+        self.assertFalse(result.value)
+
+    @requires_delegated_permission("Sites.Manage.All", "Sites.FullControl.All")
+    def test6_delete_site_content_type(self):
         ct_to_del = self.__class__.target_ct
         ct_to_del.delete_object().execute_query()
 
-    def test6_get_applicable_content_types_for_list(self):
+    @requires_delegated_permission(
+        "Sites.Read.All",
+        "Sites.FullControl.All",
+        "Sites.Manage.All",
+        "Sites.ReadWrite.All",
+    )
+    def test7_get_applicable_content_types_for_list(self):
         site = self.client.sites.root
         doc_lib = site.lists["Documents"].get().execute_query()
         cts = site.get_applicable_content_types_for_list(doc_lib.id).execute_query()

tests/onedrive/test_drive.py

@@ -1,34 +1,59 @@
+from tests.decorators import requires_delegated_permission
 from tests.graph_case import GraphTestCase
 
 
 class TestDrive(GraphTestCase):
     """OneDrive specific test case base class"""
 
-    def test1_get_drives(self):
+    @requires_delegated_permission(
+        "Files.Read",
+        "Files.Read.All",
+        "Files.ReadWrite",
+        "Files.ReadWrite.All",
+        "Sites.Read.All",
+        "Sites.ReadWrite.All",
+    )
+    def test1_list_drives(self):
         drives = self.client.drives.top(2).get().execute_query()
         self.assertLessEqual(len(drives), 2)
         for drive in drives:
             self.assertIsNotNone(drive.web_url)
 
-    def test2_get_drives_alt(self):
+    @requires_delegated_permission(
+        "Files.Read",
+        "Files.Read.All",
+        "Files.ReadWrite",
+        "Files.ReadWrite.All",
+        "Sites.Read.All",
+        "Sites.ReadWrite.All",
+    )
+    def test2_list_drives_alt(self):
         resp = self.client.execute_request_direct("drives?$top=2")
         drives = resp.json()["value"]
         self.assertLessEqual(len(drives), 2)
         for drive in drives:
             self.assertIsNotNone(drive["webUrl"])
 
-    def test3_get_first_drive(self):
-        drives = self.client.drives.get().top(10).execute_query()
-        self.assertGreaterEqual(len(drives), 1)
-        target_drive_id = drives[0].id
-
-        target_drive = self.client.drives[target_drive_id].get().execute_query()
-        self.assertEqual(target_drive.id, target_drive_id)
-
-    def test4_get_site(self):
-        site = self.client.sites.root.get().execute_query()
-        self.assertIsNotNone(site.resource_path)
+    @requires_delegated_permission(
+        "Files.Read",
+        "Files.Read.All",
+        "Files.ReadWrite",
+        "Files.ReadWrite.All",
+        "Sites.Read.All",
+        "Sites.ReadWrite.All",
+    )
+    def test4_get_drive_root(self):
+        result = self.client.me.drive.root.get().execute_query()
+        self.assertIsNotNone(result.resource_path)
 
+    @requires_delegated_permission(
+        "Files.Read",
+        "Files.Read.All",
+        "Files.ReadWrite",
+        "Files.ReadWrite.All",
+        "Sites.Read.All",
+        "Sites.ReadWrite.All",
+    )
     def test5_get_recent(self):
         items = self.client.me.drive.recent().execute_query()
         self.assertIsNotNone(items.resource_path)

tests/onedrive/test_sitepage.py

@@ -1,5 +1,6 @@
 from office365.onedrive.sitepages.site_page import SitePage
 from tests import create_unique_name, test_team_site_url
+from tests.decorators import requires_delegated_permission
 from tests.graph_case import GraphTestCase
 
 
@@ -43,6 +44,7 @@ def test4_get_site_page_pub_state(self):
     #    result = page.publish().execute_query()
     #    self.assertIsNotNone(result.resource_path)
 
+    @requires_delegated_permission("Sites.Read.All", "Sites.ReadWrite.All")
     def test6_list_site_pages(self):
         result = self.test_site.pages.top(10).get().execute_query()
         self.assertIsNotNone(result.resource_path)

tests/onedrive/test_term_store.py renamed to tests/onedrive/test_taxonomy.py

@@ -27,10 +27,10 @@ def setUpClass(cls):
     def tearDownClass(cls):
         pass
 
-    def test1_get_groups(self):
-        groups = self.target_store.groups.top(1).get().execute_query()
-        self.assertLessEqual(len(groups), 1)
-        for group in groups:
+    def test1_list_groups(self):
+        result = self.target_store.groups.top(1).get().execute_query()
+        self.assertLessEqual(len(result), 1)
+        for group in result:
             self.assertIsNotNone(group.resource_path)
 
     def test2_create_group(self):

tests/outlook/test_message_rules.py

@@ -1,26 +1,47 @@
 from office365.outlook.mail.messages.rules.actions import MessageRuleActions
 from office365.outlook.mail.messages.rules.rule import MessageRule
 from office365.outlook.mail.recipient import Recipient
+from tests.decorators import requires_delegated_permission
 from tests.graph_case import GraphTestCase
 
 
 class TestMessageRules(GraphTestCase):
     target_message_rule = None  # type: MessageRule
 
+    @requires_delegated_permission("MailboxSettings.ReadWrite")
     def test1_create_rule(self):
         actions = MessageRuleActions(
             forward_to=[Recipient.from_email("[email protected]")],
             stop_processing_rules=True,
+            mark_importance="normal",
         )
-        message_rules = (
+        result  = (
             self.client.me.mail_folders["inbox"]
             .message_rules.add("From partner", 2, actions)
             .execute_query()
         )
-        self.assertIsNotNone(message_rules.resource_path)
+        self.assertIsNotNone(result.resource_path)
+        self.__class__.target_message_rule = result
 
+    @requires_delegated_permission("MailboxSettings.Read", "MailboxSettings.ReadWrite")
     def test2_list_rules(self):
         message_rules = (
             self.client.me.mail_folders["inbox"].message_rules.get().execute_query()
         )
         self.assertIsNotNone(message_rules.resource_path)
+
+    @requires_delegated_permission("MailboxSettings.ReadWrite")
+    def test3_update_rule(self):
+        rule = self.__class__.target_message_rule
+        rule.actions.markImportance = "high"
+        rule.update().execute_query()
+
+    @requires_delegated_permission("MailboxSettings.Read", "MailboxSettings.ReadWrite")
+    def test4_get_rule(self):
+        result = self.__class__.target_message_rule.get().execute_query()
+        self.assertEqual(result.actions.markImportance, "high")
+
+    @requires_delegated_permission("MailboxSettings.ReadWrite")
+    def test5_delete_rule(self):
+        rule =self.__class__.target_message_rule
+        rule.delete_object().execute_query()