feat: adds condition class and assoc. unit tests

Author: chalmerlowe

google/cloud/bigquery/dataset.py

@@ -19,6 +19,7 @@
 import copy
 
 import typing
+from typing import Optional, List, Dict, Any, Union
 
 import google.cloud._helpers  # type: ignore
 
@@ -29,8 +30,6 @@
 from google.cloud.bigquery.encryption_configuration import EncryptionConfiguration
 from google.cloud.bigquery import external_config
 
-from typing import Optional, List, Dict, Any, Union
-
 
 def _get_table_reference(self, table_id: str) -> TableReference:
     """Constructs a TableReference.
@@ -1074,3 +1073,93 @@ def reference(self):
     model = _get_model_reference
 
     routine = _get_routine_reference
+
+
+class Condition(object):
+    """Represents a textual expression in the Common Expression Language (CEL) syntax.
+
+    Typically used for filtering or policy rules, such as in IAM Conditions
+    or BigQuery row/column access policies.
+
+    See:
+        https://cloud.google.com/iam/docs/reference/rest/Shared.Types/Expr
+        https://github.com/google/cel-spec
+
+    Args:
+        expression (str):
+            The condition expression string using CEL syntax. This is required.
+            Example: ``resource.type == "compute.googleapis.com/Instance"``
+        title (Optional[str]):
+            An optional title for the condition, providing a short summary.
+            Example: ``"Request is for a GCE instance"``
+        description (Optional[str]):
+            An optional description of the condition, providing a detailed explanation.
+            Example: ``"This condition checks whether the resource is a GCE instance."``
+    """
+
+    def __init__(
+        self,
+        expression: str,
+        title: Optional[str] = None,
+        description: Optional[str] = None,
+    ):
+        self._properties: Dict[str, Any] = {}
+        # Use setters to initialize properties, which also handle validation
+        self.expression = expression
+        self.title = title
+        self.description = description
+
+    @property
+    def title(self) -> Optional[str]:
+        """Optional[str]: The title for the condition."""
+        return self._properties.get("title")
+
+    @title.setter
+    def title(self, value: Optional[str]):
+        if value is not None and not isinstance(value, str):
+            raise ValueError("Pass a string for title, or None")
+        self._properties["title"] = value
+
+    @property
+    def description(self) -> Optional[str]:
+        """Optional[str]: The description for the condition."""
+        return self._properties.get("description")
+
+    @description.setter
+    def description(self, value: Optional[str]):
+        if value is not None and not isinstance(value, str):
+            raise ValueError("Pass a string for description, or None")
+        self._properties["description"] = value
+
+    @property
+    def expression(self) -> str:
+        """str: The expression string for the condition."""
+
+        # Cast assumes expression is always set due to __init__ validation
+        return typing.cast(str, self._properties.get("expression"))
+
+    @expression.setter
+    def expression(self, value: str):
+        if not isinstance(value, str):
+            raise ValueError("Pass a non-empty string for expression")
+        if not value:
+            raise ValueError("expression cannot be an empty string")
+        self._properties["expression"] = value
+
+    def to_api_repr(self) -> Dict[str, Any]:
+        """Construct the API resource representation of this Condition."""
+        return self._properties
+
+    @classmethod
+    def from_api_repr(cls, resource: Dict[str, Any]) -> "Condition":
+        """Factory: construct a Condition instance given its API representation."""
+
+        # Ensure required fields are present in the resource if necessary
+        if "expression" not in resource:
+            raise ValueError("API representation missing required 'expression' field.")
+
+        return cls(
+            expression=resource["expression"],
+            title=resource.get("title"),
+            description=resource.get("description"),
+        )

tests/unit/test_dataset.py

@@ -19,6 +19,7 @@
 import pytest
 from google.cloud.bigquery.dataset import (
     AccessEntry,
+    Condition,
     Dataset,
     DatasetReference,
     Table,
@@ -1228,3 +1229,150 @@ def test_table(self):
         self.assertEqual(table.table_id, "table_id")
         self.assertEqual(table.dataset_id, dataset_id)
         self.assertEqual(table.project, project)
+
+
+class TestCondition:
+    EXPRESSION = 'resource.name.startsWith("projects/my-project/instances/")'
+    TITLE = "Instance Access"
+    DESCRIPTION = "Access to instances in my-project"
+
+    @pytest.fixture
+    def condition_instance(self):
+        """Provides a Condition instance for tests."""
+        return Condition(
+            expression=self.EXPRESSION,
+            title=self.TITLE,
+            description=self.DESCRIPTION,
+        )
+
+    @pytest.fixture
+    def condition_api_repr(self):
+        """Provides the API representation for the test Condition."""
+        return {
+            "expression": self.EXPRESSION,
+            "title": self.TITLE,
+            "description": self.DESCRIPTION,
+        }
+
+    # --- Basic Functionality Tests ---
+
+    def test_constructor_and_getters_full(self, condition_instance):
+        """Test initialization with all arguments and subsequent attribute access."""
+        assert condition_instance.expression == self.EXPRESSION
+        assert condition_instance.title == self.TITLE
+        assert condition_instance.description == self.DESCRIPTION
+
+    def test_constructor_and_getters_minimal(self):
+        """Test initialization with only the required expression."""
+        condition = Condition(expression=self.EXPRESSION)
+        assert condition.expression == self.EXPRESSION
+        assert condition.title is None
+        assert condition.description is None
+
+    def test_setters(self, condition_instance):
+        """Test setting attributes after initialization."""
+        new_title = "New Title"
+        new_desc = "New Description"
+        new_expr = "request.time < timestamp('2024-01-01T00:00:00Z')"
+
+        condition_instance.title = new_title
+        assert condition_instance.title == new_title
+
+        condition_instance.description = new_desc
+        assert condition_instance.description == new_desc
+
+        condition_instance.expression = new_expr
+        assert condition_instance.expression == new_expr
+
+        # Test setting optional fields back to None
+        condition_instance.title = None
+        assert condition_instance.title is None
+        condition_instance.description = None
+        assert condition_instance.description is None
+
+    # --- API Representation Tests ---
+
+    def test_to_api_repr_full(self, condition_instance, condition_api_repr):
+        """Test converting a fully populated Condition to API representation."""
+        api_repr = condition_instance.to_api_repr()
+        assert api_repr == condition_api_repr
+
+    def test_to_api_repr_minimal(self):
+        """Test converting a minimally populated Condition to API representation."""
+        condition = Condition(expression=self.EXPRESSION)
+        expected_api_repr = {
+            "expression": self.EXPRESSION,
+            "title": None,
+            "description": None,
+        }
+        api_repr = condition.to_api_repr()
+        assert api_repr == expected_api_repr
+
+    def test_from_api_repr_full(self, condition_api_repr):
+        """Test creating a Condition from a full API representation."""
+        condition = Condition.from_api_repr(condition_api_repr)
+        assert condition.expression == self.EXPRESSION
+        assert condition.title == self.TITLE
+        assert condition.description == self.DESCRIPTION
+
+    def test_from_api_repr_minimal(self):
+        """Test creating a Condition from a minimal API representation."""
+        minimal_repr = {"expression": self.EXPRESSION}
+        condition = Condition.from_api_repr(minimal_repr)
+        assert condition.expression == self.EXPRESSION
+        assert condition.title is None
+        assert condition.description is None
+
+    def test_from_api_repr_with_extra_fields(self):
+        """Test creating a Condition from an API repr with unexpected fields."""
+        api_repr = {
+            "expression": self.EXPRESSION,
+            "title": self.TITLE,
+            "unexpected_field": "some_value",
+        }
+        condition = Condition.from_api_repr(api_repr)
+        assert condition.expression == self.EXPRESSION
+        assert condition.title == self.TITLE
+        assert condition.description is None
+        # Check that the extra field didn't get added to internal properties
+        assert "unexpected_field" not in condition._properties
+
+    #     # --- Validation Tests ---
+
+    @pytest.mark.parametrize(
+        "kwargs, error_msg",
+        [
+            ({"expression": None}, "Pass a non-empty string for expression"),  # type: ignore
+            ({"expression": ""}, "expression cannot be an empty string"),
+            ({"expression": 123}, "Pass a non-empty string for expression"),  # type: ignore
+            ({"expression": EXPRESSION, "title": 123}, "Pass a string for title, or None"),  # type: ignore
+            ({"expression": EXPRESSION, "description": False}, "Pass a string for description, or None"),  # type: ignore
+        ],
+    )
+    def test_validation_init(self, kwargs, error_msg):
+        """Test validation during __init__."""
+        with pytest.raises(ValueError, match=error_msg):
+            Condition(**kwargs)
+
+    @pytest.mark.parametrize(
+        "attribute, value, error_msg",
+        [
+            ("expression", None, "Pass a non-empty string for expression"),  # type: ignore
+            ("expression", "", "expression cannot be an empty string"),
+            ("expression", 123, "Pass a non-empty string for expression"),  # type: ignore
+            ("title", 123, "Pass a string for title, or None"),  # type: ignore
+            ("description", [], "Pass a string for description, or None"),  # type: ignore
+        ],
+    )
+    def test_validation_setters(self, condition_instance, attribute, value, error_msg):
+        """Test validation via setters."""
+        with pytest.raises(ValueError, match=error_msg):
+            setattr(condition_instance, attribute, value)
+
+    def test_validation_expression_required_from_api(self):
+        """Test ValueError is raised if expression is missing in from_api_repr."""
+        api_repr = {"title": self.TITLE}
+        with pytest.raises(
+            ValueError, match="API representation missing required 'expression' field."
+        ):
+            Condition.from_api_repr(api_repr)