Feature/enable and disable rules

TM1py/Objects/Cube.py

@@ -1,10 +1,10 @@
 # -*- coding: utf-8 -*-
-
+import base64
 import collections
 import json
 from typing import Iterable, List, Dict, Optional, Union
 
-from TM1py.Objects.Rules import Rules
+from TM1py.Objects.Rules import Rules, RULES_ENCODING_PREFIX, FEEDERS_ENCODING_PREFIX
 from TM1py.Objects.TM1Object import TM1Object
 from TM1py.Utils import format_url
 
@@ -115,3 +115,69 @@ def _construct_body(self) -> str:
         if self.has_rules:
             body_as_dict['Rules'] = str(self.rules)
         return json.dumps(body_as_dict, ensure_ascii=False)
+
+    def enable_rules(self):
+        if not self.rules.text:
+            # If there is no rule, there is nothing to do.
+            return
+
+        rules_statements = self.rules.text.splitlines()
+        if not len(rules_statements) == 1:
+            raise RuntimeError(
+                "The cube rules are not disabled correctly. "
+                "Must be 1 line of base64-encoded hash.")
+
+        encoded_rules_statement = rules_statements[0]
+        if not encoded_rules_statement.startswith(RULES_ENCODING_PREFIX):
+            raise RuntimeError(
+                f"The cube rules are not disabled correctly. "
+                f"Must start with prefix: '{RULES_ENCODING_PREFIX}'")
+
+        encoded_rule = encoded_rules_statement[len(RULES_ENCODING_PREFIX):]
+        self._rules = Rules(base64.b64decode(encoded_rule).decode('utf-8'))
+
+    def enable_feeders(self):
+        if not self.rules:
+            # If there is no rule, there is nothing to do.
+            return
+
+        rules_statements = self.rules.text.splitlines()
+
+        encoded_feeders_statement = rules_statements[-1]
+        if not encoded_feeders_statement.startswith(FEEDERS_ENCODING_PREFIX):
+            raise RuntimeError(
+                f"The cube feeders are not disabled correctly. "
+                f"First line in Feeders section must start with prefix: '{FEEDERS_ENCODING_PREFIX}'")
+
+        encoded_feeders = encoded_feeders_statement[len(FEEDERS_ENCODING_PREFIX):]
+        self._rules = Rules(
+            self.rules.text[:-len(encoded_feeders_statement)] + base64.b64decode(encoded_feeders).decode('utf-8'))
+
+    def disable_feeders(self):
+        if not self.rules:
+            # If there is no rule, there is nothing to do.
+            return
+
+        rule_statements = self.rules.text.splitlines()
+
+        # sanitize statements to simplify identification of 'feeders;' line
+        sanitized_rule_statements = [
+            rule.strip().lower()
+            for rule
+            in rule_statements]
+
+        feeders_start_index = sanitized_rule_statements.index('feeders;') + 1
+        feeders_statements = rule_statements[feeders_start_index:]
+        hashed_feeders = base64.b64encode('\n'.join(feeders_statements).encode('utf-8')).decode('utf-8')
+
+        rule_statements[feeders_start_index:] = [f"{FEEDERS_ENCODING_PREFIX}{hashed_feeders}"]
+        self._rules = Rules("\n".join(rule_statements))
+
+    def disable_rules(self):
+        if not self.rules:
+            # If there is no rule, there is nothing to do.
+            return
+
+        # Encode the entire rule
+        self._rules = Rules(
+            f"{RULES_ENCODING_PREFIX}{base64.b64encode(self.rules.text.encode('utf-8')).decode('utf-8')}")

TM1py/Objects/Rules.py

@@ -4,6 +4,10 @@
 from TM1py.Objects.TM1Object import TM1Object
 
 
+RULES_ENCODING_PREFIX = "# B64 ENCODED RULES="
+FEEDERS_ENCODING_PREFIX = "# B64 ENCODED FEEDERS="
+
+
 class Rules(TM1Object):
     """
         Abstraction of Rules on a cube.
@@ -13,7 +17,6 @@ class Rules(TM1Object):
             comments are not included.
 
     """
-    KEYWORDS = ['SKIPCHECK', 'FEEDSTRINGS', 'UNDEFVALS', 'FEEDERS']
 
     def __init__(self, rules: str):
         self._text = rules
@@ -89,3 +92,8 @@ def __iter__(self):
 
     def __str__(self):
         return self.text
