test: cover new column get/update + typed-read paths (sync + async) for diff coverage

Author: arorashivam96

tests/unit/aio/data/test_async_odata_internal.py

@@ -1893,3 +1893,162 @@ async def test_none_context_no_parentheses(self):
         client = _AsyncODataClient(auth, "https://example.crm.dynamics.com", config=config)
         headers = await client._headers()
         assert "(" not in headers["User-Agent"]
+
+
+class TestAsyncRetrieveMetadataChanges:
+    """_AsyncODataClient._retrieve_metadata_changes / _get_column typed (#203)."""
+
+    def _client_with_attrs(self, entities):
+        client = _make_client()
+        client._request_metadata_with_retry = AsyncMock(return_value=_resp(json_data={"EntityMetadata": entities}))
+        return client
+
+    async def test_typed_read_single_request_with_projection_and_criteria(self):
+        client = self._client_with_attrs(
+            [
+                {
+                    "LogicalName": "account",
+                    "Attributes": [
+                        {
+                            "@odata.type": "#Microsoft.Dynamics.CRM.StringAttributeMetadata",
+                            "LogicalName": "emailaddress1",
+                            "MaxLength": 100,
+                        }
+                    ],
+                }
+            ]
+        )
+        attrs = await client._retrieve_metadata_changes("account", attribute_logical_name="emailaddress1")
+        assert len(attrs) == 1
+        assert attrs[0]["MaxLength"] == 100
+        call = client._request_metadata_with_retry.call_args
+        assert call.args[0] == "get"
+        assert "RetrieveMetadataChanges(Query=@p1)" in call.args[1]
+        query = json.loads(call.kwargs["params"]["@p1"])
+        assert query["Criteria"]["Conditions"][0]["Value"]["Value"] == "account"
+        assert query["AttributeQuery"]["Criteria"]["Conditions"][0]["Value"]["Value"] == "emailaddress1"
+        assert "MaxLength" in query["AttributeQuery"]["Properties"]["PropertyNames"]
+
+    async def test_table_not_found_raises(self):
+        client = self._client_with_attrs([])
+        with pytest.raises(MetadataError):
+            await client._retrieve_metadata_changes("nope")
+
+    async def test_get_column_typed_returns_single_attr(self):
+        client = _make_client()
+        client._retrieve_metadata_changes = AsyncMock(return_value=[{"LogicalName": "emailaddress1", "MaxLength": 100}])
+        col = await client._get_column("account", "emailaddress1", typed=True)
+        assert col["MaxLength"] == 100
+
+    async def test_get_column_typed_empty_returns_none(self):
+        client = _make_client()
+        client._retrieve_metadata_changes = AsyncMock(return_value=[])
+        assert await client._get_column("account", "ghost", typed=True) is None
+
+
+class TestAsyncGetColumn:
+    """_AsyncODataClient._get_column non-typed paths (#203)."""
+
+    def _client(self):
+        client = _make_client()
+        client._get_entity_by_table_schema_name = AsyncMock(return_value={"MetadataId": "ent-1"})
+        client._request = AsyncMock(return_value=_resp(json_data={"LogicalName": "emailaddress1", "MaxLength": 100}))
+        return client
+
+    async def test_untyped_returns_json(self):
+        client = self._client()
+        col = await client._get_column("account", "emailaddress1")
+        assert col["MaxLength"] == 100
+        assert "Attributes(LogicalName='emailaddress1')" in client._request.call_args.args[1]
+
+    async def test_untyped_with_select_projects(self):
+        client = self._client()
+        await client._get_column("account", "emailaddress1", select=["MaxLength"])
+        assert client._request.call_args.kwargs["params"]["$select"] == "MaxLength"
+
+    async def test_untyped_404_returns_none(self):
+        client = self._client()
+        client._request = AsyncMock(side_effect=HttpError("nope", 404))
+        assert await client._get_column("account", "ghost") is None
+
+    async def test_untyped_other_error_propagates(self):
+        client = self._client()
+        client._request = AsyncMock(side_effect=HttpError("boom", 500))
+        with pytest.raises(HttpError):
+            await client._get_column("account", "emailaddress1")
+
+    async def test_untyped_table_not_found_raises(self):
+        client = self._client()
+        client._get_entity_by_table_schema_name = AsyncMock(return_value=None)
+        with pytest.raises(MetadataError):
+            await client._get_column("nope", "emailaddress1")
+
+
+class TestAsyncUpdateAttribute:
+    """_AsyncODataClient._update_attribute (#202)."""
+
+    def _client(self):
+        client = _make_client()
+        client._get_entity_by_table_schema_name = AsyncMock(return_value={"MetadataId": "ent-1"})
+        client._request = AsyncMock(
+            return_value=_resp(
+                json_data={
+                    "@odata.type": "#Microsoft.Dynamics.CRM.MemoAttributeMetadata",
+                    "MetadataId": "attr-1",
+                    "SchemaName": "new_Comment",
+                }
+            )
+        )
+        client._execute_raw = AsyncMock()
+        return client
+
+    async def test_update_issues_put_with_odata_type_and_override(self):
+        client = self._client()
+        await client._update_attribute(
+            "new_Feedback", "new_Comment", {"max_length": 4000, "display_name": "Customer Comment"}
+        )
+        client._execute_raw.assert_awaited_once()
+        req = client._execute_raw.call_args.args[0]
+        assert req.method == "PUT"
+        assert "Attributes(attr-1)" in req.url
+        body = json.loads(req.body)
+        assert body["@odata.type"] == "Microsoft.Dynamics.CRM.MemoAttributeMetadata"
+        assert body["MetadataId"] == "attr-1"
+        assert body["MaxLength"] == 4000
+        assert body["DisplayName"]["LocalizedLabels"][0]["Label"] == "Customer Comment"
+
+    async def test_update_unknown_key_raises(self):
+        client = self._client()
+        with pytest.raises(ValueError):
+            await client._update_attribute("new_Feedback", "new_Comment", {"bogus": 1})
+
+    async def test_update_empty_overrides_raises(self):
+        client = self._client()
+        with pytest.raises(TypeError):
+            await client._update_attribute("new_Feedback", "new_Comment", {})
+
+    async def test_update_table_not_found_raises(self):
+        client = self._client()
+        client._get_entity_by_table_schema_name = AsyncMock(return_value=None)
+        with pytest.raises(MetadataError):
+            await client._update_attribute("nope", "new_Comment", {"max_length": 10})
+
+    async def test_update_column_get_404_raises_metadata_error(self):
+        client = self._client()
+        client._request = AsyncMock(side_effect=HttpError("not found", 404))
+        with pytest.raises(MetadataError):
+            await client._update_attribute("new_Feedback", "ghost", {"max_length": 10})
+        client._execute_raw.assert_not_awaited()
+
+    async def test_update_column_get_500_propagates(self):
+        client = self._client()
+        client._request = AsyncMock(side_effect=HttpError("boom", 500))
+        with pytest.raises(HttpError):
+            await client._update_attribute("new_Feedback", "new_Comment", {"max_length": 10})
+
+    async def test_update_column_missing_odata_type_raises(self):
+        client = self._client()
+        client._request = AsyncMock(return_value=_resp(json_data={"SchemaName": "new_Comment"}))
+        with pytest.raises(MetadataError):
+            await client._update_attribute("new_Feedback", "new_Comment", {"max_length": 10})
+        client._execute_raw.assert_not_awaited()

