Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,11 @@
import copy
import functools
from pathlib import Path
from typing import Literal, Union, cast
from typing import Any, Literal, Union, cast

import gooddata_api_client.models as afm_models
from gooddata_api_client.model.elements_request import ElementsRequest
from gooddata_api_client.model.set_certification_request import SetCertificationRequest

from gooddata_sdk.catalog.catalog_service_base import CatalogServiceBase
from gooddata_sdk.catalog.data_source.validation.data_source import DataSourceValidator
Expand Down Expand Up @@ -249,6 +250,43 @@ def get_dependent_entities_graph_from_entry_points(
)
)

def set_certification(
self,
workspace_id: str,
entity_id: str,
entity_type: str,
certification: str | None = None,
certification_message: str | None = None,
) -> None:
"""Set certification status for an analytics entity (metric, visualization object, or analytical dashboard).

Args:
workspace_id (str):
Workspace identification string e.g. "demo"
entity_id (str):
ID of the entity to certify.
entity_type (str):
Type of the entity. One of: "metric", "visualizationObject", "analyticalDashboard"
certification (str | None):
Certification status. Use "CERTIFIED" to certify, or None to remove certification.
certification_message (str | None):
Optional message to attach to the certification.

Returns:
None
"""
kwargs: dict[str, Any] = {}
if certification is not None:
kwargs["status"] = certification
if certification_message is not None:
kwargs["message"] = certification_message
request = SetCertificationRequest(id=entity_id, type=entity_type, _check_type=False, **kwargs)
self._actions_api.set_certification(
workspace_id=workspace_id,
set_certification_request=request,
_check_return_type=False,
)

# Declarative methods for logical data model

def get_declarative_ldm(self, workspace_id: str) -> CatalogDeclarativeModel:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -264,6 +264,10 @@ class CatalogDeclarativeAnalyticalDashboard(CatalogAnalyticsBase):
]
| None
) = None
certification: str | None = None
certification_message: str | None = None
certified_at: str | None = None
certified_by: str | None = None

@staticmethod
def client_class() -> type[DeclarativeAnalyticalDashboard]:
Expand Down Expand Up @@ -314,13 +318,23 @@ def client_class() -> type[DeclarativeFilterContext]:

@define(kw_only=True)
class CatalogDeclarativeMetric(CatalogAnalyticsBase):
certification: str | None = None
certification_message: str | None = None
certified_at: str | None = None
certified_by: str | None = None

@staticmethod
def client_class() -> type[DeclarativeMetric]:
return DeclarativeMetric


@define(kw_only=True)
class CatalogDeclarativeVisualizationObject(CatalogAnalyticsBase):
certification: str | None = None
certification_message: str | None = None
certified_at: str | None = None
certified_by: str | None = None

@staticmethod
def client_class() -> type[DeclarativeVisualizationObject]:
return DeclarativeVisualizationObject
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
@define(kw_only=True)
class CatalogDependentEntitiesRequest(Base):
identifiers: list[CatalogEntityIdentifier] = field(factory=list)
relation: str | None = None

@staticmethod
def client_class() -> builtins.type[DependentEntitiesRequest]:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -362,6 +362,49 @@ def test_get_dependent_entities_graph_from_entry_points(test_config):
assert len(response.graph.nodes) == 3


def test_dependent_entities_request_relation_field():
"""Unit test verifying that CatalogDependentEntitiesRequest serializes the relation field correctly."""
request_with_relation = CatalogDependentEntitiesRequest(
identifiers=[CatalogEntityIdentifier(id="my_metric", type="metric")],
relation="DEPENDENCIES",
)
assert request_with_relation.relation == "DEPENDENCIES"
api_dict = request_with_relation.to_api().to_dict()
assert api_dict.get("relation") == "DEPENDENCIES"

request_default = CatalogDependentEntitiesRequest(
identifiers=[CatalogEntityIdentifier(id="my_metric", type="metric")],
)
assert request_default.relation is None
api_dict_default = request_default.to_api().to_dict()
assert "relation" not in api_dict_default


def test_set_certification_calls_api():
"""Unit test verifying that set_certification forwards parameters to the actions API."""
from gooddata_sdk.catalog.workspace.content_service import CatalogWorkspaceContentService

mock_actions_api = MagicMock()
service = MagicMock(spec=CatalogWorkspaceContentService)
service._actions_api = mock_actions_api
CatalogWorkspaceContentService.set_certification(
service,
workspace_id="demo",
entity_id="my_metric",
entity_type="metric",
certification="CERTIFIED",
certification_message="Approved",
)
mock_actions_api.set_certification.assert_called_once()
call_kwargs = mock_actions_api.set_certification.call_args
assert call_kwargs.kwargs["workspace_id"] == "demo"
request = call_kwargs.kwargs["set_certification_request"]
assert request["id"] == "my_metric"
assert request["type"] == "metric"
assert request["status"] == "CERTIFIED"
assert request["message"] == "Approved"


@gd_vcr.use_cassette(str(_fixtures_dir / "ldm_store_load.yaml"))
def test_ldm_store_load(test_config):
sdk = GoodDataSdk.create(host_=test_config["host"], token_=test_config["token"])
Expand Down
Loading