update_tests (#316)

* update_tests

[DOG-2933]

* unit

tests/conftest.py

@@ -0,0 +1,75 @@
+import os
+from dataclasses import dataclass
+from enum import Enum
+from typing import List, Optional
+
+import pytest
+
+from cognite.client import CogniteClient
+from cognite.client.config import ClientConfig
+from cognite.client.credentials import OAuthClientCredentials
+from cognite.client.exceptions import CogniteAPIError, CogniteNotFoundError
+
+
+class ETestType(Enum):
+    TIME_SERIES = "time_series"
+    FILES = "files"
+    RAW = "raw"
+    ASSETS = "assets"
+    EVENTS = "events"
+
+
+@dataclass
+class ParamTest:
+    test_type: ETestType
+    external_ids: Optional[List[str]] = None
+    database_name: Optional[str] = None
+    table_name: Optional[str] = None
+
+
+@pytest.fixture
+def set_upload_test(set_test_parameters: ParamTest, set_client: CogniteClient):
+    client = set_client
+    test_parameter = set_test_parameters
+    clean_test(client, test_parameter)
+    yield client, test_parameter
+    clean_test(client, test_parameter)
+
+
+@pytest.fixture
+def set_client() -> CogniteClient:
+    cognite_project = os.environ["COGNITE_PROJECT"]
+    cognite_base_url = os.environ["COGNITE_BASE_URL"]
+    cognite_token_url = os.environ["COGNITE_TOKEN_URL"]
+    cognite_client_id = os.environ["COGNITE_CLIENT_ID"]
+    cognite_client_secret = os.environ["COGNITE_CLIENT_SECRET"]
+    cognite_project_scopes = os.environ["COGNITE_TOKEN_SCOPES"].split(",")
+    client_config = ClientConfig(
+        project=cognite_project,
+        base_url=cognite_base_url,
+        credentials=OAuthClientCredentials(
+            cognite_token_url, cognite_client_id, cognite_client_secret, cognite_project_scopes
+        ),
+        client_name="extractor-utils-integration-tests",
+    )
+    return CogniteClient(client_config)
+
+
+def clean_test(client: CogniteClient, test_parameter: ParamTest):
+    if test_parameter.test_type == ETestType.TIME_SERIES:
+        client.time_series.delete(external_id=test_parameter.external_ids, ignore_unknown_ids=True)
+    elif test_parameter.test_type == ETestType.EVENTS:
+        client.events.delete(external_id=test_parameter.external_ids, ignore_unknown_ids=True)
+    elif test_parameter.test_type == ETestType.ASSETS:
+        client.assets.delete(external_id=test_parameter.external_ids, ignore_unknown_ids=True)
+    elif test_parameter.test_type == ETestType.RAW:
+        try:
+            client.raw.tables.delete(test_parameter.database_name, test_parameter.table_name)
+        except CogniteAPIError:
+            pass
+    elif test_parameter.test_type == ETestType.FILES:
+        for file in test_parameter.external_ids:
+            try:
+                client.files.delete(external_id=file)
+            except CogniteNotFoundError:
+                pass

tests/tests_integration/test_assets_integration.py

@@ -0,0 +1,66 @@
+#  Copyright 2020 Cognite AS
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import os
+import random
+from typing import Tuple
+
+import pytest
+
+from cognite.client import CogniteClient
+from cognite.client.data_classes.assets import Asset
+from cognite.extractorutils.uploader.assets import AssetUploadQueue
+from tests.conftest import ETestType, ParamTest
+
+
+@pytest.fixture
+def set_test_parameters() -> ParamTest:
+    test_id = random.randint(0, 2**31)
+    test_parameter = ParamTest(test_type=ETestType.ASSETS)
+    test_parameter.external_ids = [
+        f"util_integration_asset_test_1-{test_id}",
+        f"util_integration_asset_test_2-{test_id}",
+        f"util_integration_asset_test_3-{test_id}",
+    ]
+    return test_parameter
+
+
+@pytest.mark.parametrize("functions_runtime", ["true", "false"])
+def test_assets_upload_queue_upsert(set_upload_test: Tuple[CogniteClient, ParamTest], functions_runtime: str):
+    os.environ["COGNITE_FUNCTION_RUNTIME"] = functions_runtime
+    client, test_parameter = set_upload_test
+    queue = AssetUploadQueue(cdf_client=client)
+
+    # Upload a pair of events
+    queue.add_to_upload_queue(Asset(external_id=test_parameter.external_ids[0], description="desc", name="name"))
+    queue.add_to_upload_queue(Asset(external_id=test_parameter.external_ids[1], description="desc", name="name"))
+
+    queue.upload()
+
+    # This should result in an update and a create
+    queue.add_to_upload_queue(
+        Asset(external_id=test_parameter.external_ids[1], description="new desc", name="new name")
+    )
+    queue.add_to_upload_queue(
+        Asset(external_id=test_parameter.external_ids[2], description="new desc", name="new name")
+    )
+
+    queue.upload()
+
+    retrieved = client.assets.retrieve_multiple(external_ids=test_parameter.external_ids)
+    assert retrieved[0].description == "desc"
+    assert retrieved[1].description == "new desc"
+    assert retrieved[2].description == "new desc"
+    assert retrieved[1].name == "new name"
+    assert retrieved[2].name == "new name"