tests/unit/aio/test_async_tables.py

@@ -312,3 +312,41 @@ async def test_list_table_relationships(self, async_client, mock_od):
         result = await async_client.tables.list_table_relationships("account")
         mock_od._list_table_relationships.assert_called_once_with("account", filter=None, select=None)
         assert result == [{"SchemaName": "new_Dept_Emp"}]
+
+
+class TestAsyncTableGetUpdateColumn:
+    async def test_get_column(self, async_client, mock_od):
+        """get_column() delegates to _get_column with defaults and returns the dict."""
+        mock_od._get_column.return_value = {"LogicalName": "emailaddress1", "MaxLength": 100}
+        result = await async_client.tables.get_column("account", "emailaddress1")
+        mock_od._get_column.assert_called_once_with("account", "emailaddress1", typed=False, select=None)
+        assert result["MaxLength"] == 100
+
+    async def test_get_column_typed_with_select(self, async_client, mock_od):
+        """get_column() forwards typed and select."""
+        mock_od._get_column.return_value = None
+        await async_client.tables.get_column("account", "emailaddress1", typed=True, select=["MaxLength"])
+        mock_od._get_column.assert_called_once_with("account", "emailaddress1", typed=True, select=["MaxLength"])
+
+    async def test_update_column(self, async_client, mock_od):
+        """update_column() delegates to _update_attribute and returns the column name."""
+        mock_od._update_attribute.return_value = "new_Comment"
+        result = await async_client.tables.update_column("new_Feedback", "new_Comment", {"max_length": 4000})
+        mock_od._update_attribute.assert_called_once_with("new_Feedback", "new_Comment", {"max_length": 4000})
+        assert result == "new_Comment"
+
+    async def test_update_columns(self, async_client, mock_od):
+        """update_columns() updates each column and returns the names in order."""
+        result = await async_client.tables.update_columns(
+            "new_Feedback",
+            {"new_Comment": {"max_length": 4000}, "new_Rating": {"max_value": 10}},
+        )
+        assert result == ["new_Comment", "new_Rating"]
+        assert mock_od._update_attribute.await_count == 2
+
+    async def test_update_columns_empty_raises(self, async_client, mock_od):
+        """update_columns() rejects an empty mapping before touching the client."""
+        import pytest
+
+        with pytest.raises(TypeError):
+            await async_client.tables.update_columns("new_Feedback", {})

