mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 13:01:38 +00:00
* merge branch develop with main * bump desktop version * fix the merge * sync Cargo.lock to desktop 2.0.0 * fix cli conftest.py --------- Co-authored-by: deeleeramone <>
3526 lines
123 KiB
Python
3526 lines
123 KiB
Python
"""Tests for openbb_imf.utils.table_presentation."""
|
|
|
|
# ruff: noqa: I001
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from openbb_core.app.model.abstract.warning import OpenBBWarning
|
|
|
|
from openbb_imf.utils.table_presentation import (
|
|
HierarchyContext,
|
|
build_order_title_level,
|
|
check_missing_country_data,
|
|
extract_unit_from_label,
|
|
extract_unit_scale_from_title,
|
|
format_unit_suffix,
|
|
is_bop_suffix_only,
|
|
parse_unit_and_scale,
|
|
pivot_indicator_mode,
|
|
pivot_table_data,
|
|
pivot_table_mode,
|
|
strip_title_suffix,
|
|
)
|
|
|
|
|
|
class TestExtractUnitFromLabel:
|
|
"""Tests for extract_unit_from_label."""
|
|
|
|
def test_empty_returns_none(self):
|
|
"""Empty string yields None."""
|
|
assert extract_unit_from_label("") is None
|
|
|
|
def test_paren_with_unit_keyword(self):
|
|
"""Trailing parenthetical containing a keyword is returned."""
|
|
assert (
|
|
extract_unit_from_label("GDP (Millions of US Dollar)")
|
|
== "Millions of US Dollar"
|
|
)
|
|
|
|
def test_paren_no_keyword_returns_none(self):
|
|
"""Trailing parenthetical with no unit keyword returns None."""
|
|
assert extract_unit_from_label("GDP (seasonally adjusted)") is None
|
|
|
|
def test_paren_no_leading_space_returns_none(self):
|
|
"""A '(' without leading space is not treated as a unit."""
|
|
assert extract_unit_from_label("(USD)") is None
|
|
|
|
def test_comma_per_phrase(self):
|
|
"""A ', X per Y' suffix yields the last part."""
|
|
assert (
|
|
extract_unit_from_label("Trade, US dollars per metric tonne")
|
|
== "US dollars per metric tonne"
|
|
)
|
|
|
|
def test_comma_unit_keyword(self):
|
|
"""A ', percent' style suffix is returned."""
|
|
assert extract_unit_from_label("Inflation, percent") == "percent"
|
|
|
|
def test_comma_no_keyword_returns_none(self):
|
|
"""A trailing comma part that has no keyword returns None."""
|
|
assert extract_unit_from_label("Trade, exports") is None
|
|
|
|
def test_no_comma_no_paren(self):
|
|
"""A plain label returns None."""
|
|
assert extract_unit_from_label("Population") is None
|
|
|
|
|
|
class TestParseUnitAndScale:
|
|
"""Tests for parse_unit_and_scale."""
|
|
|
|
def test_empty_returns_none_none(self):
|
|
"""Empty input yields (None, None)."""
|
|
assert parse_unit_and_scale("") == (None, None)
|
|
assert parse_unit_and_scale(None) == (None, None)
|
|
|
|
def test_percent_of_pattern(self):
|
|
"""'Percent of GDP' splits into ('Percent', 'of Gdp')."""
|
|
unit, scale = parse_unit_and_scale("Percent of GDP")
|
|
assert unit == "Percent"
|
|
assert scale == "of Gdp"
|
|
|
|
def test_per_pattern(self):
|
|
"""'US Dollar per metric tonne' splits into unit and scale."""
|
|
unit, scale = parse_unit_and_scale("US Dollar per metric tonne")
|
|
assert unit == "US Dollar"
|
|
assert scale == "Per Metric Tonne"
|
|
|
|
def test_scale_prefix(self):
|
|
"""'Millions, X' yields unit X and scale 'Millions'."""
|
|
unit, scale = parse_unit_and_scale("Millions, US Dollar")
|
|
assert unit == "US Dollar"
|
|
assert scale == "Millions"
|
|
|
|
def test_scale_suffix(self):
|
|
"""'X, Millions' yields unit X and scale 'Millions'."""
|
|
unit, scale = parse_unit_and_scale("US Dollar, Millions")
|
|
assert unit == "US Dollar"
|
|
assert scale == "Millions"
|
|
|
|
def test_unit_keyword_at_end(self):
|
|
"""A trailing unit keyword after comma extracts unit."""
|
|
unit, scale = parse_unit_and_scale("Some scale, Percent")
|
|
assert unit == "Percent"
|
|
assert scale == "Some scale"
|
|
|
|
def test_scale_only(self):
|
|
"""A bare scale word becomes scale with None unit."""
|
|
unit, scale = parse_unit_and_scale("Per capita")
|
|
assert unit is None
|
|
assert scale == "Per capita"
|
|
|
|
def test_unknown_string_returns_as_unit(self):
|
|
"""An unrecognized string is returned as a unit only."""
|
|
unit, scale = parse_unit_and_scale("Strange unit")
|
|
assert unit == "Strange unit"
|
|
assert scale is None
|
|
|
|
|
|
class TestStripTitleSuffix:
|
|
"""Tests for strip_title_suffix."""
|
|
|
|
def test_transactions_suffix(self):
|
|
"""', Transactions' suffix is stripped."""
|
|
assert strip_title_suffix("Goods, Transactions") == "Goods"
|
|
|
|
def test_stocks_suffix(self):
|
|
"""', Stocks' suffix is stripped."""
|
|
assert strip_title_suffix("Assets, Stocks") == "Assets"
|
|
|
|
def test_paren_unit_suffix_stripped(self):
|
|
"""A trailing parenthetical with a unit pattern is stripped."""
|
|
assert strip_title_suffix("GDP (Millions)") == "GDP"
|
|
|
|
def test_paren_non_unit_kept(self):
|
|
"""A trailing parenthetical without unit pattern is kept."""
|
|
assert strip_title_suffix("GDP (Seasonally adjusted)") == (
|
|
"GDP (Seasonally adjusted)"
|
|
)
|
|
|
|
def test_plain_unchanged(self):
|
|
"""A plain title is returned unchanged."""
|
|
assert strip_title_suffix("GDP") == "GDP"
|
|
|
|
|
|
class TestFormatUnitSuffix:
|
|
"""Tests for format_unit_suffix."""
|
|
|
|
def test_both(self):
|
|
"""Unit and scale combine into ' (unit, scale)'."""
|
|
assert format_unit_suffix("US Dollar", "Millions") == " (US Dollar, Millions)"
|
|
|
|
def test_unit_only(self):
|
|
"""Unit alone yields ' (unit)'."""
|
|
assert format_unit_suffix("Percent", None) == " (Percent)"
|
|
|
|
def test_skips_units_scale(self):
|
|
"""The literal scale 'Units' is filtered out."""
|
|
assert format_unit_suffix("Item", "Units") == " (Item)"
|
|
|
|
def test_dash_and_nan_filtered(self):
|
|
"""Values like '-' and 'nan' are filtered out."""
|
|
assert format_unit_suffix("-", "nan") == ""
|
|
assert format_unit_suffix(None, None) == ""
|
|
|
|
def test_non_string_input(self):
|
|
"""Non-string inputs are ignored."""
|
|
assert format_unit_suffix(123, 456) == ""
|
|
|
|
|
|
class TestExtractUnitScaleFromTitle:
|
|
"""Tests for extract_unit_scale_from_title."""
|
|
|
|
def test_empty(self):
|
|
"""Empty input yields (None, None)."""
|
|
assert extract_unit_scale_from_title("") == (None, None)
|
|
|
|
def test_paren_single_scale(self):
|
|
"""A single scale-only parenthetical produces only a scale."""
|
|
unit, scale = extract_unit_scale_from_title("GDP (Millions)")
|
|
assert unit is None
|
|
assert scale == "Millions"
|
|
|
|
def test_paren_single_unit_only(self):
|
|
"""A single non-scale-pattern paren returns it as unit if pattern matches."""
|
|
unit, scale = extract_unit_scale_from_title("GDP (US dollar, Percent)")
|
|
assert unit == "US dollar"
|
|
assert scale == "Percent"
|
|
|
|
def test_paren_single_non_pattern(self):
|
|
"""A single unit element without a known scale becomes unit."""
|
|
unit, scale = extract_unit_scale_from_title("GDP (Foo Percent)")
|
|
assert unit == "Foo Percent"
|
|
assert scale is None
|
|
|
|
def test_paren_two_parts(self):
|
|
"""A two-part parenthetical yields (first, second)."""
|
|
unit, scale = extract_unit_scale_from_title("GDP (US Dollar, Millions)")
|
|
assert unit == "US Dollar"
|
|
assert scale == "Millions"
|
|
|
|
def test_paren_with_no_scale_pattern_falls_through(self):
|
|
"""A paren without a known scale pattern falls through to comma logic."""
|
|
unit, scale = extract_unit_scale_from_title("Trade, US dollars")
|
|
assert unit == "US dollars"
|
|
assert scale is None
|
|
|
|
def test_no_match_returns_none(self):
|
|
"""A plain title yields (None, None)."""
|
|
assert extract_unit_scale_from_title("Population") == (None, None)
|
|
|
|
|
|
class TestIsBopSuffixOnly:
|
|
"""Tests for is_bop_suffix_only."""
|
|
|
|
def test_empty(self):
|
|
"""Empty string returns False."""
|
|
assert is_bop_suffix_only("") is False
|
|
|
|
def test_only_punctuation(self):
|
|
"""Punctuation-only normalizes to empty and returns True."""
|
|
assert is_bop_suffix_only(", :") is True
|
|
|
|
def test_lowercase_first_word(self):
|
|
"""A lowercase first word is treated as BOP suffix only."""
|
|
assert is_bop_suffix_only("net flows") is True
|
|
|
|
def test_bop_term(self):
|
|
"""Recognized BOP terms return True."""
|
|
assert is_bop_suffix_only("Credit") is True
|
|
assert is_bop_suffix_only("Assets (excl. reserves)") is True
|
|
|
|
def test_non_bop_term(self):
|
|
"""A normal title returns False."""
|
|
assert is_bop_suffix_only("Goods") is False
|
|
|
|
def test_bop_paren_stripping(self):
|
|
"""A trailing paren is stripped before bop check."""
|
|
assert is_bop_suffix_only("Assets (some context)") is True
|
|
|
|
|
|
class TestHierarchyContextSiblings:
|
|
"""Tests for HierarchyContext._get_true_siblings and sibling prefix logic."""
|
|
|
|
def test_sort_on_init(self):
|
|
"""Order list is sorted by order value."""
|
|
ctx = HierarchyContext(
|
|
[(3.0, "C", 1, False), (1.0, "A", 0, True), (2.0, "B", 1, False)]
|
|
)
|
|
assert [t[0] for t in ctx.order_title_level] == [1.0, 2.0, 3.0]
|
|
|
|
def test_missing_target_returns_empty(self):
|
|
"""If target order not found, returns []."""
|
|
ctx = HierarchyContext([(1.0, "A", 0, False)])
|
|
assert ctx._get_true_siblings(99.0, 0) == []
|
|
|
|
def test_siblings_consecutive(self):
|
|
"""Consecutive same-level rows are siblings."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Parent", 0, True),
|
|
(2.0, "Goods, Credit", 1, False),
|
|
(3.0, "Goods, Debit", 1, False),
|
|
(4.0, "Goods, Net", 1, False),
|
|
]
|
|
)
|
|
sibs = ctx._get_true_siblings(2.0, 1)
|
|
assert {t for _, t in sibs} == {"Goods, Credit", "Goods, Debit", "Goods, Net"}
|
|
|
|
def test_sibling_common_prefix_below_three(self):
|
|
"""Fewer than 3 siblings yields no common prefix."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Goods, Credit", 1, False),
|
|
(2.0, "Goods, Debit", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_sibling_common_prefix(1.0, "Goods, Credit") is None
|
|
|
|
def test_sibling_common_prefix_target_missing(self):
|
|
"""Target order not in list returns None."""
|
|
ctx = HierarchyContext([(1.0, "X", 0, False)])
|
|
assert ctx.find_sibling_common_prefix(99.0, "X") is None
|
|
|
|
def test_sibling_common_prefix_bop_endings_returns_none(self):
|
|
"""All-BOP-ending siblings skip prefix detection."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Goods, Credit", 1, False),
|
|
(2.0, "Goods, Debit", 1, False),
|
|
(3.0, "Goods, Net", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_sibling_common_prefix(1.0, "Goods, Credit") is None
|
|
|
|
def test_sibling_common_prefix_real_match(self):
|
|
"""A shared multi-segment prefix among non-BOP siblings is detected."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade, Goods, Exports", 1, False),
|
|
(2.0, "Trade, Goods, Imports", 1, False),
|
|
(3.0, "Trade, Goods, Balance", 1, False),
|
|
]
|
|
)
|
|
prefix = ctx.find_sibling_common_prefix(1.0, "Trade, Goods, Exports")
|
|
assert prefix == "Trade, Goods, "
|
|
|
|
def test_sibling_common_prefix_displayed_filter(self):
|
|
"""displayed_orders filter culls to < 3 siblings -> None."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade, A", 1, False),
|
|
(2.0, "Trade, B", 1, False),
|
|
(3.0, "Trade, C", 1, False),
|
|
]
|
|
)
|
|
assert (
|
|
ctx.find_sibling_common_prefix(1.0, "Trade, A", displayed_orders={1.0})
|
|
is None
|
|
)
|
|
|
|
def test_sibling_common_prefix_no_common(self):
|
|
"""If no common first segment, returns None."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "A, X", 1, False),
|
|
(2.0, "B, Y", 1, False),
|
|
(3.0, "C, Z", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_sibling_common_prefix(1.0, "A, X") is None
|
|
|
|
def test_sibling_common_prefix_target_does_not_match(self):
|
|
"""Target title that doesn't start with common prefix returns None."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade, A, X", 1, False),
|
|
(2.0, "Trade, A, Y", 1, False),
|
|
(3.0, "Trade, A, Z", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_sibling_common_prefix(1.0, "Different, A, X") is None
|
|
|
|
def test_sibling_common_prefix_empty_titles(self):
|
|
"""Empty sibling titles drop count below threshold."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "", 1, False),
|
|
(2.0, "", 1, False),
|
|
(3.0, "", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_sibling_common_prefix(1.0, "X") is None
|
|
|
|
|
|
class TestHierarchyContextBopGroup:
|
|
"""Tests for HierarchyContext.find_bop_group_prefix."""
|
|
|
|
def test_no_bop_suffix(self):
|
|
"""A title without a BOP suffix returns None."""
|
|
ctx = HierarchyContext([(1.0, "X", 0, False)])
|
|
assert ctx.find_bop_group_prefix(1.0, "Goods") is None
|
|
|
|
def test_net_suffix_skipped(self):
|
|
"""A ', Net' suffix never returns a group prefix."""
|
|
ctx = HierarchyContext([(1.0, "Goods, Net", 0, False)])
|
|
assert ctx.find_bop_group_prefix(1.0, "Goods, Net") is None
|
|
|
|
def test_target_missing(self):
|
|
"""Missing target order returns None."""
|
|
ctx = HierarchyContext([(1.0, "X", 0, False)])
|
|
assert ctx.find_bop_group_prefix(99.0, "Goods, Credit") is None
|
|
|
|
def test_no_matching_net(self):
|
|
"""Credit without a matching Net sibling returns None."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Goods, Credit", 0, False),
|
|
(2.0, "Goods, Debit", 0, False),
|
|
]
|
|
)
|
|
assert ctx.find_bop_group_prefix(1.0, "Goods, Credit") is None
|
|
|
|
def test_returns_prefix(self):
|
|
"""Credit with a matching Net sibling returns the base prefix."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Goods, Credit", 0, False),
|
|
(2.0, "Goods, Debit", 0, False),
|
|
(3.0, "Goods, Net", 0, False),
|
|
]
|
|
)
|
|
assert ctx.find_bop_group_prefix(1.0, "Goods, Credit") == "Goods, "
|
|
|
|
def test_displayed_filter(self):
|
|
"""displayed_orders filter removes matching Net."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Goods, Credit", 0, False),
|
|
(2.0, "Goods, Debit", 0, False),
|
|
(3.0, "Goods, Net", 0, False),
|
|
]
|
|
)
|
|
assert (
|
|
ctx.find_bop_group_prefix(1.0, "Goods, Credit", displayed_orders={1.0, 2.0})
|
|
is None
|
|
)
|
|
|
|
|
|
class TestHierarchyContextBestPrefix:
|
|
"""Tests for HierarchyContext.find_best_prefix."""
|
|
|
|
def test_missing_target(self):
|
|
"""Missing target order returns None."""
|
|
ctx = HierarchyContext([(1.0, "X", 0, True)])
|
|
assert ctx.find_best_prefix(99.0, "Anything") is None
|
|
|
|
def test_no_ancestors(self):
|
|
"""No ancestor rows -> None."""
|
|
ctx = HierarchyContext([(1.0, "Goods, Trade", 0, False)])
|
|
assert ctx.find_best_prefix(1.0, "Goods, Trade") is None
|
|
|
|
def test_direct_prefix(self):
|
|
"""An ancestor title that is a prefix is returned."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade", 0, True),
|
|
(2.0, "Trade, Goods", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_prefix(2.0, "Trade, Goods") == "Trade"
|
|
|
|
def test_displayed_filter_excludes_ancestor(self):
|
|
"""Ancestor not in displayed_orders is skipped."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade", 0, True),
|
|
(2.0, "Trade, Goods", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_prefix(2.0, "Trade, Goods", displayed_orders=set()) is None
|
|
|
|
def test_empty_ancestor_skipped(self):
|
|
"""Empty ancestor titles are skipped."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "", 0, True),
|
|
(2.0, "Trade, Goods", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_prefix(2.0, "Trade, Goods") is None
|
|
|
|
def test_phrase_match_with_separator(self):
|
|
"""An ancestor key phrase can match via ': ' separator."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade Survey", 0, True),
|
|
(2.0, "Trade: Detail", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_best_prefix(2.0, "Trade: Detail")
|
|
assert result == "Trade: "
|
|
|
|
def test_bop_only_remainder_returns_none(self):
|
|
"""If remainder after stripping is BOP-only, return None."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade", 0, True),
|
|
(2.0, "Trade, Credit", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_prefix(2.0, "Trade, Credit") is None
|
|
|
|
def test_picks_longest_prefix(self):
|
|
"""When multiple ancestors match, the longest wins."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade", 0, True),
|
|
(2.0, "Trade Goods", 1, True),
|
|
(3.0, "Trade Goods, Exports", 2, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_prefix(3.0, "Trade Goods, Exports") == "Trade Goods"
|
|
|
|
def test_skip_repeat_levels(self):
|
|
"""Once a level is seen, deeper rows at that level are skipped."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Foo", 0, True),
|
|
(2.0, "Bar", 0, True),
|
|
(3.0, "Bar, X", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_best_prefix(3.0, "Bar, X")
|
|
assert result == "Bar"
|
|
|
|
|
|
class TestHierarchyContextBestSuffix:
|
|
"""Tests for HierarchyContext.find_best_suffix."""
|
|
|
|
def test_missing_target(self):
|
|
"""Missing target order returns None."""
|
|
ctx = HierarchyContext([(1.0, "X", 0, True)])
|
|
assert ctx.find_best_suffix(99.0, "X, Y") is None
|
|
|
|
def test_no_ancestor_parts(self):
|
|
"""No ancestors leaves nothing to match -> None."""
|
|
ctx = HierarchyContext([(1.0, "Goods, Trade", 0, False)])
|
|
assert ctx.find_best_suffix(1.0, "Goods, Trade") is None
|
|
|
|
def test_displayed_filter(self):
|
|
"""Ancestor filtered out by displayed_orders is ignored."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Special", 0, True),
|
|
(2.0, "X, Special", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_suffix(2.0, "X, Special", displayed_orders=set()) is None
|
|
|
|
def test_empty_ancestor_skipped(self):
|
|
"""Empty ancestor titles are skipped."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "", 0, True),
|
|
(2.0, "X, Y", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_suffix(2.0, "X, Y") is None
|
|
|
|
def test_protected_suffix_not_stripped(self):
|
|
"""Protected BOP suffixes like 'Assets' are not removed."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Assets", 0, True),
|
|
(2.0, "X, Assets", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_suffix(2.0, "X, Assets") is None
|
|
|
|
def test_returns_matching_suffix(self):
|
|
"""A regular ancestor word that appears as suffix is returned."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Special", 0, True),
|
|
(2.0, "X, Special", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_suffix(2.0, "X, Special") == ", Special"
|
|
|
|
def test_debtors_creditors_expansion(self):
|
|
"""Debtors/Creditors ancestors are detected as suffix ancestors."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Debtors group", 0, True),
|
|
(2.0, "Foo, Debtors group", 1, False),
|
|
]
|
|
)
|
|
ctx2 = HierarchyContext(
|
|
[
|
|
(1.0, "Creditors group", 0, True),
|
|
(2.0, "X, Creditors group", 1, False),
|
|
]
|
|
)
|
|
ctx3 = HierarchyContext(
|
|
[
|
|
(1.0, "Financial assets", 0, True),
|
|
(2.0, "X, Financial assets", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_best_suffix(2.0, "Foo, Debtors group") == ", Debtors group"
|
|
assert ctx2.find_best_suffix(2.0, "X, Creditors group") == ", Creditors group"
|
|
assert ctx3.find_best_suffix(2.0, "X, Financial assets") == ", Financial assets"
|
|
|
|
def test_liabilities_expansion(self):
|
|
"""A 'Liabilities' ancestor adds 'Net incurrence of liabilities' and 'Total liabilities'."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Liabilities", 0, True),
|
|
(2.0, "X, Total liabilities", 1, False),
|
|
]
|
|
)
|
|
assert (
|
|
ctx.find_best_suffix(2.0, "X, Total liabilities") == ", Total liabilities"
|
|
)
|
|
|
|
|
|
class TestHierarchyContextAncestorPartPrefix:
|
|
"""Tests for HierarchyContext.find_ancestor_part_prefix."""
|
|
|
|
def test_missing_target(self):
|
|
"""Missing target order returns None."""
|
|
ctx = HierarchyContext([(1.0, "X", 0, True)])
|
|
assert ctx.find_ancestor_part_prefix(99.0, "X, Y") is None
|
|
|
|
def test_no_ancestor_parts(self):
|
|
"""No ancestors -> None."""
|
|
ctx = HierarchyContext([(1.0, "Goods, Trade", 0, False)])
|
|
assert ctx.find_ancestor_part_prefix(1.0, "Goods, Trade") is None
|
|
|
|
def test_single_word_skipped(self):
|
|
"""A single-word ancestor that's not strippable is ignored."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Foo", 0, True),
|
|
(2.0, "Foo, X", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_ancestor_part_prefix(2.0, "Foo, X") is None
|
|
|
|
def test_assets_strippable(self):
|
|
"""The single word 'Assets' is in the strippable set."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Assets", 0, True),
|
|
(2.0, "Assets, X", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_ancestor_part_prefix(2.0, "Assets, X") == "Assets, "
|
|
|
|
def test_multi_word_ancestor_prefix(self):
|
|
"""Multi-word ancestors strip a matching prefix."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Financial corporations", 0, True),
|
|
(2.0, "Financial corporations, X", 1, False),
|
|
]
|
|
)
|
|
assert (
|
|
ctx.find_ancestor_part_prefix(2.0, "Financial corporations, X")
|
|
== "Financial corporations, "
|
|
)
|
|
|
|
def test_displayed_filter(self):
|
|
"""Ancestor filtered by displayed_orders is ignored."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Financial corporations", 0, True),
|
|
(2.0, "Financial corporations, X", 1, False),
|
|
]
|
|
)
|
|
assert (
|
|
ctx.find_ancestor_part_prefix(
|
|
2.0, "Financial corporations, X", displayed_orders=set()
|
|
)
|
|
is None
|
|
)
|
|
|
|
def test_bop_remainder_skipped(self):
|
|
"""A remainder that's BOP-only is skipped (loop continues)."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Financial corporations", 0, True),
|
|
(2.0, "Financial corporations, Net", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_ancestor_part_prefix(2.0, "Financial corporations, Net") is None
|
|
|
|
def test_self_skip(self):
|
|
"""Ancestors equal to target title are skipped."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Foo bar", 0, True),
|
|
(2.0, "Foo bar", 1, False),
|
|
]
|
|
)
|
|
assert ctx.find_ancestor_part_prefix(2.0, "Foo bar") is None
|
|
|
|
def test_child_prefix_long_form(self):
|
|
"""A long comma-prefix matching an ancestor word triggers fallback."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "International investment position", 0, True),
|
|
(2.0, "International investment, Detail", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_ancestor_part_prefix(2.0, "International investment, Detail")
|
|
assert result == "International investment, "
|
|
|
|
|
|
class TestHierarchyContextSimplifyTitle:
|
|
"""Tests for HierarchyContext.simplify_title."""
|
|
|
|
def test_strip_suffix_only(self):
|
|
"""A bare title with a unit suffix is stripped."""
|
|
ctx = HierarchyContext([(1.0, "Plain (Millions)", 0, False)])
|
|
assert ctx.simplify_title(1.0, "Plain (Millions)") == "Plain"
|
|
|
|
def test_prefix_then_suffix(self):
|
|
"""Prefix from ancestor is removed."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade", 0, True),
|
|
(2.0, "Trade, Goods", 1, False),
|
|
]
|
|
)
|
|
assert ctx.simplify_title(2.0, "Trade, Goods") == "Goods"
|
|
|
|
def test_bop_only_restored(self):
|
|
"""If stripping leaves a BOP-only term, original is restored."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade", 0, True),
|
|
(2.0, "Trade, Net", 1, False),
|
|
(3.0, "Trade, Credit", 1, False),
|
|
]
|
|
)
|
|
out = ctx.simplify_title(2.0, "Trade, Net")
|
|
assert out == "Trade, Net"
|
|
|
|
def test_empty_restored(self):
|
|
"""If stripping yields empty, original is restored."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Foo", 0, True),
|
|
(2.0, "Foo", 1, False),
|
|
]
|
|
)
|
|
out = ctx.simplify_title(2.0, "Foo")
|
|
assert out == "Foo"
|
|
|
|
def test_bop_group_stripped_keeps_short(self):
|
|
"""BOP group stripping keeps a short like 'Credit'."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Goods, Credit", 0, False),
|
|
(2.0, "Goods, Debit", 0, False),
|
|
(3.0, "Goods, Net", 0, False),
|
|
]
|
|
)
|
|
out = ctx.simplify_title(1.0, "Goods, Credit")
|
|
assert out == "Credit"
|
|
|
|
def test_sibling_prefix_strip(self):
|
|
"""A common multi-segment sibling prefix is stripped."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Trade, Goods, Exports", 1, False),
|
|
(2.0, "Trade, Goods, Imports", 1, False),
|
|
(3.0, "Trade, Goods, Balance", 1, False),
|
|
]
|
|
)
|
|
out = ctx.simplify_title(1.0, "Trade, Goods, Exports")
|
|
assert out == "Exports"
|
|
|
|
def test_best_suffix_strip(self):
|
|
"""An ancestor suffix that appears is removed."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Special", 0, True),
|
|
(2.0, "X, Special", 1, False),
|
|
]
|
|
)
|
|
out = ctx.simplify_title(2.0, "X, Special")
|
|
assert out == "X"
|
|
|
|
|
|
class TestBuildOrderTitleLevel:
|
|
"""Tests for build_order_title_level."""
|
|
|
|
def test_header_row_picked(self):
|
|
"""The header row with longest title is used."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"title": "Short",
|
|
"level": 0,
|
|
"is_category_header": True,
|
|
},
|
|
{
|
|
"order": 1.0,
|
|
"title": "Longer header title",
|
|
"level": 0,
|
|
"is_category_header": True,
|
|
},
|
|
]
|
|
)
|
|
out = build_order_title_level(df)
|
|
assert out == [(1.0, "Longer header title", 0, True)]
|
|
|
|
def test_data_row_when_no_header(self):
|
|
"""Without a header row, the longest data row is used."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"order": 2.0,
|
|
"title": "A",
|
|
"level": 1,
|
|
"is_category_header": False,
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"title": "AAAA",
|
|
"level": 1,
|
|
"is_category_header": False,
|
|
},
|
|
]
|
|
)
|
|
out = build_order_title_level(df)
|
|
assert out == [(2.0, "AAAA", 1, False)]
|
|
|
|
def test_drops_na_orders(self):
|
|
"""Rows with NaN order are dropped."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"order": None,
|
|
"title": "X",
|
|
"level": 0,
|
|
"is_category_header": False,
|
|
},
|
|
{
|
|
"order": 1.0,
|
|
"title": "Y",
|
|
"level": 0,
|
|
"is_category_header": False,
|
|
},
|
|
]
|
|
)
|
|
out = build_order_title_level(df)
|
|
assert out == [(1.0, "Y", 0, False)]
|
|
|
|
def test_sorted(self):
|
|
"""Results are sorted by order ascending."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"order": 3.0,
|
|
"title": "C",
|
|
"level": 0,
|
|
"is_category_header": False,
|
|
},
|
|
{
|
|
"order": 1.0,
|
|
"title": "A",
|
|
"level": 0,
|
|
"is_category_header": False,
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"title": "B",
|
|
"level": 0,
|
|
"is_category_header": False,
|
|
},
|
|
]
|
|
)
|
|
out = build_order_title_level(df)
|
|
assert [t[0] for t in out] == [1.0, 2.0, 3.0]
|
|
|
|
|
|
class TestCheckMissingCountryData:
|
|
"""Tests for check_missing_country_data."""
|
|
|
|
def test_no_warning_when_data_present(self):
|
|
"""Country with data in selected dates does not warn."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{"country": "USA", "country_code": "USA", "date": "2020-01-01"},
|
|
]
|
|
)
|
|
check_missing_country_data(df, ["USA"], ["2020-01-01"], ["USA"])
|
|
|
|
def test_warns_when_missing(self):
|
|
"""Country missing data in selected dates produces a warning."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{"country": "USA", "country_code": "USA", "date": "2019-01-01"},
|
|
{"country": "GBR", "country_code": "GBR", "date": "2020-01-01"},
|
|
]
|
|
)
|
|
with pytest.warns(OpenBBWarning, match="No data for 'USA'"):
|
|
check_missing_country_data(df, ["USA"], ["2020-01-01"], ["USA", "GBR"])
|
|
|
|
def test_match_by_country_code(self):
|
|
"""A requested code matches via country_code column."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"country": "United States",
|
|
"country_code": "USA",
|
|
"date": "2019-01-01",
|
|
}
|
|
]
|
|
)
|
|
with pytest.warns(OpenBBWarning):
|
|
check_missing_country_data(df, ["USA"], ["2020-01-01"], ["United States"])
|
|
|
|
|
|
class TestPivotIndicatorMode:
|
|
"""Tests for pivot_indicator_mode."""
|
|
|
|
def test_basic_pivot(self):
|
|
"""A basic dataset pivots into MultiIndex DataFrame."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": "GDP",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 100.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
},
|
|
{
|
|
"title": "GDP",
|
|
"country": "USA",
|
|
"date": "2021-01-01",
|
|
"value": 110.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
},
|
|
]
|
|
)
|
|
result = pivot_indicator_mode(df, ["2020-01-01", "2021-01-01"], ["USA"])
|
|
assert not result.empty
|
|
assert ("GDP", "USA", "US Dollar", "Millions") in result.index
|
|
|
|
def test_skips_nan_title(self):
|
|
"""Rows with NaN title are skipped."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": None,
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 5.0,
|
|
"unit": "US",
|
|
"scale": "M",
|
|
}
|
|
]
|
|
)
|
|
result = pivot_indicator_mode(df, ["2020-01-01"], ["USA"])
|
|
assert result.empty
|
|
|
|
def test_country_with_no_data_skipped(self):
|
|
"""Country with no rows for a title is skipped."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": "GDP",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 1.0,
|
|
"unit": "U",
|
|
"scale": "S",
|
|
}
|
|
]
|
|
)
|
|
result = pivot_indicator_mode(df, ["2020-01-01"], ["USA", "GBR"])
|
|
assert ("GDP", "USA", "U", "S") in result.index
|
|
assert ("GDP", "GBR", "U", "S") not in result.index
|
|
|
|
def test_dash_unit_treated_as_none_then_parsed(self):
|
|
"""'-' unit/scale gets replaced via title parsing."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": "GDP (US Dollar, Millions)",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 1.0,
|
|
"unit": "-",
|
|
"scale": "-",
|
|
}
|
|
]
|
|
)
|
|
result = pivot_indicator_mode(df, ["2020-01-01"], ["USA"])
|
|
assert (
|
|
"GDP (US Dollar, Millions)",
|
|
"USA",
|
|
"US Dollar",
|
|
"Millions",
|
|
) in result.index
|
|
|
|
def test_all_zero_dropped(self):
|
|
"""A series whose only values are zero is omitted."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": "GDP",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 0.0,
|
|
"unit": "U",
|
|
"scale": "S",
|
|
}
|
|
]
|
|
)
|
|
result = pivot_indicator_mode(df, ["2020-01-01"], ["USA"])
|
|
assert result.empty
|
|
|
|
|
|
class TestPivotTableMode:
|
|
"""Tests for pivot_table_mode."""
|
|
|
|
@staticmethod
|
|
def _base_df(rows: list[dict[str, Any]]) -> pd.DataFrame:
|
|
"""Build a DataFrame from row dicts with required columns defaulted."""
|
|
defaults = {
|
|
"title": "",
|
|
"level": 0,
|
|
"order": 0.0,
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": float("nan"),
|
|
"unit": "",
|
|
"scale": "",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": None,
|
|
"parent_id": None,
|
|
"indicator_code": None,
|
|
"symbol": "S",
|
|
}
|
|
out = []
|
|
for r in rows:
|
|
d = dict(defaults)
|
|
d.update(r)
|
|
out.append(d)
|
|
return pd.DataFrame(out)
|
|
|
|
def test_simple_hierarchy(self):
|
|
"""A simple two-row hierarchy pivots into rows with indented titles."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
"parent_id": None,
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "TRD_G",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade Table", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert not result.empty
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("Trade" in t for t in titles)
|
|
|
|
def test_no_data_returns_empty_df(self):
|
|
"""If no rows have any data, result is an empty DataFrame."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade Table", "dataflow_id": "BOP"}}
|
|
try:
|
|
pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
except KeyError:
|
|
pass
|
|
|
|
def test_inherited_unit_scale_from_parent(self):
|
|
"""A child without unit/scale inherits from its parent's order."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"indicator_code": "TRD_G",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade Table", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert not result.empty
|
|
|
|
def test_isora_filters_non_topic_headers(self):
|
|
"""For ISORA dataflows, non-topic headers are skipped."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Some random header",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "TRD_G",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "ISORA Survey", "dataflow_id": "ISORA"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_isora_topic_header_kept(self):
|
|
"""For ISORA, topic-style headers (numbered) are kept."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "1. Some topic",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "1. Some topic, Detail",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 7.0,
|
|
"unit": "Percent",
|
|
"scale": "Units",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "ISORA Survey", "dataflow_id": "ISORA"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert not result.empty
|
|
|
|
def test_first_level0_data_adds_table_header(self):
|
|
"""If the first level-0 row is data (not a header), a table header row is added."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Direct Data",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n1",
|
|
"value": 1.5,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "DD",
|
|
},
|
|
]
|
|
)
|
|
meta = {
|
|
"table": {
|
|
"hierarchy_name": "Trade Statistics",
|
|
"dataflow_id": "BOP",
|
|
}
|
|
}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("TRADE STATISTICS" in t for t in titles)
|
|
|
|
def test_dimension_grouping_with_counterpart_country(self):
|
|
"""Counterpart country grouping renders ▸ group / country rows."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 10.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "G001",
|
|
"counterpart_country": "World",
|
|
"indicator_code": "B",
|
|
"series_id": "s1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "GBR",
|
|
"counterpart_country": "United Kingdom",
|
|
"indicator_code": "B",
|
|
"series_id": "s2",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 3.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "G002",
|
|
"counterpart_country": "Europe",
|
|
"indicator_code": "B",
|
|
"series_id": "s3",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert not result.empty
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("World" in t or "Bilateral" in t for t in titles)
|
|
|
|
def test_sector_multi_dim(self):
|
|
"""Multiple SECTOR codes group under a header with sector indents."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Indicator",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Detail",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S1",
|
|
"sector": "Households",
|
|
"indicator_code": "I1",
|
|
"series_id": "p1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Detail",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S2",
|
|
"sector": "Corporations",
|
|
"indicator_code": "I1",
|
|
"series_id": "p2",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Indicator", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert not result.empty
|
|
|
|
def test_bop_credit_debit_net_grouping(self):
|
|
"""BOP Credit/Debit/Net sibling grouping is exercised."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Goods, Credit",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "G_C",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 1,
|
|
"title": "Goods, Debit",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "G_D",
|
|
},
|
|
{
|
|
"order": 4.0,
|
|
"level": 1,
|
|
"title": "Goods, Net",
|
|
"hierarchy_node_id": "n4",
|
|
"parent_id": "n1",
|
|
"value": 3.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "G_N",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert not result.empty
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("Credit" in t or "Debit" in t or "Net" in t for t in titles)
|
|
|
|
|
|
class TestPivotTableData:
|
|
"""Tests for pivot_table_data dispatcher."""
|
|
|
|
def test_no_hierarchy_indicator_mode(self):
|
|
"""When no 'order' column exists, indicator mode is used."""
|
|
records = [
|
|
{
|
|
"title": "GDP",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 100.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
}
|
|
]
|
|
result = pivot_table_data(records, country="USA", limit=None, metadata={})
|
|
assert isinstance(result, pd.DataFrame)
|
|
assert ("GDP", "USA", "US Dollar", "Millions") in result.index
|
|
|
|
def test_with_hierarchy_table_mode(self):
|
|
"""When 'order' column has values, table mode is used."""
|
|
records = [
|
|
{
|
|
"title": "Trade",
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 0.0,
|
|
"level": 0,
|
|
"order": 1.0,
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
"parent_id": None,
|
|
"indicator_code": None,
|
|
"unit": "",
|
|
"scale": "",
|
|
"symbol": "S",
|
|
},
|
|
{
|
|
"title": "Trade, Goods",
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 5.0,
|
|
"level": 1,
|
|
"order": 2.0,
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"indicator_code": "TRD_G",
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"symbol": "S",
|
|
},
|
|
]
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_data(records, country="USA", limit=1, metadata=meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_limit_zero_returns_all_dates(self):
|
|
"""limit=0 falls back to all dates."""
|
|
records = [
|
|
{
|
|
"title": "GDP",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 100.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
}
|
|
]
|
|
result = pivot_table_data(records, country=None, limit=0, metadata={})
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
|
|
class TestPivotTableModeExtras:
|
|
"""Tests targeting specific branches in pivot_table_mode."""
|
|
|
|
@staticmethod
|
|
def _base_df(rows: list[dict[str, Any]]) -> pd.DataFrame:
|
|
"""Same defaults as TestPivotTableMode._base_df."""
|
|
defaults = {
|
|
"title": "",
|
|
"level": 0,
|
|
"order": 0.0,
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": float("nan"),
|
|
"unit": "",
|
|
"scale": "",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": None,
|
|
"parent_id": None,
|
|
"indicator_code": None,
|
|
"symbol": "S",
|
|
}
|
|
out = []
|
|
for r in rows:
|
|
d = dict(defaults)
|
|
d.update(r)
|
|
out.append(d)
|
|
return pd.DataFrame(out)
|
|
|
|
def test_parent_lookup_via_suffix_pattern(self):
|
|
"""Parent_id matched via '___pid' suffix lookup branch."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Parent",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "PFX___N1",
|
|
"parent_id": None,
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Child",
|
|
"hierarchy_node_id": "PFX___N2",
|
|
"parent_id": "N1", # No direct match; suffix lookup matches PFX___N1
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "C",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Parent", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_indicator_code_fallback_title(self):
|
|
"""When title is empty, indicator_code drives the raw_title path."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Parent",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "gross_domestic_product",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Parent", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_parent_unit_scale_inheritance_via_traversal(self):
|
|
"""A grandchild inherits unit/scale from grandparent via traversal."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Root",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "r",
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Mid",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "m",
|
|
"parent_id": "r",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Leaf",
|
|
"hierarchy_node_id": "leaf",
|
|
"parent_id": "m",
|
|
"value": 7.0,
|
|
"indicator_code": "L",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Root", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_first_level0_data_matches_hierarchy_name(self):
|
|
"""If first level-0 title matches hierarchy_name, no extra table header is added."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n1",
|
|
"value": 3.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "TR",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert not any(t.strip() == "TRADE" for t in titles)
|
|
|
|
def test_bop_suffix_only_header_skipped(self):
|
|
"""Headers whose title is a BOP-only suffix are skipped."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Real Header",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Credit",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Real Header, Goods",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n2",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Real Header", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert not any(t.strip().endswith("Credit") and "▸" in t for t in titles)
|
|
|
|
def test_financial_corporations_prefix_strip(self):
|
|
"""Header titles starting with 'Financial corporations, ' have that prefix stripped."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Financial corporations, Survey",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Financial corporations, Survey, Detail",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {
|
|
"table": {"hierarchy_name": "Financial corporations", "dataflow_id": "BOP"}
|
|
}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("Survey" in t for t in titles)
|
|
|
|
def test_depository_corporations_prefix_strip(self):
|
|
"""Header titles starting with 'Depository corporations, ' have prefix stripped."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Depository corporations, Items",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Depository corporations, Items, Specific",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {
|
|
"table": {"hierarchy_name": "Depository corporations", "dataflow_id": "BOP"}
|
|
}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("Items" in t for t in titles)
|
|
|
|
def test_paren_unit_suffix_stripped_in_pivot(self):
|
|
"""A trailing parenthetical unit suffix is captured separately."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Indicator (US Dollar)",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Detail, Transactions",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "D",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Indicator", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_bop_credit_promotion_to_ancestor_title(self):
|
|
"""A child Credit row is rewritten to use its ancestor title as base."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Goods Total",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "h0",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Goods Total, Credit",
|
|
"hierarchy_node_id": "h1",
|
|
"parent_id": "h0",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Goods Total", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_two_dates_with_limit(self):
|
|
"""Two dates with limit pivots correctly through pivot_table_data."""
|
|
records = [
|
|
{
|
|
"title": "Trade",
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 0.0,
|
|
"level": 0,
|
|
"order": 1.0,
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
"parent_id": None,
|
|
"indicator_code": None,
|
|
"unit": "",
|
|
"scale": "",
|
|
"symbol": "S",
|
|
},
|
|
{
|
|
"title": "Trade, Goods",
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 5.0,
|
|
"level": 1,
|
|
"order": 2.0,
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"indicator_code": "TRD_G",
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"symbol": "S",
|
|
},
|
|
{
|
|
"title": "Trade, Goods",
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2021-01-01",
|
|
"value": 6.0,
|
|
"level": 1,
|
|
"order": 2.0,
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"indicator_code": "TRD_G",
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"symbol": "S",
|
|
},
|
|
]
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_data(records, country="USA", limit=2, metadata=meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_counterpart_country_individual_only(self):
|
|
"""All counterpart countries are individual (no group code)."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "GBR",
|
|
"counterpart_country": "United Kingdom",
|
|
"indicator_code": "B",
|
|
"series_id": "s1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 3.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "DEU",
|
|
"counterpart_country": "Germany",
|
|
"indicator_code": "B",
|
|
"series_id": "s2",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("United Kingdom" in t or "Germany" in t for t in titles)
|
|
|
|
def test_counterpart_country_all_zero_skipped(self):
|
|
"""Zero-only counterpart country values are skipped."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 0.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "G001",
|
|
"counterpart_country": "World",
|
|
"indicator_code": "B",
|
|
"series_id": "s1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 0.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "G002",
|
|
"counterpart_country": "Europe",
|
|
"indicator_code": "B",
|
|
"series_id": "s2",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 0.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "GBR",
|
|
"counterpart_country": "United Kingdom",
|
|
"indicator_code": "B",
|
|
"series_id": "s3",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert all("Europe" not in t for t in titles)
|
|
|
|
def test_no_symbol_column_fallback(self):
|
|
"""When no dim_code columns and no symbol, the loop just skips."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": "GDP",
|
|
"level": 0,
|
|
"order": 1.0,
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 5.0,
|
|
"unit": "",
|
|
"scale": "",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n1",
|
|
"parent_id": None,
|
|
"indicator_code": "GDP",
|
|
"symbol": "S",
|
|
}
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "GDP", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_isora_topic_indicator(self):
|
|
"""ISORA dataflow with 'INDICATORS BY TOPIC' in title keeps the header."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "INDICATORS BY TOPIC",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Detail value",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 7.0,
|
|
"unit": "Percent",
|
|
"scale": "Units",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {
|
|
"table": {
|
|
"hierarchy_name": "INDICATORS BY TOPIC",
|
|
"dataflow_id": "OTHER",
|
|
}
|
|
}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("INDICATORS BY TOPIC" in t for t in titles)
|
|
|
|
def test_dim_value_in_title_only_uses_iterrows_path(self):
|
|
"""Test that title without unit/scale parses from title for an order with NaN unit."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Population (Millions)",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n1",
|
|
"value": 100.0,
|
|
"indicator_code": "POP",
|
|
"unit": "",
|
|
"scale": "",
|
|
}
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Population", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
|
|
class TestPivotIndicatorModeExtras:
|
|
"""Extra tests for pivot_indicator_mode branches."""
|
|
|
|
def test_no_unit_no_scale_parsed_from_title(self):
|
|
"""When unit/scale are missing, they get parsed from title."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": "GDP (US Dollar, Millions)",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 1.0,
|
|
"unit": None,
|
|
"scale": None,
|
|
}
|
|
]
|
|
)
|
|
result = pivot_indicator_mode(df, ["2020-01-01"], ["USA"])
|
|
assert not result.empty
|
|
|
|
def test_missing_date_value(self):
|
|
"""A date with no value is stored as None."""
|
|
df = pd.DataFrame(
|
|
[
|
|
{
|
|
"title": "GDP",
|
|
"country": "USA",
|
|
"date": "2020-01-01",
|
|
"value": 5.0,
|
|
"unit": "U",
|
|
"scale": "S",
|
|
},
|
|
]
|
|
)
|
|
result = pivot_indicator_mode(df, ["2020-01-01", "2021-01-01"], ["USA"])
|
|
assert not result.empty
|
|
|
|
|
|
class TestExtractUnitFromLabelExtra:
|
|
"""Additional tests for extract_unit_from_label edge cases."""
|
|
|
|
def test_paren_with_no_leading_space_at_zero(self):
|
|
"""A label that is exactly '(X)' has paren_start == 0 and returns None via the path."""
|
|
assert extract_unit_from_label("(Percent)") is None
|
|
|
|
def test_comma_split_no_second_part(self):
|
|
"""No comma at all means rsplit returns one part."""
|
|
assert extract_unit_from_label("Population") is None
|
|
|
|
|
|
class TestPivotTableModeMoreBranches:
|
|
"""More branch tests for pivot_table_mode."""
|
|
|
|
@staticmethod
|
|
def _base_df(rows: list[dict[str, Any]]) -> pd.DataFrame:
|
|
"""Default row builder with NaN value default."""
|
|
defaults = {
|
|
"title": "",
|
|
"level": 0,
|
|
"order": 0.0,
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": float("nan"),
|
|
"unit": "",
|
|
"scale": "",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": None,
|
|
"parent_id": None,
|
|
"indicator_code": None,
|
|
"symbol": "S",
|
|
}
|
|
out = []
|
|
for r in rows:
|
|
d = dict(defaults)
|
|
d.update(r)
|
|
out.append(d)
|
|
return pd.DataFrame(out)
|
|
|
|
def test_header_with_own_data_and_dim_groups(self):
|
|
"""A parent order that has own data AND multiple sector dim groups."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Indicator",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
"value": 100.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S1",
|
|
"sector": "Households",
|
|
"indicator_code": "X",
|
|
"series_id": "a",
|
|
},
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Indicator",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
"value": 200.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S2",
|
|
"sector": "Firms",
|
|
"indicator_code": "X",
|
|
"series_id": "b",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Indicator, Sub",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 50.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S1",
|
|
"sector": "Households",
|
|
"indicator_code": "Y",
|
|
"series_id": "c",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Indicator, Sub",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 60.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S2",
|
|
"sector": "Firms",
|
|
"indicator_code": "Y",
|
|
"series_id": "d",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Indicator", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_multi_dim_non_uniform_unit_scale(self):
|
|
"""A multi-dim order with mixed units/scales pulls suffix from first dr."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Parent",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "Percent",
|
|
"scale": "Units",
|
|
"sector_code": "S1",
|
|
"sector": "A",
|
|
"indicator_code": "X",
|
|
"series_id": "p1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S2",
|
|
"sector": "B",
|
|
"indicator_code": "X",
|
|
"series_id": "p2",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Parent", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_isora_underscored_promoted_title_continues(self):
|
|
"""An ISORA title with '___' is skipped."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "1. INDICATORS BY TOPIC",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "internal___thing",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Detail",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n2",
|
|
"value": 7.0,
|
|
"unit": "Percent",
|
|
"scale": "Units",
|
|
"indicator_code": "Y",
|
|
},
|
|
]
|
|
)
|
|
meta = {
|
|
"table": {
|
|
"hierarchy_name": "1. INDICATORS BY TOPIC",
|
|
"dataflow_id": "ISORA",
|
|
}
|
|
}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_bop_header_skipped_via_skipped_parent(self):
|
|
"""A BOP-only header gets tracked as skipped parent."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Credit", # BOP-only header
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"indicator_code": "CRD",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Trade, Credit, Goods",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n2",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_bop_credit_promotion_with_meaningful_ancestor(self):
|
|
"""A child Credit row picks up the ancestor header that starts with its base."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Goods Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "h0",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Goods, Credit",
|
|
"hierarchy_node_id": "h1",
|
|
"parent_id": "h0",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Goods Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_counterpart_country_individual_zero_skipped(self):
|
|
"""Individual counterpart with all-zero values is skipped."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "GBR",
|
|
"counterpart_country": "United Kingdom",
|
|
"indicator_code": "B",
|
|
"series_id": "s1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 0.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "DEU",
|
|
"counterpart_country": "Germany",
|
|
"indicator_code": "B",
|
|
"series_id": "s2",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
titles = [idx[0] for idx in result.index]
|
|
assert any("United Kingdom" in t for t in titles)
|
|
|
|
def test_bop_skipped_parent_with_underscores(self):
|
|
"""Skipped BOP parent with '___' in node_id triggers the suffix-split branch."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "prefix___n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Credit",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "PFX___N2",
|
|
"parent_id": "prefix___n1",
|
|
"indicator_code": None,
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Trade, Credit, Goods",
|
|
"hierarchy_node_id": "X___N3",
|
|
"parent_id": "PFX___N2",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_bop_skipped_parent_only_indicator_code(self):
|
|
"""A skipped header with only indicator_code (no node id) is tracked."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Credit",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": None,
|
|
"parent_id": "n1",
|
|
"indicator_code": "CRED_IND",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Trade, Credit, Goods",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "CRED_IND",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_get_sort_value_non_numeric(self):
|
|
"""get_sort_value catches non-numeric values via try/except."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": "not-a-number",
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "GBR",
|
|
"counterpart_country": "United Kingdom",
|
|
"indicator_code": "B",
|
|
"series_id": "s1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Bilateral",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": "also-not-numeric",
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"counterpart_country_code": "G001",
|
|
"counterpart_country": "World",
|
|
"indicator_code": "B",
|
|
"series_id": "s2",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
|
|
class TestHierarchyContextMore:
|
|
"""Targeted branch coverage for HierarchyContext methods."""
|
|
|
|
def test_break_on_lower_level_sibling_lookup(self):
|
|
"""_get_true_siblings breaks once a lower-level row is encountered after target."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "ParentA", 0, True),
|
|
(2.0, "A1", 1, False),
|
|
(3.0, "A2", 1, False),
|
|
(4.0, "ParentB", 0, True),
|
|
(5.0, "B1", 1, False),
|
|
]
|
|
)
|
|
sibs = ctx._get_true_siblings(2.0, 1)
|
|
titles = {t for _, t in sibs}
|
|
assert titles == {"A1", "A2"}
|
|
|
|
def test_best_suffix_repeats_level_skipped(self):
|
|
"""A second ancestor at the same level is skipped (line 697)."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Older Special", 0, True),
|
|
(2.0, "Closer Special", 0, True),
|
|
(3.0, "X, Closer Special", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_best_suffix(3.0, "X, Closer Special")
|
|
assert result == ", Closer Special"
|
|
|
|
def test_best_suffix_net_acq_branch(self):
|
|
"""'Net acquisition of financial assets' ancestor is preserved."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Net acquisition of financial assets", 0, True),
|
|
(2.0, "X, Net acquisition of financial assets", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_best_suffix(2.0, "X, Net acquisition of financial assets")
|
|
assert result == ", Net acquisition of financial assets"
|
|
|
|
def test_ancestor_part_prefix_repeats_level_skipped(self):
|
|
"""Repeat ancestor levels are skipped via levels_seen (line 778)."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Older corporations", 0, True),
|
|
(2.0, "Financial corporations", 0, True),
|
|
(3.0, "Financial corporations, X", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_ancestor_part_prefix(3.0, "Financial corporations, X")
|
|
assert result == "Financial corporations, "
|
|
|
|
def test_ancestor_part_prefix_liabilities_expansion(self):
|
|
"""Liabilities ancestor expansion is exercised (lines 797-799)."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Liabilities", 0, True),
|
|
(2.0, "Total liabilities, X", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_ancestor_part_prefix(2.0, "Total liabilities, X")
|
|
assert result == "Total liabilities, "
|
|
|
|
def test_ancestor_part_prefix_normalized_dash(self):
|
|
"""A normalized dash mismatch falls through to the normalized branch."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Financial-Corporations", 0, True),
|
|
(2.0, "Financial corporations, Detail", 1, False),
|
|
]
|
|
)
|
|
result = ctx.find_ancestor_part_prefix(2.0, "Financial corporations, Detail")
|
|
assert result == "Financial corporations, "
|
|
|
|
def test_simplify_title_only_paren_strip(self):
|
|
"""A title with only a unit paren -- no prefix/suffix to find."""
|
|
ctx = HierarchyContext([(1.0, "Plain (Percent)", 0, False)])
|
|
assert ctx.simplify_title(1.0, "Plain (Percent)") == "Plain"
|
|
|
|
|
|
class TestExtractUnitFromLabelMore:
|
|
"""A few more targeted unit-extraction tests."""
|
|
|
|
def test_paren_with_percent_sign(self):
|
|
"""A trailing '(%)' is detected."""
|
|
assert extract_unit_from_label("Inflation (%)") == "%"
|
|
|
|
|
|
class TestPivotTableModeMoreEdgeCases:
|
|
"""Additional pivot_table_mode edge cases."""
|
|
|
|
@staticmethod
|
|
def _base_df(rows: list[dict[str, Any]]) -> pd.DataFrame:
|
|
"""Default builder with NaN value."""
|
|
defaults = {
|
|
"title": "",
|
|
"level": 0,
|
|
"order": 0.0,
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": float("nan"),
|
|
"unit": "",
|
|
"scale": "",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": None,
|
|
"parent_id": None,
|
|
"indicator_code": None,
|
|
"symbol": "S",
|
|
}
|
|
out = []
|
|
for r in rows:
|
|
d = dict(defaults)
|
|
d.update(r)
|
|
out.append(d)
|
|
return pd.DataFrame(out)
|
|
|
|
def test_parent_lookup_break_when_missing(self):
|
|
"""An unresolvable parent_id chain breaks the parent traversal loop."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "completely_missing_id",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_unit_like_transformation_filtered(self):
|
|
"""A TYPE_OF_TRANSFORMATION dim with a unit-like value is filtered from label."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Detail",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"type_of_transformation_code": "USD",
|
|
"type_of_transformation": "US dollar",
|
|
"sector_code": "S1",
|
|
"sector": "Households",
|
|
"indicator_code": "X",
|
|
"series_id": "p1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Detail",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 7.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"type_of_transformation_code": "USD",
|
|
"type_of_transformation": "US dollar",
|
|
"sector_code": "S2",
|
|
"sector": "Firms",
|
|
"indicator_code": "X",
|
|
"series_id": "p2",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_isora_underscored_promoted_title(self):
|
|
"""An ISORA promoted header with '___' in title is skipped (line 1714)."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "1. Topic",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Abc___Xyz",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Detail",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n2",
|
|
"value": 7.0,
|
|
"unit": "Percent",
|
|
"scale": "Units",
|
|
"indicator_code": "Y",
|
|
},
|
|
]
|
|
)
|
|
meta = {
|
|
"table": {
|
|
"hierarchy_name": "1. INDICATORS BY TOPIC",
|
|
"dataflow_id": "ISORA",
|
|
}
|
|
}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_track_skipped_parent_continues_for_empty(self):
|
|
"""_track_skipped_parent_ids continues when a value is empty (line 1598)."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Credit",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": None,
|
|
"parent_id": "n1",
|
|
"indicator_code": None,
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Trade, Credit, Goods",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n2",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_promote_level_break_on_missing_parent(self):
|
|
"""_promote_level_if_parent_skipped breaks when lookup fails (line 1622)."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Credit",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "missing_parent",
|
|
"indicator_code": "C",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Trade, Credit, Goods",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n2",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_prune_deeper_headers(self):
|
|
"""When level drops, deeper levels' last_meaningful_header is pruned."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Root",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "r",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Mid",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "m",
|
|
"parent_id": "r",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Mid leaf",
|
|
"hierarchy_node_id": "ml",
|
|
"parent_id": "m",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
{
|
|
"order": 4.0,
|
|
"level": 0,
|
|
"title": "OtherRoot",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "r2",
|
|
},
|
|
{
|
|
"order": 5.0,
|
|
"level": 1,
|
|
"title": "OtherRoot, sub",
|
|
"hierarchy_node_id": "r2s",
|
|
"parent_id": "r2",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "Y",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Root", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_paren_unit_suffix_kept_for_promoted_header(self):
|
|
"""A promoted header carrying parenthetical unit applies original_unit_suffix."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Indicator (Percent of GDP)",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Indicator, X",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "Percent",
|
|
"scale": "Units",
|
|
"indicator_code": "X",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 1,
|
|
"title": "Indicator, Y",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n1",
|
|
"value": 10.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "Y",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Indicator", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_dim_dedup_same_grouping_key(self):
|
|
"""When two raw rows share the same grouping_key, the later is appended (1562-1565)."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Parent",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S1",
|
|
"sector": "A",
|
|
"type_of_transformation_code": "T1",
|
|
"type_of_transformation": "Type1",
|
|
"indicator_code": "X",
|
|
"series_id": "p1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S1",
|
|
"sector": "A",
|
|
"type_of_transformation_code": "T1",
|
|
"type_of_transformation": "Type1",
|
|
"indicator_code": "X",
|
|
"series_id": "p2",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 3.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S2",
|
|
"sector": "B",
|
|
"type_of_transformation_code": "T2",
|
|
"type_of_transformation": "Type2",
|
|
"indicator_code": "X",
|
|
"series_id": "p3",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Parent", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_bop_promotion_to_ancestor_via_meaningful_header(self):
|
|
"""BOP suffix child finds a non-BOP ancestor header and rewrites itself."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Goods Trade Account",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "h0",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Goods, Credit",
|
|
"hierarchy_node_id": "h1",
|
|
"parent_id": "h0",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 1,
|
|
"title": "Goods, Debit",
|
|
"hierarchy_node_id": "h2",
|
|
"parent_id": "h0",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "Y",
|
|
},
|
|
]
|
|
)
|
|
meta = {
|
|
"table": {"hierarchy_name": "Goods Trade Account", "dataflow_id": "BOP"}
|
|
}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_raw_row_unit_scale_parsed_from_title(self):
|
|
"""A row missing unit/scale picks them up from title parsing."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Goods (US Dollar, Millions)",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "",
|
|
"scale": "",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_empty_order_df_skipped(self):
|
|
"""If sorted_orders iterates a value not in df, the loop continues (line 1669)."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "TRD_G",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
|
|
class TestParseUnitAndScaleExtra:
|
|
"""Additional tests for parse_unit_and_scale edge cases."""
|
|
|
|
def test_ratio_of_pattern(self):
|
|
"""'Ratio of X' splits into unit Ratio and scale 'of X'."""
|
|
unit, scale = parse_unit_and_scale("Ratio of GDP")
|
|
assert unit == "Ratio"
|
|
assert scale == "of Gdp"
|
|
|
|
def test_number_of_pattern(self):
|
|
"""'Number of X' splits into unit Number."""
|
|
unit, scale = parse_unit_and_scale("Number of People")
|
|
assert unit == "Number"
|
|
assert scale == "of People"
|
|
|
|
def test_index_of_pattern(self):
|
|
"""'Index of X' splits into unit Index."""
|
|
unit, scale = parse_unit_and_scale("Index of Prices")
|
|
assert unit == "Index"
|
|
assert scale == "of Prices"
|
|
|
|
|
|
class TestHierarchyContextEmptyAncestor:
|
|
"""Edge cases for ancestor traversal with empty titles."""
|
|
|
|
def test_ancestor_part_prefix_skips_empty_title(self):
|
|
"""An ancestor row with an empty title is skipped via continue."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "", 0, True),
|
|
(2.0, "Top context", 1, True),
|
|
(3.0, "Top context, Detail", 2, False),
|
|
]
|
|
)
|
|
out = ctx.find_ancestor_part_prefix(3.0, "Top context, Detail")
|
|
assert out == "Top context, "
|
|
|
|
def test_ancestor_part_prefix_financial_assets_alias(self):
|
|
"""An ancestor containing 'Financial assets' adds the alias to ancestor_parts."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Holdings, Financial assets", 0, True),
|
|
(2.0, "Financial assets, Bonds", 1, False),
|
|
]
|
|
)
|
|
out = ctx.find_ancestor_part_prefix(2.0, "Financial assets, Bonds")
|
|
assert out == "Financial assets, "
|
|
|
|
def test_ancestor_part_prefix_case_insensitive_match(self):
|
|
"""A case-insensitive match returns the matched slice from target_title."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Other Investment", 0, True),
|
|
(2.0, "other investment, Loans", 1, False),
|
|
]
|
|
)
|
|
out = ctx.find_ancestor_part_prefix(2.0, "other investment, Loans")
|
|
assert out == "other investment, "
|
|
|
|
def test_ancestor_part_prefix_case_insensitive_bop_remainder_skipped(self):
|
|
"""A case-insensitive match whose remainder is BOP-only continues the loop."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Other Investment", 0, True),
|
|
(2.0, "other investment, Net", 1, False),
|
|
]
|
|
)
|
|
out = ctx.find_ancestor_part_prefix(2.0, "other investment, Net")
|
|
assert out is None
|
|
|
|
|
|
class TestSimplifyTitleExtraPaths:
|
|
"""Additional paths through simplify_title."""
|
|
|
|
def test_part_prefix_strip_via_ancestor_part(self):
|
|
"""An ancestor whose part matches strips via the part_prefix loop."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Z, Assets", 0, True),
|
|
(2.0, "Assets, X", 1, False),
|
|
]
|
|
)
|
|
assert ctx.simplify_title(2.0, "Assets, X") == "X"
|
|
|
|
def test_sibling_strip_leaves_bop_only_restores(self):
|
|
"""A sibling-prefix strip that leaves only 'Net' restores original title."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Foo, Bar, Net", 0, False),
|
|
(2.0, "Foo, Bar, X", 0, False),
|
|
(3.0, "Foo, Bar, Y", 0, False),
|
|
]
|
|
)
|
|
assert ctx.simplify_title(1.0, "Foo, Bar, Net") == "Foo, Bar, Net"
|
|
|
|
def test_part_prefix_consumes_full_title_restores(self):
|
|
"""A part_prefix strip that empties the title restores original."""
|
|
ctx = HierarchyContext(
|
|
[
|
|
(1.0, "Z, Assets", 0, True),
|
|
(2.0, "Assets, ", 1, False),
|
|
]
|
|
)
|
|
assert ctx.simplify_title(2.0, "Assets, ") == "Assets, "
|
|
|
|
|
|
class TestPivotTableModeCoverageGaps:
|
|
"""Targeted pivot_table_mode coverage gaps."""
|
|
|
|
@staticmethod
|
|
def _base_df(rows: list[dict[str, Any]]) -> pd.DataFrame:
|
|
"""Default row builder.
|
|
|
|
Parameters
|
|
----------
|
|
rows : list[dict[str, Any]]
|
|
List of partial row dicts merged with defaults.
|
|
|
|
Returns
|
|
-------
|
|
pd.DataFrame
|
|
Built DataFrame.
|
|
"""
|
|
defaults = {
|
|
"title": "",
|
|
"level": 0,
|
|
"order": 0.0,
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
"date": "2020-01-01",
|
|
"value": float("nan"),
|
|
"unit": "",
|
|
"scale": "",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": None,
|
|
"parent_id": None,
|
|
"indicator_code": None,
|
|
"symbol": "S",
|
|
}
|
|
out = []
|
|
for r in rows:
|
|
d = dict(defaults)
|
|
d.update(r)
|
|
out.append(d)
|
|
return pd.DataFrame(out)
|
|
|
|
def test_country_without_data_for_order_skipped(self):
|
|
"""A country lacking rows at a given order hits the country_df.empty continue."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "G",
|
|
"country": "USA",
|
|
"country_code": "USA",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 6.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "G",
|
|
"country": "CAN",
|
|
"country_code": "CAN",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA", "CAN", "MEX"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_no_symbol_column_falls_back_to_empty(self):
|
|
"""A DataFrame with neither dim_code cols nor symbol triggers the symbol-fallback line."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "GDP (Millions of US Dollar)",
|
|
"is_category_header": False,
|
|
"hierarchy_node_id": "n1",
|
|
"value": 10.0,
|
|
"indicator_code": "G",
|
|
},
|
|
]
|
|
).drop(columns=["symbol", "indicator_code"])
|
|
meta = {"table": {"hierarchy_name": "GDP", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_unit_scale_parsed_from_title_fills_missing(self):
|
|
"""A row without unit/scale parses them from the title."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade, Goods (Millions, US Dollar)",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"indicator_code": "G",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_format_dim_labels_empty_grouping_key_returns_empty(self):
|
|
"""An empty grouping_key yields an empty dim label string."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Top",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, A",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "A",
|
|
"sector_code": "S1",
|
|
"sector": "AA",
|
|
"series_id": "a",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, B",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "B",
|
|
"sector_code": "S2",
|
|
"sector": "BB",
|
|
"series_id": "b",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, C",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 3.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "C",
|
|
"series_id": "c",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Top", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_unit_like_transformation_filtered_from_labels(self):
|
|
"""A unit-like TYPE_OF_TRANSFORMATION value is filtered from dim labels."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Top",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"type_of_transformation_code": "T1",
|
|
"type_of_transformation": "US dollar",
|
|
"indicator_code": "A",
|
|
"series_id": "a",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"type_of_transformation_code": "T2",
|
|
"type_of_transformation": "Euro",
|
|
"indicator_code": "A",
|
|
"series_id": "b",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Top", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_track_skipped_with_empty_string_node_id(self):
|
|
"""A bop-only header with empty hierarchy_node_id continues past the empty value."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Trade",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Credit",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "",
|
|
"parent_id": "n1",
|
|
"indicator_code": "CRED",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Trade, Credit, Goods",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "CRED",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "X",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Trade", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_data_row_best_suffix_stripped(self):
|
|
"""A data row whose title ends with an ancestor's name strips the suffix."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Survey Metric",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Detail, Survey Metric",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "D",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Survey Metric", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_data_row_part_prefix_stripped(self):
|
|
"""A data row whose title starts with an ancestor part strips the prefix."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Z, Trade total",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Trade total, Detail",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 5.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "D",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Z, Trade total", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_bop_promotion_skips_empty_and_bop_ancestor_levels(self):
|
|
"""BOP promotion walks past empty and bop-ending ancestor levels to a meaningful one."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Top",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, Sub, Net",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
},
|
|
{
|
|
"order": 3.0,
|
|
"level": 2,
|
|
"title": "Top, Sub, Net, Detail",
|
|
"hierarchy_node_id": "n3",
|
|
"parent_id": "n2",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "D",
|
|
},
|
|
{
|
|
"order": 4.0,
|
|
"level": 3,
|
|
"title": "Top, Sub, Net, Detail, Credit",
|
|
"hierarchy_node_id": "n4",
|
|
"parent_id": "n3",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"indicator_code": "E",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Top", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|
|
|
|
def test_multi_dim_no_counterpart_with_empty_grouping_row(self):
|
|
"""A row at a multi-dim order with no dim values falls into the empty-grouping branch."""
|
|
df = self._base_df(
|
|
[
|
|
{
|
|
"order": 1.0,
|
|
"level": 0,
|
|
"title": "Top",
|
|
"is_category_header": True,
|
|
"hierarchy_node_id": "n1",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 1.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S1",
|
|
"sector": "A",
|
|
"indicator_code": "X",
|
|
"series_id": "a",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 2.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": "S2",
|
|
"sector": "B",
|
|
"indicator_code": "X",
|
|
"series_id": "b",
|
|
},
|
|
{
|
|
"order": 2.0,
|
|
"level": 1,
|
|
"title": "Top, Item",
|
|
"hierarchy_node_id": "n2",
|
|
"parent_id": "n1",
|
|
"value": 3.0,
|
|
"unit": "US Dollar",
|
|
"scale": "Millions",
|
|
"sector_code": None,
|
|
"sector": None,
|
|
"indicator_code": "X",
|
|
"series_id": "c",
|
|
},
|
|
]
|
|
)
|
|
meta = {"table": {"hierarchy_name": "Top", "dataflow_id": "BOP"}}
|
|
result = pivot_table_mode(df, ["2020-01-01"], ["USA"], meta)
|
|
assert isinstance(result, pd.DataFrame)
|