tests: added new tests

cognitedata · Mar 6, 2025 · 8ba4c98 · 8ba4c98
1 parent dc76d44
commit 8ba4c98
Showing 1 changed file with 68 additions and 0 deletions.
diff --git a/tests/tests_unit/test_api/test_data_modeling/test_instances.py b/tests/tests_unit/test_api/test_data_modeling/test_instances.py
@@ -1,7 +1,14 @@
 from __future__ import annotations
 
+import json
+import math
+import re
+
 import pytest
+from responses import RequestsMock
 
+from cognite.client import CogniteClient
+from cognite.client.data_classes.aggregations import Count
 from cognite.client.data_classes.data_modeling.ids import ViewId
 from cognite.client.data_classes.data_modeling.query import SourceSelector
 from tests.tests_unit.test_api.test_data_modeling.conftest import make_test_view
@@ -36,3 +43,64 @@ def test_instances_api_dump_instance_source(self, sources, expected):
         # ViewIdentifier = Union[ViewId, Tuple[str, str], Tuple[str, str, str]]
         # ViewIdentifier | Sequence[ViewIdentifier] | View | Sequence[View]
         assert expected == [source.dump() for source in SourceSelector._load_list(sources)]
+
+
+class TestAggregate:
+    @pytest.mark.usefixtures("disable_gzip")
+    @pytest.mark.parametrize("limit", [None, -1, math.inf])
+    def test_aggregate_maximum(
+        self, limit: int | float | None, rsps: RequestsMock, cognite_client: CogniteClient
+    ) -> None:
+        url = re.compile(r".*/models/instances/aggregate$")
+        response = {
+            "items": [
+                {
+                    "instanceType": "node",
+                    "group": {"site": "MyLocation"},
+                    "aggregates": [
+                        {
+                            "aggregate": "count",
+                            "property": "site",
+                            "value": 42,
+                        }
+                    ],
+                },
+            ]
+        }
+        rsps.add(rsps.POST, url, status=200, json=response)
+
+        _ = cognite_client.data_modeling.instances.aggregate(
+            ViewId("my_space", "MyView", "v1"), Count("externalId"), group_by="site", limit=limit
+        )
+        assert len(rsps.calls) == 1
+        call = rsps.calls[0]
+        body = json.loads(call.request.body)
+        assert "limit" in body
+        assert body["limit"] == cognite_client.data_modeling.instances._AGGREGATE_LIMIT
+
+
+class TestSearch:
+    @pytest.mark.usefixtures("disable_gzip")
+    @pytest.mark.parametrize("limit", [None, -1, math.inf])
+    def test_search_maximum(self, limit: int | float | None, rsps: RequestsMock, cognite_client: CogniteClient) -> None:
+        url = re.compile(r".*/models/instances/search$")
+        response = {
+            "items": [
+                {
+                    "instanceType": "node",
+                    "version": 1,
+                    "space": "my_instance_space",
+                    "externalId": "my_instance_id",
+                    "createdTime": 0,
+                    "lastUpdatedTime": 0,
+                },
+            ]
+        }
+        rsps.add(rsps.POST, url, status=200, json=response)
+
+        _ = cognite_client.data_modeling.instances.search(ViewId("my_space", "MyView", "v1"), "dummy text", limit=limit)
+        assert len(rsps.calls) == 1
+        call = rsps.calls[0]
+        body = json.loads(call.request.body)
+        assert "limit" in body
+        assert body["limit"] == cognite_client.data_modeling.instances._SEARCH_LIMIT