IntegrationTest/main.pymodified
import 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.")