tests/unit/data/test_odata_internal.py

@@ -3189,6 +3189,65 @@ def test_update_table_not_found_raises(self):
         with self.assertRaises(MetadataError):
             self.od._update_attribute("nope", "new_Comment", {"max_length": 10})
 
+    def test_update_column_get_404_raises_metadata_error(self):
+        """A 404 fetching the existing attribute -> MetadataError (column not found)."""
+        self.od._request = MagicMock(side_effect=HttpError("not found", 404))
+        with self.assertRaises(MetadataError):
+            self.od._update_attribute("new_Feedback", "ghost", {"max_length": 10})
+        self.od._execute_raw.assert_not_called()
+
+    def test_update_column_get_500_propagates(self):
+        """A non-404 error fetching the existing attribute propagates unchanged."""
+        self.od._request = MagicMock(side_effect=HttpError("boom", 500))
+        with self.assertRaises(HttpError):
+            self.od._update_attribute("new_Feedback", "new_Comment", {"max_length": 10})
+
+    def test_update_column_missing_odata_type_raises(self):
+        """Existing attribute lacking @odata.type/MetadataId -> MetadataError, no PUT."""
+        self.od._request = MagicMock(return_value=_mock_response(json_data={"SchemaName": "new_Comment"}))
+        with self.assertRaises(MetadataError):
+            self.od._update_attribute("new_Feedback", "new_Comment", {"max_length": 10})
+        self.od._execute_raw.assert_not_called()
+
+
+class TestGetColumn(unittest.TestCase):
+    """Unit tests for _ODataClient._get_column (typed and non-typed paths, #203)."""
+
+    def setUp(self):
+        self.od = _make_odata_client()
+        self.od._get_entity_by_table_schema_name = MagicMock(return_value={"MetadataId": "ent-1"})
+        self.od._request = MagicMock(
+            return_value=_mock_response(json_data={"LogicalName": "emailaddress1", "MaxLength": 100})
+        )
+
+    def test_untyped_returns_json(self):
+        col = self.od._get_column("account", "emailaddress1")
+        self.assertEqual(col["MaxLength"], 100)
+        url = self.od._request.call_args.args[1]
+        self.assertIn("Attributes(LogicalName='emailaddress1')", url)
+
+    def test_untyped_with_select_projects(self):
+        self.od._get_column("account", "emailaddress1", select=["MaxLength"])
+        self.assertEqual(self.od._request.call_args.kwargs["params"]["$select"], "MaxLength")
+
+    def test_untyped_404_returns_none(self):
+        self.od._request = MagicMock(side_effect=HttpError("nope", 404))
+        self.assertIsNone(self.od._get_column("account", "ghost"))
+
+    def test_untyped_other_error_propagates(self):
+        self.od._request = MagicMock(side_effect=HttpError("boom", 500))
+        with self.assertRaises(HttpError):
+            self.od._get_column("account", "emailaddress1")
+
+    def test_untyped_table_not_found_raises(self):
+        self.od._get_entity_by_table_schema_name = MagicMock(return_value=None)
+        with self.assertRaises(MetadataError):
+            self.od._get_column("nope", "emailaddress1")
+
+    def test_typed_empty_returns_none(self):
+        self.od._retrieve_metadata_changes = MagicMock(return_value=[])
+        self.assertIsNone(self.od._get_column("account", "ghost", typed=True))
+
 
 class TestRetrieveMetadataChanges(unittest.TestCase):
     """Unit tests for _ODataClient._retrieve_metadata_changes / _get_column typed (#203)."""

