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.biggest_file.setup import setup_biggest_file
from src.biggest_file.assert_queries import assert_fn_get_biggest_file
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)
# -------------------------------------------------------
# Biggest file (xyz.fn_GetBiggestFile)
# -------------------------------------------------------
logging.info("Testing biggest file selection for project...")
bf_currency_id = get_currency_id(pgsql, "GBP")
# Two projects: the first gets a full file set (largest live file expected, soft-deleted/NULL/tie
# cases covered), the second gets only a soft-deleted file (zero-rows path). Using two projects
# also verifies the function filters by ProjectShardId only.
bf_projects = setup_projects_for_portfolio(
pgsql=pgsql, currency_id=bf_currency_id, number_of_projects=2
)
bf_expected = setup_biggest_file(pgsql=pgsql, projects=bf_projects)
for bf_project in bf_projects:
assert_fn_get_biggest_file(
pgsql=pgsql,
project_id=bf_project["ProjectId"],
expected_file_reference_id=bf_expected[bf_project["ProjectId"]],
)
logging.info("Biggest file assertions passed.")
logging.info("All assertions passed successfully.")