import re import pytest from monday_com_orca_backend.api.routers.data.enums import RequestParam from monday_com_orca_backend.api.routers.enums import BaseRequestParam from monday_com_orca_backend.enums import HttpMethod from tests.conftest import TEST_RELATIONSHIP_MANAGER_EMAIL from .conftest import is_non_empty_list ROUTE_PREFIX: str = "/api/data/" class TestCountries: endpoint = f"{ROUTE_PREFIX}countries" @pytest.fixture def response_data(self, get_collection_response_data): """Fetch country data and validate the response.""" data = get_collection_response_data( self.endpoint, expect_empty=False, expect_more=False ) return data @pytest.mark.parametrize( "expected_key,validator", [ ("id", lambda x: isinstance(x, int) and x > 0), ("name", lambda x: isinstance(x, str) and len(x.replace(" ", "")) > 2), ("code", lambda x: x is None or re.match(r"^[A-Z]{2}$", x)), ], ) def test_get_countries_valid(self, response_data, expected_key, validator): """Validate the response structure and data types.""" for country in response_data: assert ( expected_key in country ), f"Expected key '{expected_key}' in country object" assert validator(country[expected_key]), ( f"Validator failed for key '{expected_key}' with value " f"{country[expected_key]}" ) _expect_country_codes_sample: set[str] = {"GB", "ES", "FR", "DE", "IT", "US", "CA"} def test_countries_non_regions_include_codes(self, response_data): """Validate that countries include specific ISO codes, using a sample.""" response_codes = {country["code"] for country in response_data} assert self._expect_country_codes_sample.issubset(response_codes), ( f"Expected country codes {self._expect_country_codes_sample} " f"to be present in the response." ) def test_countries_include_regions(self, response_data): """Validate that the response includes regions.""" assert any( country["name"] == "GSA" and country["code"] is None for country in response_data ), "Expected 'GSA' region in the response, with code None (as it's a region)" class TestLabelMeta: endpoint = f"{ROUTE_PREFIX}label_meta" def test_get_label_meta_valid(self, test_client): label_ids = {33960, 25223, 22221} url = ( f"{self.endpoint}" f"?{RequestParam.LABEL_IDS}={','.join(map(str, label_ids))}" ) response = test_client.get(url) assert response.status_code == 200 response_content = response.json() data = response_content["data"] returned_label_ids = {item["id"] for item in data} assert returned_label_ids == label_ids assert ( response_content["more"] is False ), "Expected 'more' to be False, as this endpoint does not support pagination" class TestRelationshipManagerClients: endpoint = f"{ROUTE_PREFIX}relationship_manager_clients" def test_get_rel_manager_clients_email( self, test_client, get_collection_response_data ): limit = 25 data = get_collection_response_data( self.endpoint, method=HttpMethod.POST, json={ RequestParam.RELATIONSHIP_MANAGER_EMAILS: [ TEST_RELATIONSHIP_MANAGER_EMAIL ], BaseRequestParam.LIMIT: limit, }, expect_empty=False, expect_more=False, ) assert len(data) > 0, ( "Expected at least one object in the response for the given email, " f"but got none for email {TEST_RELATIONSHIP_MANAGER_EMAIL}" ) relationship_manager_emails = { obj["relationship_manager_email"] for obj in data } relationship_manager_ids = {obj["relationship_manager_id"] for obj in data} assert len(relationship_manager_emails) == 1, ( "Expected all objects to have the same 'relationship_manager_email', " f"but got {relationship_manager_emails}" ) assert len(relationship_manager_ids) == 1, ( "Expected all objects to have the same 'relationship_manager_id', " f"but got {relationship_manager_ids}" ) assert TEST_RELATIONSHIP_MANAGER_EMAIL in relationship_manager_emails, ( f"Expected the email {TEST_RELATIONSHIP_MANAGER_EMAIL} to be present " "in the response, but it was not found" ) assert ( len(data) == limit ), f"Expected exactly {limit} objects in the response, but got {len(data)}" class TestRelationshipManagerStats: endpoint = f"{ROUTE_PREFIX}relationship_manager_stats" _yearly_growth_fields = ["yearly_growth", "yearly_growth_fiscal"] _top_clients_fields = ["top_clients_ltm", "top_clients_fiscal_year"] @pytest.fixture def response_data(self, get_collection_response_data): """Fetch relationship manager stats data and validate the response.""" data = get_collection_response_data( self.endpoint, expect_empty=False, expect_more=False ) return data @pytest.fixture def response_filter_by_label_id(self, test_client, response_data): use_top_clients_field = self._top_clients_fields[0] first_rel_manager_with_at_least_three_clients = next( (obj for obj in response_data if len(obj[use_top_clients_field]) >= 3), None ) assert first_rel_manager_with_at_least_three_clients, ( f"Expected at least one object with a " f"'{use_top_clients_field}' list of length >= 3" ) sample_label_ids = set( list( { client_["label_id"] for client_ in first_rel_manager_with_at_least_three_clients[ use_top_clients_field ] } )[:3] ) resp = test_client.get( f"{self.endpoint}" f"?{RequestParam.LABEL_IDS}={','.join(map(str, sample_label_ids))}" ) resp_content = resp.json() data = resp_content["data"] assert is_non_empty_list(data) return data, sample_label_ids def test_get_rel_manager_stats_valid(self, response_data): assert all(obj["relationship_manager"] for obj in response_data), ( "All objects should have a 'relationship_manager' key and" " its value must be a name" ) def test_get_rel_manager_stats_fiscal_year_valid(self, response_data): assert all(1985 < obj["fiscal_year"] < 2100 for obj in response_data), ( "All objects should have a 'fiscal_year' key and" " its value must be a valid fiscal year" ) def test_get_rel_manager_stats_fiscal_year_ytd_gross_revenue(self, response_data): # Find an object with fiscal_year_ytd_gross_revenue_usd # and fiscal_year_ytd_gross_revenue_usd_prev_year not 0 for this # test, it will suffice. sample_obj = next( ( obj for obj in response_data if obj.get("fiscal_year_ytd_gross_revenue_usd") and obj.get("fiscal_year_ytd_gross_revenue_usd_prev_year") ), None, ) assert sample_obj, ( "Expected at least one object with 'fiscal_year_ytd_gross_revenue_usd' " "and 'fiscal_year_ytd_gross_revenue_usd_prev_year' keys, with their " "values not being 0" ) fiscal_year_ytd_gross_revenue_usd = sample_obj[ "fiscal_year_ytd_gross_revenue_usd" ] fiscal_year_ytd_gross_revenue_usd_prev_year = sample_obj[ "fiscal_year_ytd_gross_revenue_usd_prev_year" ] # Unlikely that two non-zero values are equal, so we assert they are different # to be sure the SQL query is not using the same value for both fields. assert ( fiscal_year_ytd_gross_revenue_usd != fiscal_year_ytd_gross_revenue_usd_prev_year ), ( "Expected 'fiscal_year_ytd_gross_revenue_usd' and " "'fiscal_year_ytd_gross_revenue_usd_prev_year' to be different, " f"but got {fiscal_year_ytd_gross_revenue_usd} and " f"{fiscal_year_ytd_gross_revenue_usd_prev_year}" ) def test_get_rel_manager_stats_certain_fields_are_lists_of_dicts( self, response_data ): first_object = response_data[0] fields_must_be_lists_of_dicts = [ "yearly_growth", "yearly_growth_fiscal", *self._top_clients_fields, ] for field_name in fields_must_be_lists_of_dicts: assert isinstance(first_object[field_name], list), ( f"Expected field '{field_name}' to be a list, " f"but got {type(first_object[field_name])}" ) assert all( isinstance(item, dict) for item in first_object[field_name] ), f"Expected all items in field '{field_name}' to be dictionaries" def test_get_rel_manager_stats_filter_by_label_id_top_clients( self, response_filter_by_label_id ): data, sample_label_ids = response_filter_by_label_id for rel_manager in data: for field_name in self._top_clients_fields: top_clients = rel_manager[field_name] assert is_non_empty_list( top_clients ), f"Expected at least one object in the '{field_name}' list" assert all( top_client["label_id"] in sample_label_ids for top_client in top_clients ), "All objects expected to have a sample label_id" def test_get_rel_manager_stats_yearly_growths_different(self, response_data): for sample_rm in response_data: yearly_growth_values = [ sample_rm[field] for field in self._yearly_growth_fields ] try: assert any( val != yearly_growth_values[0] for val in yearly_growth_values[1:] ), ( f"Expected at least one difference among " f"{self._yearly_growth_fields} for " f"relationship_manager_email=" f"{sample_rm.get('relationship_manager_email')}" ) except AssertionError: # If all values are [], do not fail the test, as this might happen if # the relationship manager has no clients or no revenue data. if all(val == [] for val in yearly_growth_values): continue raise @pytest.mark.parametrize("field_name", _yearly_growth_fields) def test_get_rel_manager_stats_filter_by_label_id_yearly_growth( self, test_client, response_data, response_filter_by_label_id, field_name ): """Check that yearly growth data changes when filtering by label_id.""" unfiltered_data = response_data filtered_data, sample_label_ids = response_filter_by_label_id for rel_manager in filtered_data: email = rel_manager["relationship_manager_email"] unfiltered = next( obj for obj in unfiltered_data if obj["relationship_manager_email"] == email ) filtered_growth = { item["year"]: item["current_revenue_usd"] for item in rel_manager[field_name] } unfiltered_growth = { item["year"]: item["current_revenue_usd"] for item in unfiltered[field_name] } assert set(filtered_growth) <= set( unfiltered_growth ), "Filtered yearly growth years should be a subset of unfiltered" for year, current in filtered_growth.items(): assert current < unfiltered_growth[year], ( "Current revenue in the filtered data should be less " "than in the unfiltered data for the same year" ) def test_get_rel_manager_stats_filter_by_date_yearmonth( self, get_collection_response_data ): date_yearmonth = "2024-01" data = get_collection_response_data( f"{self.endpoint}?{RequestParam.DATE_YEARMONTH}={date_yearmonth}", expect_empty=False, expect_more=False, ) assert data, "Expected at least one object in the response, but got none" assert all(obj["period_date"] == f"{date_yearmonth}-01" for obj in data), ( f"All objects in the response should have a 'date_yearmonth'" f" of '{date_yearmonth}-01'" ) class TestScorecardFilterMeta: endpoint = f"{ROUTE_PREFIX}scorecard_filter_meta" def test_get_scorecard_filter_meta_all( self, test_client, get_collection_response_data ): data = get_collection_response_data( self.endpoint, method=HttpMethod.POST, json={}, expect_empty=False, expect_more=False, ) null_relationship_manager = next( (obj for obj in data if obj["relationship_manager_email"] is None), None ) assert null_relationship_manager is not None, ( "Expected at least one object with 'relationship_manager_email' set to " "None. This is the 'null' relationship manager." ) assert ( null_relationship_manager["relationship_manager"] is None ), "Expected the 'name' of the null relationship manager to be None" first_non_null_relationship_manager = next( (obj for obj in data if obj["relationship_manager_email"] is not None), None ) assert first_non_null_relationship_manager is not None, ( "Expected at least one object with 'relationship_manager_email' " "set to a non-null value" ) for obj in data: for date_field in [ "min_period_date", "max_period_date", ]: if date_field in obj and obj[date_field]: assert re.match(r"^\d{4}-\d{2}-\d{2}$", obj[date_field]), ( f"Expected '{date_field}' to be in 'YYYY-MM-DD' format, " f"but got {obj[date_field]}" ) def test_get_scorecard_filter_meta_one( self, test_client, get_collection_response_data ): data = get_collection_response_data( self.endpoint, method=HttpMethod.POST, json={ RequestParam.RELATIONSHIP_MANAGER_EMAILS: [ TEST_RELATIONSHIP_MANAGER_EMAIL ], }, expect_empty=False, expect_more=False, ) assert ( len(data) == 1 ), f"Expected exactly one object in the response, but got {len(data)}" first_obj = data[0] assert ( first_obj["relationship_manager_email"] == TEST_RELATIONSHIP_MANAGER_EMAIL ), ( "Expected the first object to have the same email as the one used" " to query the endpoint" ) first_client = first_obj["clients"][0] assert ( first_client["label_id"] > 0 ), "Expected the first client's 'label_id' to be a positive integer" assert first_client[ "name" ], "Expected the first client's 'name' to be a non-empty string" class TestScorecardRelationshipManagers: endpoint = f"{ROUTE_PREFIX}scorecard_relationship_managers" def test_get_scorecard_relationship_managers_valid( self, get_collection_response_data ): data = get_collection_response_data( self.endpoint, expect_empty=False, expect_more=False, ) assert is_non_empty_list(data), "Expected at least one object in the response" expected_fields = { "email", "full_name", } for obj in data: for field in expected_fields: assert field in obj, f"Expected field '{field}' in object {obj}" assert isinstance(obj[field], str) and obj[field], ( f"Expected field '{field}' to be a non-empty string, " f"but got {obj[field]}" ) class TestScorecardClients: endpoint = f"{ROUTE_PREFIX}scorecard_clients" def test_get_rel_manager_scorecard_clients_valid( self, get_collection_response_data ): data = get_collection_response_data( f"{self.endpoint}" f"?{RequestParam.RELATIONSHIP_MANAGER_EMAILS}={TEST_RELATIONSHIP_MANAGER_EMAIL}", expect_empty=False, expect_more=False, ) assert all( obj["label_id"] for obj in data[:10] ), "All objects in sample should have a 'label_id' key" def test_get_rel_manager_scorecard_clients_filter_by_label_id( self, test_client, get_collection_response_data ): limit = 500 data = get_collection_response_data( f"{self.endpoint}?{BaseRequestParam.LIMIT}={limit}", expect_empty=False, expect_more=True, ) assert ( len(data) <= limit ), f"Expected max {limit} objects in the response, but got a greater number" data, sample_label_ids = _fetch_1000_records_filtered_by_label_ids_sample( test_client, self.endpoint, data ) assert all( obj["label_id"] in sample_label_ids for obj in data ), "All objects expected to have a sample label_id" def test_get_rel_manager_scorecard_clients_filter_by_date( self, get_collection_response_data ): limit = 500 date_yearmonth = "2024-01" data = get_collection_response_data( f"{self.endpoint}" f"?{BaseRequestParam.LIMIT}={limit}" f"&{RequestParam.DATE_YEARMONTH}={date_yearmonth}", expect_empty=False, ) assert all(obj["period_date"] == f"{date_yearmonth}-01" for obj in data), ( f"All objects in the response should have " f"a 'period_date' of '{date_yearmonth}-01'" ) class TestReleases: endpoint = f"{ROUTE_PREFIX}releases" def test_releases_rel_manager_dates(self, get_collection_response_data): data = get_collection_response_data( f"{self.endpoint}" f"?{RequestParam.RELATIONSHIP_MANAGER_EMAILS}={TEST_RELATIONSHIP_MANAGER_EMAIL}" f"&{BaseRequestParam.DATE_FROM}=1991-01-01" f"&{BaseRequestParam.DATE_TO}=1991-12-31", expect_empty=False, expect_more=False, ) assert all( obj["relationship_manager_email"] == TEST_RELATIONSHIP_MANAGER_EMAIL for obj in data ), ( "All objects in sample should have the same email as the one used" " to query the endpoint" ) assert all( obj["label_id"] for obj in data[:10] ), "All objects in sample should have a 'label_id' key" assert all( "date_release" in obj and obj["date_release"][:4] == "1991" for obj in data ), "All objects expected to have a 'date_release' in the year 1991" def test_releases_limit_and_priorities(self, get_collection_response_data): priority = "B" data = get_collection_response_data( f"{self.endpoint}" f"?{BaseRequestParam.LIMIT}=50" f"&{RequestParam.PRIORITIES}={priority}", expect_empty=False, ) assert ( len(data) == 50 ), "Expected 50 objects in the response, but got a different number" for obj in data: priorities = obj["priorities"] assert len(priorities) > 0, ( "Expected all objects to have a 'priority' key with a non-empty value, " "as we're filtering by priority" ) assert all( p["priority"] == priority for p in priorities ), f"Expected all priorities to be '{priority}', but got different values" def test_releases_rel_manager_order_defaults(self, get_collection_response_data): expected_order = "desc" expected_order_by = "date_release" data = get_collection_response_data( f"{self.endpoint}" f"?{RequestParam.RELATIONSHIP_MANAGER_EMAILS}={TEST_RELATIONSHIP_MANAGER_EMAIL}" f"&{BaseRequestParam.LIMIT}=100", expect_empty=False, expect_more=False, ) assert all( data[i][expected_order_by] >= data[i + 1][expected_order_by] for i in range(len(data) - 1) ), f"Data is not sorted in {expected_order} order by {expected_order_by}" def test_releases_rel_manager_order_custom(self, get_collection_response_data): expected_order = "asc" expected_order_by = "label_id" data = get_collection_response_data( f"{self.endpoint}" f"?{RequestParam.RELATIONSHIP_MANAGER_EMAILS}={TEST_RELATIONSHIP_MANAGER_EMAIL}" f"&{BaseRequestParam.ORDER}={expected_order}" f"&{BaseRequestParam.ORDER_BY}={expected_order_by}" f"&{BaseRequestParam.LIMIT}=100", expect_empty=False, ) assert all( data[i][expected_order_by] <= data[i + 1][expected_order_by] for i in range(len(data) - 1) ), f"Data is not sorted in {expected_order} order by {expected_order_by}" def test_get_rel_manager_releases_filter_by_label_id( self, test_client, get_collection_response_data ): data = get_collection_response_data( f"{self.endpoint}" f"?{RequestParam.RELATIONSHIP_MANAGER_EMAILS}={TEST_RELATIONSHIP_MANAGER_EMAIL}" f"&{BaseRequestParam.LIMIT}=10000", expect_empty=False, expect_more=False, ) assert is_non_empty_list( data ), "Expected at least one object in the response, but got none" # We fetched a large number of records to ensure we have enough data # for getting a sample of 1000 records data, sample_label_ids = _fetch_1000_records_filtered_by_label_ids_sample( test_client, self.endpoint, data ) assert {obj["label_id"] for obj in data} <= set( sample_label_ids ), "All objects expected to have one of the sample label_ids" def _fetch_1000_records_filtered_by_label_ids_sample( client, endpoint, data ) -> tuple[list, set]: """Fetch a sample of 1000 records filtered by 3 random unique label_ids extracted from the provided data. """ sample_label_ids = set(list({obj["label_id"] for obj in data})[:3]) assert len(sample_label_ids) > 1, ( "Expected at least 2 unique label_ids in the sample data, to test" " filtering by multiple label_ids" ) response = client.get( f"{endpoint}" f"?{RequestParam.LABEL_IDS}={','.join(map(str, sample_label_ids))}" f"&{BaseRequestParam.LIMIT}=1000" ) data = response.json()["data"] assert is_non_empty_list(data) return data, sample_label_ids