diff --git a/libs/labelbox/src/labelbox/client.py b/libs/labelbox/src/labelbox/client.py index 75755afcc..05e8d26f9 100644 --- a/libs/labelbox/src/labelbox/client.py +++ b/libs/labelbox/src/labelbox/client.py @@ -1861,7 +1861,8 @@ def is_feature_schema_archived( raise ResourceNotFoundError(Ontology, ontology_id) else: raise LabelboxError( - "Failed to get the feature schema archived status." + "Failed to get the feature schema archived status. " + f"The server answered {response.status_code}: {response.text}" ) def get_model_slice(self, slice_id) -> ModelSlice: diff --git a/libs/labelbox/tests/conftest.py b/libs/labelbox/tests/conftest.py index 94d756170..dc50f0513 100644 --- a/libs/labelbox/tests/conftest.py +++ b/libs/labelbox/tests/conftest.py @@ -161,6 +161,35 @@ def service_api_key() -> str: return service_api_key +# The API counts requests per user in a fixed 60 second window and every CI +# job uses the same API key, so a 429 is the other jobs' traffic and not a +# defect in the code under test. Reruns come too soon to outlast the window. +RATE_LIMIT_WINDOW_SECONDS = 60 +RATE_LIMIT_FALLBACK_WAIT_SECONDS = 20 +RATE_LIMIT_MAX_WAITS = 4 + + +def wait_out_rate_limit(send): + """Wraps `requests.Session.send` to resend a request answered with 429.""" + + def send_and_wait(request, **kwargs): + for _ in range(RATE_LIMIT_MAX_WAITS): + response = send(request, **kwargs) + if response.status_code != requests.codes.too_many_requests: + return response + # Retry-After is the time left in the window, if the API sends it. + wait = RATE_LIMIT_FALLBACK_WAIT_SECONDS + retry_after = response.headers.get("Retry-After", "") + if retry_after.isdigit(): + wait = min(int(retry_after), RATE_LIMIT_WINDOW_SECONDS) + 1 + # Without a trace in the log a wait only looks like a slow test. + print(f"API rate limit reached, resending in {wait} seconds") + time.sleep(wait) + return send(request, **kwargs) + + return send_and_wait + + class IntegrationClient(Client): def __init__(self, environ: str) -> None: api_url = graphql_url(environ) @@ -173,6 +202,8 @@ def __init__(self, environ: str) -> None: rest_endpoint=rest_endpoint, ) self.queries = [] + # GraphQL and REST calls both leave through this session's send(). + self.connection.send = wait_out_rate_limit(self.connection.send) def execute(self, query=None, params=None, check_naming=True, **kwargs): if check_naming and query is not None: @@ -184,6 +215,21 @@ def execute(self, query=None, params=None, check_naming=True, **kwargs): kwargs["timeout"] = 30.0 return super().execute(query, params, **kwargs) + def _create_project(self, input) -> Project: + # createProject returns the project even when the server failed to + # finish setting it up (seen when many are created at once). It then + # has no editor and connect_ontology cannot attach one: create again. + for _ in range(2): + project = super()._create_project(input) + if project.labeling_frontend() is not None: + return project + # The ID is what the API's logs can be searched for. + print(f"Project {project.uid} has no editor, creating it again") + project.delete() + # Jitter, so workers that failed together do not retry together. + time.sleep(randint(10, 30) / 10) + return super()._create_project(input) + class AdminClient(Client): def __init__(self, env): diff --git a/libs/labelbox/tests/data/annotation_import/test_mea_prediction_import.py b/libs/labelbox/tests/data/annotation_import/test_mea_prediction_import.py index 9a286b55e..96dbe0556 100644 --- a/libs/labelbox/tests/data/annotation_import/test_mea_prediction_import.py +++ b/libs/labelbox/tests/data/annotation_import/test_mea_prediction_import.py @@ -1,3 +1,4 @@ +import time import uuid from labelbox import parser from labelbox.data.annotation_types.annotation import ObjectAnnotation @@ -25,7 +26,19 @@ """ +# Every test that creates an MEA prediction import is skipped: the backend +# rejects the createModelErrorAnalysisPredictionImport mutation, which is the +# only thing ModelRun.add_predictions() and MEAPredictionImport.create*() call. +# Tests that only read a model run still work and stay enabled. +mea_prediction_import_removed = pytest.mark.skip( + reason="MEA (Model Error Analysis) prediction import is no longer supported " + "by the backend: the prediction import service rejects every import of " + "that type, so ModelRun.add_predictions() fails with '9 FAILED_PRECONDITION: " + "MEA prediction import has been deprecated' (python-monorepo #2515)." +) + +@mea_prediction_import_removed @pytest.mark.order(1) def test_create_from_objects( model_run_with_data_rows, @@ -54,6 +67,7 @@ def test_create_from_objects( ) +@mea_prediction_import_removed def test_create_from_objects_global_key( client, model_run_with_data_rows, @@ -83,6 +97,7 @@ def test_create_from_objects_global_key( ) +@mea_prediction_import_removed def test_create_from_objects_with_confidence( predictions_with_confidence, model_run_with_data_rows, @@ -118,6 +133,7 @@ def test_create_from_objects_with_confidence( ) +@mea_prediction_import_removed def test_create_from_objects_all_project_labels( model_run_with_all_project_labels, object_predictions_for_annotation_import, @@ -147,24 +163,62 @@ def test_create_from_objects_all_project_labels( ) +def _export_model_run_labels( + model_run: ModelRun, expected_data_row_count: int, timeout_seconds=180 +): + """The first label of each data row of a model run, once the export has them. + + `upsert_labels` returns when the labels are attached to the model run, and + the export catches up a little later: until then a data row comes out with + no label for the run. So export again while any is missing, and fail if + they have not all arrived in time. + """ + deadline = time.time() + timeout_seconds + while True: + export_task = model_run.export() + export_task.wait_till_done() + labels_by_data_row = [ + ( + data_row.json["data_row"]["id"], + data_row.json["experiments"][model_run.model_id]["runs"][ + model_run.uid + ]["labels"], + ) + for data_row in export_task.get_buffered_stream() + ] + if len(labels_by_data_row) >= expected_data_row_count and all( + labels for _, labels in labels_by_data_row + ): + return [ + (data_row_id, labels[0]) + for data_row_id, labels in labels_by_data_row + ] + + missing = [ + data_row_id + for data_row_id, labels in labels_by_data_row + if not labels + ] + assert time.time() < deadline, ( + f"After {timeout_seconds}s the export of model run {model_run.uid} " + f"had {len(labels_by_data_row)} of {expected_data_row_count} data " + f"rows, {len(missing)} of them without a label" + ) + time.sleep(5) + + def test_model_run_project_labels( model_run_with_all_project_labels: ModelRun, model_run_predictions ): model_run = model_run_with_all_project_labels - export_task = model_run.export() - export_task.wait_till_done() - stream = export_task.get_buffered_stream() + expected_data_row_count = len( + set(prediction["dataRow"]["id"] for prediction in model_run_predictions) + ) # exports to list of tuples (data_row_id, label) needed to adapt test to export v2 instead of export v1 since data rows ids are not at label level in export v2. - model_run_exported_labels = [ - ( - data_row.json["data_row"]["id"], - data_row.json["experiments"][model_run.model_id]["runs"][ - model_run.uid - ]["labels"][0], - ) - for data_row in stream - ] + model_run_exported_labels = _export_model_run_labels( + model_run, expected_data_row_count + ) labels_indexed_by_name = {} @@ -196,6 +250,7 @@ def test_model_run_project_labels( ) +@mea_prediction_import_removed def test_create_from_label_objects( model_run_with_data_rows, object_predictions_for_annotation_import, @@ -266,6 +321,7 @@ def test_create_from_label_objects( ) +@mea_prediction_import_removed def test_create_from_local_file( tmp_path, model_run_with_data_rows, @@ -301,6 +357,7 @@ def test_create_from_local_file( ) +@mea_prediction_import_removed def test_predictions_with_custom_metrics( model_run, object_predictions_for_annotation_import, @@ -328,6 +385,7 @@ def test_predictions_with_custom_metrics( ) +@mea_prediction_import_removed def test_get(client, model_run_with_data_rows, annotation_import_test_helpers): name = str(uuid.uuid4()) url = "https://storage.googleapis.com/labelbox-public-bucket/predictions_test_v2.ndjson" @@ -344,6 +402,7 @@ def test_get(client, model_run_with_data_rows, annotation_import_test_helpers): annotation_import.wait_until_done() +@mea_prediction_import_removed @pytest.mark.slow def test_wait_till_done(model_run_predictions, model_run_with_data_rows): name = str(uuid.uuid4()) diff --git a/libs/labelbox/tests/data/annotation_import/test_model_run.py b/libs/labelbox/tests/data/annotation_import/test_model_run.py index 1174115c5..5be33c8d7 100644 --- a/libs/labelbox/tests/data/annotation_import/test_model_run.py +++ b/libs/labelbox/tests/data/annotation_import/test_model_run.py @@ -52,17 +52,15 @@ def test_model_run_no_config(rand_gen, model): assert model_run.name == name -def test_model_run_delete(client, model_run): - models_before = list(client.get_models()) - model_before = models_before[0] - before = list(model_before.model_runs()) +def test_model_run_delete(model, model_run): + # The model this test created, not the newest in the organization: other + # runs of this suite create and delete models there at the same time. + before = list(model.model_runs()) + assert model_run.uid in {mr.uid for mr in before} - model_run = before[0] model_run.delete() - models_after = list(client.get_models()) - model_after = models_after[0] - after = list(model_after.model_runs()) + after = list(model.model_runs()) after_uids = {mr.uid for mr in after} assert model_run.uid not in after_uids diff --git a/libs/labelbox/tests/embedding_cleanup.py b/libs/labelbox/tests/embedding_cleanup.py index eed2a0062..ab0da8c72 100644 --- a/libs/labelbox/tests/embedding_cleanup.py +++ b/libs/labelbox/tests/embedding_cleanup.py @@ -5,10 +5,15 @@ from random import uniform from typing import Any, Callable, Iterable, List, Optional -from lbox.exceptions import LabelboxError +from lbox.exceptions import LabelboxError, MalformedQueryException EMBEDDING_NAME_PREFIX_V2 = "sdk-int-ci-v2-" EMBEDDING_CAP_ERROR_SNIPPET = "Max limit of custom embeddings" +# Creating and deleting an embedding are GraphQL mutations, and the API +# reports the cap with the MALFORMED_REQUEST code. The SDK raises that as +# MalformedQueryException, which does not derive from LabelboxError, so +# catching LabelboxError alone lets the cap error through unhealed. +EMBEDDING_CAP_ERROR_TYPES = (LabelboxError, MalformedQueryException) EMBEDDING_STALE_TTL_SECONDS = 12 * 3600 LEGACY_NAME_RE = re.compile(r"^sdk-int-[0-9a-f]{32}$") @@ -40,7 +45,7 @@ def select_stale_embeddings(embeddings: Iterable[Any], now: float) -> List[Any]: return stale_embeddings -def is_embedding_cap_error(error: LabelboxError) -> bool: +def is_embedding_cap_error(error: Exception) -> bool: return EMBEDDING_CAP_ERROR_SNIPPET in str(error) @@ -59,7 +64,7 @@ def create_embedding_with_heal( for attempt in range(1, _MAX_CREATE_ATTEMPTS + 1): try: return create_embedding() - except LabelboxError as error: + except EMBEDDING_CAP_ERROR_TYPES as error: if not is_embedding_cap_error(error): raise @@ -77,8 +82,11 @@ def create_embedding_with_heal( for embedding in stale_embeddings: try: delete_embedding(embedding) - except LabelboxError: + except EMBEDDING_CAP_ERROR_TYPES: # Another worker may have deleted the same stale embedding. + # Several jobs hit the cap at once and sweep the same + # names, and the delete is a GraphQL mutation too, so it + # can fail with either kind of error. pass last_swept_count = len(stale_embeddings) diff --git a/libs/labelbox/tests/integration/test_client_errors.py b/libs/labelbox/tests/integration/test_client_errors.py index b5444eadf..f070387cb 100644 --- a/libs/labelbox/tests/integration/test_client_errors.py +++ b/libs/labelbox/tests/integration/test_client_errors.py @@ -9,6 +9,13 @@ import labelbox.client from labelbox.schema.media_type import MediaType +# Deployed API environments replace the message of every GraphQL validation +# error with this fixed string, so that schema details (field names, "Did you +# mean ..." hints) are not exposed to clients. The error code is unchanged. +# Only an API that does not hide schema details, such as one run locally, +# still returns the detailed message. +MASKED_VALIDATION_ERROR_MESSAGE = "GraphQL validation error" + def test_missing_api_key(): key = os.environ.get(lbox.request_client._LABELBOX_API_KEY, None) @@ -41,7 +48,10 @@ def test_syntax_error(client): def test_semantic_error(client): with pytest.raises(lbox.exceptions.InvalidQueryError) as excinfo: client.execute("query {bbb {id}}", check_naming=False) - assert excinfo.value.message.startswith('Cannot query field "bbb"') + message = excinfo.value.message + assert message == MASKED_VALIDATION_ERROR_MESSAGE or message.startswith( + 'Cannot query field "bbb"' + ) def test_timeout_error(client, project): @@ -59,11 +69,24 @@ def test_timeout_error(client, project): def test_query_complexity_error(client): - with pytest.raises(lbox.exceptions.ValidationFailedError) as excinfo: + # The complexity limit is a validation rule, so where validation messages + # are masked its "Query complexity limit exceeded" message is replaced as + # well. The client then has nothing left to tell it apart from any other + # validation failure and raises InvalidQueryError; ValidationFailedError is + # only reachable against an API that does not hide schema details. + with pytest.raises( + ( + lbox.exceptions.InvalidQueryError, + lbox.exceptions.ValidationFailedError, + ) + ) as excinfo: client.execute( "{projects {datasets {dataRows {labels {id}}}}}", check_naming=False ) - assert excinfo.value.message == "Query complexity limit exceeded" + if isinstance(excinfo.value, lbox.exceptions.ValidationFailedError): + assert excinfo.value.message == "Query complexity limit exceeded" + else: + assert excinfo.value.message == MASKED_VALIDATION_ERROR_MESSAGE def test_resource_not_found_error(client): diff --git a/libs/labelbox/tests/integration/test_data_rows.py b/libs/labelbox/tests/integration/test_data_rows.py index ba3c108d5..99cc051bb 100644 --- a/libs/labelbox/tests/integration/test_data_rows.py +++ b/libs/labelbox/tests/integration/test_data_rows.py @@ -1,5 +1,6 @@ import json import os +import time import uuid from datetime import datetime, timezone from tempfile import NamedTemporaryFile @@ -1034,6 +1035,38 @@ def _wait_for_task(task, timeout_seconds=600): ), f"Task {task.uid} did not complete within {timeout_seconds}s: {task.status}" +def _create_data_row_once_global_key_is_released( + dataset, data_row_payload, global_key, timeout_seconds=180 +): + """Create a data row with the global key of a data row deleted just before. + + The key is not free the moment `delete()` returns: the import looks global + keys up in the catalog search index, and the deleted data row leaves that + index a little later. Until it has, the import reports the key as a + duplicate. So repeat the import while that is the only thing it reports, + and fail on any other error or if the key is not released in time. + """ + still_held = f"Duplicate global key: '{global_key}'" + sleep_time = 5 + deadline = time.time() + timeout_seconds + while True: + task = dataset.create_data_rows([data_row_payload]) + _wait_for_task(task) + + failed_data_rows = task.failed_data_rows + if not failed_data_rows: + return task + + messages = [row["message"] for row in failed_data_rows] + assert messages == [still_held], f"Unexpected errors: {messages}" + if time.time() >= deadline: + raise AssertionError( + f"Global key '{global_key}' was still reported as a duplicate " + f"{timeout_seconds}s after its data row was deleted" + ) + time.sleep(sleep_time) + + def test_data_row_delete_and_create_with_same_global_key( client, dataset, sample_image ): @@ -1065,8 +1098,9 @@ def test_data_row_delete_and_create_with_same_global_key( client.get_data_row(new_data_row_id).delete() # should successfully insert new datarow now - task = dataset.create_data_rows([data_row_payload]) - _wait_for_task(task) + task = _create_data_row_once_global_key_is_released( + dataset, data_row_payload, global_key_1 + ) assert task.result[0]["global_key"] == global_key_1 diff --git a/libs/labelbox/tests/integration/test_label.py b/libs/labelbox/tests/integration/test_label.py index 54e929efc..8de6c5184 100644 --- a/libs/labelbox/tests/integration/test_label.py +++ b/libs/labelbox/tests/integration/test_label.py @@ -41,12 +41,15 @@ def test_label_update(configured_project_with_label): assert label.label == "something else" -def test_label_bulk_deletion(configured_project_with_label): +def test_label_bulk_deletion(configured_project_with_label, label_helpers): project, _, _, _ = configured_project_with_label for _ in range(2): # only run twice, already have one label in the fixture project.create_label() + # A finished label import is not listed by `project.labels()` right away + # (the fixture waits for its own label the same way), so wait for all three + label_helpers.wait_for_labels(project, 3) labels = project.labels() l1 = next(labels) l2 = next(labels) diff --git a/libs/labelbox/tests/integration/test_project.py b/libs/labelbox/tests/integration/test_project.py index a1bb23d52..dd31ed33e 100644 --- a/libs/labelbox/tests/integration/test_project.py +++ b/libs/labelbox/tests/integration/test_project.py @@ -95,18 +95,18 @@ def delete_tag(tag_id: str): assert tagA.uid is not None tags = org.get_resource_tags() - lenA = len(tags) - assert lenA > 0 + assert tagA.uid in {tag.uid for tag in tags} tagB = client.get_organization().create_resource_tag(tagB) assert tagB.text == textB assert "#" + tagB.color == colorB assert tagB.uid is not None + # Look for the two tags themselves, not for a longer list: other runs of + # this suite create and delete tags in the same organization meanwhile, so + # the number of tags can go either way between two readings. tags = client.get_organization().get_resource_tags() - lenB = len(tags) - assert lenB > 0 - assert lenB > lenA + assert {tagA.uid, tagB.uid} <= {tag.uid for tag in tags} project_resource_tag = client.get_project( p1.uid diff --git a/libs/labelbox/tests/unit/test_embedding_cleanup.py b/libs/labelbox/tests/unit/test_embedding_cleanup.py index 882648d7a..cb179404b 100644 --- a/libs/labelbox/tests/unit/test_embedding_cleanup.py +++ b/libs/labelbox/tests/unit/test_embedding_cleanup.py @@ -2,8 +2,9 @@ from unittest.mock import Mock, call, patch import pytest -from lbox.exceptions import LabelboxError +from lbox.exceptions import LabelboxError, MalformedQueryException +from labelbox import Client from tests.embedding_cleanup import ( EMBEDDING_CAP_ERROR_SNIPPET, EMBEDDING_NAME_PREFIX_V2, @@ -42,8 +43,21 @@ def _v2_name(created_at: int, suffix: str = "0123456789") -> str: return f"{EMBEDDING_NAME_PREFIX_V2}{created_at}-{suffix}" -def _cap_error() -> LabelboxError: - return LabelboxError(f"{EMBEDDING_CAP_ERROR_SNIPPET}: 10") +# The message the API answers with when the organization is at the cap. +CAP_MESSAGE = f"{EMBEDDING_CAP_ERROR_SNIPPET} reached : 10 per organization" + + +def _cap_error() -> MalformedQueryException: + # The API reports the cap with the MALFORMED_REQUEST code, which the SDK + # raises as MalformedQueryException: not a LabelboxError. + return MalformedQueryException(CAP_MESSAGE) + + +def _graphql_response(body: dict) -> Mock: + response = Mock() + response.status_code = 200 + response.json.return_value = body + return response def test_build_and_parse_embedding_name_round_trip(): @@ -96,7 +110,11 @@ def test_select_stale_embeddings_keeps_only_expired_custom_v2_names(): def test_embedding_cap_error_matcher_is_fail_closed(): assert is_embedding_cap_error(_cap_error()) + assert is_embedding_cap_error(LabelboxError(CAP_MESSAGE)) assert not is_embedding_cap_error(LabelboxError("permission denied")) + assert not is_embedding_cap_error( + MalformedQueryException("Name contains invalid characters") + ) def test_workflow_boolean_input_is_parsed_defensively(): @@ -140,6 +158,56 @@ def test_heal_retries_twice_then_creates_and_deletes_only_stale_v2(): assert sleep.call_args_list == [call(4), call(4)] +def test_heal_sweeps_on_the_cap_error_client_create_embedding_raises(): + # Goes through Client.execute, so the heal is checked against the + # exception the SDK raises for the API's answer, not against a stand-in. + now_value = 1_725_000_000 + stale = _embedding( + "stale", + _v2_name(now_value - EMBEDDING_STALE_TTL_SECONDS - 1), + ) + delete = Mock() + client = Client(api_key="api_key") + responses = [ + _graphql_response( + { + "errors": [ + { + "message": CAP_MESSAGE, + "extensions": {"code": "MALFORMED_REQUEST"}, + } + ] + } + ), + _graphql_response( + { + "data": { + "createEmbedding": { + "id": "created", + "name": "custom", + "dims": 8, + "custom": True, + } + } + } + ), + ] + + with patch.object(client.connection, "send", side_effect=responses): + embedding = create_embedding_with_heal( + create_embedding=lambda: client.create_embedding("custom", 8), + list_embeddings=Mock(return_value=[stale]), + delete_embedding=delete, + sleep=Mock(), + now=Mock(return_value=now_value), + retry_delay=Mock(return_value=4), + print_fn=Mock(), + ) + + assert embedding.id == "created" + delete.assert_called_once_with(stale) + + def test_heal_third_cap_error_is_terminal_without_another_sweep_or_sleep(): final_error = _cap_error() create = Mock(side_effect=[_cap_error(), _cap_error(), final_error]) @@ -148,7 +216,7 @@ def test_heal_third_cap_error_is_terminal_without_another_sweep_or_sleep(): sleep = Mock() print_fn = Mock() - with pytest.raises(LabelboxError) as raised: + with pytest.raises(MalformedQueryException) as raised: create_embedding_with_heal( create_embedding=create, list_embeddings=list_embeddings, @@ -170,14 +238,21 @@ def test_heal_third_cap_error_is_terminal_without_another_sweep_or_sleep(): ) -def test_heal_non_cap_error_is_reraised_without_side_effects(): - non_cap_error = LabelboxError("permission denied") +@pytest.mark.parametrize( + "non_cap_error", + [ + LabelboxError("permission denied"), + MalformedQueryException("Name contains invalid characters"), + ], + ids=["LabelboxError", "MalformedQueryException"], +) +def test_heal_non_cap_error_is_reraised_without_side_effects(non_cap_error): create = Mock(side_effect=non_cap_error) list_embeddings = Mock() delete = Mock() sleep = Mock() - with pytest.raises(LabelboxError) as raised: + with pytest.raises(type(non_cap_error)) as raised: create_embedding_with_heal( create_embedding=create, list_embeddings=list_embeddings, @@ -257,3 +332,38 @@ def test_cleanup_real_run_attempts_every_candidate_and_reports_failures(): assert run_cleanup(client, dry_run=False, now=1_725_000_000) == 1 first.delete.assert_called_once_with() second.delete.assert_called_once_with() + + +@pytest.mark.parametrize( + "already_gone", + [ + MalformedQueryException("Embedding not found"), + LabelboxError("Embedding not found"), + ], +) +def test_heal_carries_on_when_a_stale_embedding_is_already_gone(already_gone): + # Jobs for several Python versions hit the cap together and sweep the same + # stale names, so a delete can find its embedding already deleted. The + # delete is a GraphQL mutation, so it can fail with either kind of error. + now_value = 1_725_000_000 + gone = _embedding( + "gone", _v2_name(now_value - EMBEDDING_STALE_TTL_SECONDS - 2) + ) + stale = _embedding( + "stale", _v2_name(now_value - EMBEDDING_STALE_TTL_SECONDS - 1) + ) + created = object() + delete = Mock(side_effect=[already_gone, None]) + + result = create_embedding_with_heal( + create_embedding=Mock(side_effect=[_cap_error(), created]), + list_embeddings=Mock(return_value=[gone, stale]), + delete_embedding=delete, + sleep=Mock(), + now=Mock(return_value=now_value), + retry_delay=Mock(return_value=4), + print_fn=Mock(), + ) + + assert result is created + assert delete.call_args_list == [call(gone), call(stale)]