tests/unit/test_tables_operations.py

@@ -462,6 +462,54 @@ def test_list_columns_with_select_and_filter(self):
             typed=False,
         )
 
+    # ------------------------------------------------- get_column / update_column
+
+    def test_get_column(self):
+        """get_column() delegates to _get_column with defaults and returns the dict."""
+        expected = {"LogicalName": "emailaddress1", "MaxLength": 100}
+        self.client._odata._get_column.return_value = expected
+
+        result = self.client.tables.get_column("account", "emailaddress1")
+
+        self.client._odata._get_column.assert_called_once_with("account", "emailaddress1", typed=False, select=None)
+        self.assertEqual(result, expected)
+
+    def test_get_column_typed_with_select(self):
+        """get_column() forwards typed and select."""
+        self.client._odata._get_column.return_value = None
+
+        self.client.tables.get_column("account", "emailaddress1", typed=True, select=["MaxLength"])
+
+        self.client._odata._get_column.assert_called_once_with(
+            "account", "emailaddress1", typed=True, select=["MaxLength"]
+        )
+
+    def test_update_column(self):
+        """update_column() delegates to _update_attribute and returns the column name."""
+        self.client._odata._update_attribute.return_value = "new_Comment"
+
+        result = self.client.tables.update_column("new_Feedback", "new_Comment", {"max_length": 4000})
+
+        self.client._odata._update_attribute.assert_called_once_with(
+            "new_Feedback", "new_Comment", {"max_length": 4000}
+        )
+        self.assertEqual(result, "new_Comment")
+
+    def test_update_columns(self):
+        """update_columns() updates each column and returns the names in order."""
+        result = self.client.tables.update_columns(
+            "new_Feedback",
+            {"new_Comment": {"max_length": 4000}, "new_Rating": {"max_value": 10}},
+        )
+
+        self.assertEqual(result, ["new_Comment", "new_Rating"])
+        self.assertEqual(self.client._odata._update_attribute.call_count, 2)
+
+    def test_update_columns_empty_raises(self):
+        """update_columns() rejects an empty mapping before touching the client."""
+        with self.assertRaises(TypeError):
+            self.client.tables.update_columns("new_Feedback", {})
+
     # ------------------------------------------------- list_relationships
 
     def test_list_relationships(self):