tests/test_cpi.py
Rama main · réplica de lectura del repositorio original.
"""The CPI the 2027 CGT rules index a cost base by (decisions.md #149).
ITAA 1997 as amended by Act No. 49 of 2026: money spent from 1 July 2027 is
indexed by the index number for the quarter of the CGT event over the number
for the quarter it was spent (s 960-275(1B)), worked to three decimal places,
rounding up from 5 (s 960-275(5)), and not at all at a factor of 1 or less
(s 960-270(2)). The index number is the All Groups CPI for the eight capitals,
first published for the quarter (s 960-280(1)); after a change of reference
period only numbers on the new one count (s 960-280(3)).
"""
from __future__ import annotations
import datetime as dt
from decimal import Decimal
import pytest
from freezegun import freeze_time
from sqlalchemy import select
from app import cpi, providers
from app.models import CpiQuarter
BASE = "Sep 2025 = 100.0"
D = dt.date
def _abs(quarters: dict[str, float], base: str = BASE) -> dict:
"""What the ABS data API answers, cut to the parts that are read: the
observation periods, one series of values, and the reference period."""
periods = list(quarters)
return {"data": {
"structures": [{
"dimensions": {"observation": [{"id": "TIME_PERIOD",
"values": [{"id": p} for p in periods]}]},
"attributes": {"dataSet": [{"id": "BASE_PERIOD", "values": [{"name": base}]}]},
}],
"dataSets": [{"attributes": [0], "series": {"0:0:0:0:0": {"observations": {
str(i): [quarters[p]] for i, p in enumerate(periods)}}}}],
}}
@pytest.fixture
def abs_says(monkeypatch):
"""The ABS, answering with whatever the test puts in `answer`, and the
requests it was sent."""
seen = {"calls": [], "answer": _abs({})}
def get_json(url, params):
seen["calls"].append((url, params))
if isinstance(seen["answer"], Exception):
raise seen["answer"]
return seen["answer"]
monkeypatch.setattr(providers, "_get_json", get_json)
return seen
def _stored(pf) -> dict:
return {row.quarter: (row.index_number, row.reference_period)
for row in pf.scalars(select(CpiQuarter))}
def _store(pf, quarters: dict[dt.date, str], base: str = BASE):
for quarter, number in quarters.items():
pf.add(CpiQuarter(quarter=quarter, index_number=Decimal(number), reference_period=base))
pf.flush()
@pytest.mark.parametrize("day, quarter", [
(D(2027, 7, 1), D(2027, 7, 1)), (D(2027, 9, 30), D(2027, 7, 1)),
(D(2027, 12, 31), D(2027, 10, 1)), (D(2028, 2, 29), D(2028, 1, 1)),
(D(2028, 6, 30), D(2028, 4, 1)),
])
def test_a_day_falls_in_its_calendar_quarter(day, quarter):
assert cpi.quarter(day) == quarter
def test_the_factor_is_the_ratio_to_three_places_rounding_up_from_five(pf):
"""103.25 over 100.00 is 1.0325: 1.033, where rounding half to even (what
Decimal does by default) would give 1.032."""
_store(pf, {D(2027, 7, 1): "100.00", D(2028, 4, 1): "103.25", D(2028, 7, 1): "110.28"})
book = cpi.CpiBook(pf)
assert book.factor(D(2027, 7, 1), D(2028, 5, 15)) == Decimal("1.033")
assert book.factor(D(2027, 8, 20), D(2028, 9, 30)) == Decimal("1.103")
def test_at_one_or_less_nothing_is_indexed(pf):
_store(pf, {D(2027, 7, 1): "100.00", D(2027, 10, 1): "99.80", D(2028, 1, 1): "100.04"})
book = cpi.CpiBook(pf)
assert book.factor(D(2027, 7, 1), D(2027, 11, 1)) == 1 # deflation: not below 1
assert book.factor(D(2027, 7, 1), D(2028, 1, 1)) == 1 # 1.0004 is 1.000
def test_in_the_same_quarter_the_factor_is_one_without_its_number(pf):
"""A number over itself: a sale in the September 2027 quarter of a parcel
held through 30 June 2027 need not wait for that quarter's CPI."""
assert cpi.CpiBook(pf).factor(D(2027, 7, 1), D(2027, 8, 15)) == 1
def test_a_quarter_not_yet_published_gives_no_factor(pf):
_store(pf, {D(2027, 7, 1): "100.00"})
assert cpi.CpiBook(pf).factor(D(2027, 7, 1), D(2027, 12, 1)) is None
@freeze_time("2027-11-15")
def test_the_feed_stores_each_quarter_it_could_need(pf, abs_says):
abs_says["answer"] = _abs({"2027-Q3": 103.12})
added = cpi.refresh(pf)
assert added == 1
assert _stored(pf) == {D(2027, 7, 1): (Decimal("103.120"), BASE)}
[(url, params)] = abs_says["calls"]
# All Groups (10001), original (10), eight capitals (50), quarterly, from
# the first quarter a 2027 cost base can be indexed from.
assert url.endswith("/ABS,CPI,2.0.0/1.10001.10.50.Q")
assert params == {"format": "jsondata", "startPeriod": "2027-Q3"}
@freeze_time("2028-02-10")
def test_a_revised_number_does_not_replace_the_first_published(pf, abs_says):
_store(pf, {D(2027, 7, 1): "103.12"})
abs_says["answer"] = _abs({"2027-Q3": 103.30, "2027-Q4": 104.01})
added = cpi.refresh(pf)
assert added == 1
assert _stored(pf) == {D(2027, 7, 1): (Decimal("103.120"), BASE),
D(2027, 10, 1): (Decimal("104.010"), BASE)}
@freeze_time("2028-02-10")
def test_a_new_reference_period_replaces_the_series(pf, abs_says):
"""Numbers on two reference periods cannot be divided by each other."""
_store(pf, {D(2027, 7, 1): "103.12"})
abs_says["answer"] = _abs({"2027-Q3": 100.0, "2027-Q4": 100.69}, base="Sep 2027 = 100.0")
cpi.refresh(pf)
assert _stored(pf) == {D(2027, 7, 1): (Decimal("100.000"), "Sep 2027 = 100.0"),
D(2027, 10, 1): (Decimal("100.690"), "Sep 2027 = 100.0")}
@pytest.mark.parametrize("today, stored", [
("2027-06-30", {}), # nothing can be indexed yet
("2027-09-30", {}), # September's not finished
("2027-11-15", {D(2027, 7, 1): "103.12"}), # the newest finished one is here
])
def test_nothing_is_asked_when_there_is_nothing_new_to_have(pf, abs_says, today, stored):
_store(pf, stored)
with freeze_time(today):
added = cpi.refresh(pf)
assert (added, abs_says["calls"]) == (0, [])
@freeze_time("2027-11-15")
def test_the_feed_run_asks_and_carries_on_when_the_abs_does_not_answer(pf, abs_says,
monkeypatch):
from app import pricefeed
from app.settings import PortfolioSettings
abs_says["answer"] = _abs({"2027-Q3": 103.12})
counts = pricefeed.run_feed(pf, PortfolioSettings())
abs_says["answer"] = TimeoutError("timed out")
pf.query(CpiQuarter).delete()
failed = pricefeed.run_feed(pf, PortfolioSettings())
assert counts["cpi"] == 1
assert failed["cpi"] == 0
assert pricefeed.last_run_sources["cpi:ABS"] == "abs failed: timed out"