IntegrationTest/main.pymodifiedimport logging
from src.portfolio.assert_queries import (
assert_fn_get_portfolio_issues_json,
assert_fn_get_portfolio_progress_json,
generate_expected_issues,
generate_expected_progress,
get_actual_issues_values,
get_actual_progress_values,
)
from src.portfolio.pg_setup import (
ProjectOverlap,
get_currency_id,
get_issues_fks,
setup_issues,
setup_progress_data,
setup_projects_for_portfolio,
)
from src.postgresql_helper import PostgresqlHelper
from src.key_milestone.setup import setup_key_milestones
from src.key_milestone.assert_queries import assert_fn_get_portfolio_key_milestones
from src.name_unique.scenario import assert_name_unique_indexes
logging.basicConfig(level=logging.INFO)
pgsql = PostgresqlHelper(port=5490)
currency_id = get_currency_id(pgsql, "GBP")
for value in ProjectOverlap:
logging.info(f"Testing with project calendar overlap: {value.name}")
projects = setup_projects_for_portfolio(
pgsql=pgsql, currency_id=currency_id, number_of_projects=3
)
portfolio_id = projects[0]["PortfolioId"]
input_df = setup_progress_data(
projects=projects,
project_calendar_overlap=value,
pgsql=pgsql,
)
assert_fn_get_portfolio_progress_json(
actual=get_actual_progress_values(pgsql, portfolio_id=portfolio_id),
expected=generate_expected_progress(input_df),
)
logging.info(f"Assertions passed for overlap type: {value.name}.")
logging.info("Testing issues aggregation for portfolio...")
projects = setup_projects_for_portfolio(
pgsql=pgsql, currency_id=currency_id, number_of_projects=3
)
portfolio_id = projects[0]["PortfolioId"]
i_open_status, i_closed_status, i_types, i_severty = get_issues_fks(
pgsql, [proj["ProjectShardId"] for proj in projects]
)
input_issue_df = setup_issues(
pgsql=pgsql,
projects=projects,
issue_type_id_lookup=i_types,
issue_severity_category_ids=i_severty,
issue_open_status_id_lookup=i_open_status,
issue_closed_status_id_lookup=i_closed_status,
count=50,
closed_fraction=0.5,
seed=42,
)
assert_fn_get_portfolio_issues_json(
actual=get_actual_issues_values(
pgsql, portfolio_id=portfolio_id, interval_type="MONTH"
),
expected=generate_expected_issues(
projects=projects,
issues_df=input_issue_df,
currency_code="GBP",
interval_type="MONTH",
),
)
assert_fn_get_portfolio_issues_json(
actual=get_actual_issues_values(
pgsql, portfolio_id=portfolio_id, interval_type="QUARTER"
),
expected=generate_expected_issues(
projects=projects,
issues_df=input_issue_df,
currency_code="GBP",
interval_type="QUARTER",
),
)
# -------------------------------------------------------
# Key milestones (reporting.fn_GetPortfolioKeyMilestones)
# -------------------------------------------------------
logging.info("Testing key milestones for portfolio...")
km_currency_id = get_currency_id(pgsql, "GBP")
# Two separate portfolios so we also verify the function only returns milestones
# belonging to the requested portfolio.
for label in ("portfolio A", "portfolio B"):
km_projects = setup_projects_for_portfolio(
pgsql=pgsql, currency_id=km_currency_id, number_of_projects=3
)
km_portfolio_id = km_projects[0]["PortfolioId"]
km_expected = setup_key_milestones(pgsql=pgsql, projects=km_projects)
assert_fn_get_portfolio_key_milestones(
pgsql=pgsql, portfolio_id=km_portfolio_id, expected=km_expected
)
logging.info("Key milestone assertions passed for %s.", label)
# -------------------------------------------------------
# Per-project unique Name on xyz.Asset / xyz.CommissioningSystem
# -------------------------------------------------------
logging.info("Testing per-project unique Name indexes for Asset/CommissioningSystem...")
name_unique_projects = setup_projects_for_portfolio(
pgsql=pgsql, currency_id=currency_id, number_of_projects=2
)
assert_name_unique_indexes(pgsql=pgsql, projects=name_unique_projects)
logging.info("Per-project unique Name assertions passed.")
logging.info("All assertions passed successfully.")