+
+    def __bool__(self):
+        if len(self.text):
+            return True
+        return False

TM1py/Services/CubeService.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+import base64
 import json
 import random
 from typing import List, Iterable, Dict
@@ -13,6 +14,8 @@
 from TM1py.Utils import format_url, require_version, require_data_admin, case_and_space_insensitive_equals
 
 
+
+
 class CubeService(ObjectService):
     """ Service to handle Object Updates for TM1 Cubes
 
@@ -231,7 +234,7 @@ def search_for_dimension(self, dimension_name: str, skip_control_cubes: bool = F
 
     def search_for_dimension_substring(self, substring: str, skip_control_cubes: bool = False,
                                        **kwargs) -> Dict[str, List[str]]:
-        """ Ask TM1 Server for a dictinary of cube names with the dimension whose name contains the substring
+        """ Ask TM1 Server for a dictionary of cube names with the dimension whose name contains the substring
 
         :param substring: string to search for in dim name
         :param skip_control_cubes: bool, True will exclude control cubes from result
@@ -249,6 +252,44 @@ def search_for_dimension_substring(self, substring: str, skip_control_cubes: boo
         cube_dict = {entry['Name']: [dim['Name'] for dim in entry['Dimensions']] for entry in response.json()['value']}
         return cube_dict
 
+    def disable_rules(self, cube_name: str) -> None:
+        """
+        Disable the entire cube rule by substituting it with its base64-encoded hash
+
+        :param cube_name: name of the cube
+        """
+        cube = self.get(cube_name)
+        cube.disable_rules()
+        self.update(cube)
+
+    def disable_feeders(self, cube_name: str) -> None:
+        """
+        Disable the feeders by substituting it with its base64-encoded hash
+
+        :param cube_name: name of the cube
+        """
+        cube = self.get(cube_name)
+        cube.disable_feeders()
+        self.update(cube)
+
+    def enable_rules(self, cube_name: str) -> None:
+        """ Enable the disabled cube rules by decoding the base64-encoded hash
+
+        :param cube_name: name of the cube
+        """
+        cube = self.get(cube_name)
+        cube.enable_rules()
+        self.update(cube)
+
+    def enable_feeders(self, cube_name: str) -> None:
+        """ Enable the disabled cube rules by decoding the base64-encoded hash
+
+        :param cube_name: name of the cube
+        """
+        cube = self.get(cube_name)
+        cube.enable_feeders()
+        self.update(cube)
+
     def search_for_rule_substring(self, substring: str, skip_control_cubes: bool = False, case_insensitive=True,
                                   space_insensitive=True, **kwargs) -> List[Cube]:
         """ get all cubes from TM1 Server as TM1py.Cube instances where rules for given cube contain specified substring
@@ -405,4 +446,4 @@ def get_vmt(self, cube_name: str):
     def set_vmt(self, cube_name: str, vmt: int):
         url = format_url("/Cubes('{}')", cube_name)
         payload = {"ViewStorageMinTime": vmt}
-        response = self._rest.PATCH(url=url, data=json.dumps(payload))
+        response = self._rest.PATCH(url=url, data=json.dumps(payload))

Tests/CubeService_test.py

@@ -301,6 +301,32 @@ def test_get_measure_dimension(self):
 
         self.assertEqual(self.dimension_names[-1], measure_dimension)
 
+    def test_disable_rules_enable_rules(self):
+        original = Rules(
+            "#comment1\n"
+            "FEEDSTRINGS;\n"
+            "UNDEFVALS;\n"
+            "#comment2\n"
+            "SKIPCHECK;\n"
+            "#comment3\n"
+            "#comment4\n"
+            "[]=N:2;\n"
+            "#find_me_comment\n"
+            "FEEDERS;\n"
+            "#comment5\n"
+            "#comment6"
+        )
+        c = self.tm1.cubes.get(self.cube_name)
+        c.rules = original
+        self.tm1.cubes.update(c)
+
+        self.assertEqual(self.tm1.cubes.get(c.name).has_rules, True)
+
+        self.tm1.cubes.disable_rules(c.name)
+        self.tm1.cubes.enable_rules(c.name)
+
+        self.assertEqual(c.rules.text, original.text)
+
     def tearDown(self):
         self.tm1.cubes.delete(self.cube_name)
         if self.tm1.cubes.exists(self.cube_name_to_delete):