File: test_feature_importance.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (61 lines) | stat: -rw-r--r-- 2,931 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from devtools_testutils import AzureRecordedTestCase, recorded_by_proxy
import personalizer_helpers
import time
from datetime import date
import uuid


class TestFeatureImportances(AzureRecordedTestCase):
    @personalizer_helpers.PersonalizerPreparer()
    @recorded_by_proxy
    def test_run_feature_importance(self, **kwargs):
        variables = kwargs.pop("variables", {})
        feature_importance_id = variables.setdefault("test_run_feature_importance_id", str(uuid.uuid4()))
        personalizer_endpoint = kwargs.pop("personalizer_preset_endpoint_single_slot")
        personalizer_api_key = kwargs.pop("personalizer_preset_api_key_single_slot")
        client = personalizer_helpers.create_personalizer_admin_client(personalizer_endpoint, personalizer_api_key)
        feature_importance_name = "python_sdk_test_feature_imp"
        start_time = date.fromisoformat("2022-09-20")
        end_time = date.fromisoformat("2022-09-30")
        feature_importance_contract = {
            "name": feature_importance_name,
            "startTime": start_time,
            "endTime": end_time,
        }
        client.create_feature_importance(feature_importance_id, feature_importance_contract)
        self.wait_for_feature_importance_to_finish(client, feature_importance_id)
        feature_importance = client.get_feature_importance(feature_importance_id)
        assert feature_importance["id"] == feature_importance_id
        assert feature_importance["name"] == feature_importance_name
        assert feature_importance["status"] == "Succeeded"
        client.delete_feature_importance(feature_importance_id)
        return variables

    @personalizer_helpers.PersonalizerPreparer()
    @recorded_by_proxy
    def test_list_feature_importances(self, **kwargs):
        personalizer_endpoint = kwargs.pop("personalizer_endpoint_single_slot")
        personalizer_api_key = kwargs.pop("personalizer_api_key_single_slot")
        client = personalizer_helpers.create_personalizer_admin_client(personalizer_endpoint, personalizer_api_key)
        client.list_feature_importances()

    def is_feature_importance_final(self, client, feature_importance_id):
        feature_importance = client.get_feature_importance(feature_importance_id)
        return (
            feature_importance["status"] == "Succeeded"
            or feature_importance["status"] == "Failed"
            or feature_importance["status"] == "Timeout"
            or feature_importance["status"] == "Canceled"
        )

    def wait_for_feature_importance_to_finish(self, client, feature_importance_id):
        while not self.is_feature_importance_final(client, feature_importance_id):
            self.sleep(60)

    def sleep(self, delay):
        if self.is_live:
            time.sleep(delay)