analyticsapi-engines-python-sdk/tests/test_vault_calculations_api.py at master · factset/analyticsapi-engines-python-sdk · GitHub

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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import unittest
import time

from fds.analyticsapi.engines.api.components_api import ComponentsApi
from fds.analyticsapi.engines.model.component_summary import ComponentSummary
from fds.analyticsapi.engines.api.configurations_api import ConfigurationsApi
from fds.analyticsapi.engines.api.vault_calculations_api import VaultCalculationsApi
from fds.analyticsapi.engines.model.vault_calculation_parameters import VaultCalculationParameters
from fds.analyticsapi.engines.model.vault_calculation_parameters_root import VaultCalculationParametersRoot
from fds.analyticsapi.engines.model.vault_identifier import VaultIdentifier
from fds.analyticsapi.engines.model.vault_date_parameters import VaultDateParameters

from common_functions import CommonFunctions
from api_workflow import run_api_workflow_with_assertions


class TestVaultCalculationsApi(unittest.TestCase):
    def setUp(self):
        api_client = CommonFunctions.build_api_client()
        self.vault_calculations_api = VaultCalculationsApi(api_client)
        self.configurations_api = ConfigurationsApi(api_client)
        self.components_api = ComponentsApi(api_client)
        self.page_number = 1

    def test_single_unit_scenario(self):
        create_step_name = "create_calculation"
        read_status_step_name = "read_status"
        read_result_step_name = "read_result"

        def create_calculation(test_context):
            print("Creating single unit calculation")
            components = self.components_api.get_vault_components(
                document="CLIENT:/YETI/YETI-API-TEST",
                _return_http_data_only=True)
            component_summary = ComponentSummary(
                name="7D-Returns", category="General / 7D")

            component_id = [id for id in list(
                components.data.keys()) if components.data[id] == component_summary][0]
            account = "CLIENT:/YETI/YETI-API-TEST.ACCT"
            vault_account = VaultIdentifier(id=account)
            vault_dates = VaultDateParameters(
                startdate="20211231", enddate="20220131", frequency="Monthly")

            configurations = self.configurations_api.get_vault_configurations(
                account,
                _return_http_data_only = True)
            configuration_id = list(configurations.data.keys())[0]

            vault_calculation_parameters = {"1": VaultCalculationParameters(componentid=component_id, account=vault_account,
                                                                            dates=vault_dates, configid=configuration_id)}

            vault_calculation_parameter_root = VaultCalculationParametersRoot(
                data=vault_calculation_parameters)

            post_and_calculate_response = self.vault_calculations_api.post_and_calculate(
                vault_calculation_parameters_root=vault_calculation_parameter_root
            )

            self.assertTrue(post_and_calculate_response[1] == 201 or post_and_calculate_response[1] == 202,
                            "Response for create_calculation should have been 201 or 202")

            if post_and_calculate_response[1] == 201:
                return {
                    "continue_workflow": False,
                    "next_request": None,
                    "test_context": None
                }
            elif post_and_calculate_response[1] == 202:
                test_context["calculation_id"] = post_and_calculate_response[2]["X-Factset-Api-Calculation-Id"]
                return {
                    "continue_workflow": True,
                    "next_request": read_status_step_name,
                    "test_context": test_context
                }

        def read_calculation_status(test_context):
            print("Reading single unit calculation status")
            calculation_id = test_context["calculation_id"]
            print("Calculation Id: " + calculation_id)

            status_response = self.vault_calculations_api.get_calculation_status_by_id(id=calculation_id)

            self.assertTrue(status_response[1] == 202 and (
                status_response[0].data.status in ("Queued", "Executing")))

            while status_response[1] == 202 and (status_response[0].data.status in ("Queued", "Executing")):
                max_age = '5'
                age_value = status_response[2].get("cache-control")
                if age_value is not None:
                    max_age = age_value.replace("max-age=", "")
                print('Sleeping: ' + max_age)
                time.sleep(int(max_age))
                status_response = self.vault_calculations_api.get_calculation_status_by_id(id=calculation_id)

                test_context["calculation_units"] = status_response[0].data.units.items()

            return {
                "continue_workflow": True,
                "next_request": read_result_step_name,
                "test_context": test_context
            }

        def read_calculation_unit_result(test_context):
            calculation_id = test_context["calculation_id"]
            for (calculation_unit_id, calculation_unit) in test_context["calculation_units"]:
                result_response = self.vault_calculations_api.get_calculation_unit_result_by_id(id=calculation_id,
                                                                                                unit_id=calculation_unit_id)
                self.assertEqual(
                    result_response[1], 200, "Get calculation result should have succeeded")

            return {
                "continue_workflow": False,
                "next_request": read_result_step_name,
                "test_context": test_context
            }

        workflow_specification = {
            create_step_name: create_calculation,
            read_status_step_name: read_calculation_status,
            read_result_step_name: read_calculation_unit_result
        }
        starting_request = workflow_specification['create_calculation']
        test_context = {}
        run_api_workflow_with_assertions(
            workflow_specification, starting_request, test_context)

    def test_get_all_calculations_scenario(self):
        response = self.vault_calculations_api.get_all_calculations(page_number=self.page_number)
        self.assertEqual(response[1], 200, "Response should be 200 - Success")


if __name__ == '__main__':
    unittest.main(failfast=True)