63 lines
2.4 KiB
Python
63 lines
2.4 KiB
Python
from datetime import datetime, timezone
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from central.app.datasets import bootstrap_v1_extra as parsers
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.models import Base, RtcCstIbsCbsEntry
|
|
|
|
ROOT = Path(__file__).parents[2] / "sources" / "raw" / "bootstrap_v1"
|
|
CCLASS = ROOT / "cClassTrib 2026-06-22.xlsx"
|
|
CBS = ROOT / "IT 2026.002 v.1.00 - Aliquotas da CBS.xlsx"
|
|
pytestmark = pytest.mark.skipif(not (CCLASS.exists() and CBS.exists()), reason="RTC bootstrap files not present locally")
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def _cclass():
|
|
return parsers.parse_rtc_cclasstrib(CCLASS.read_bytes())
|
|
|
|
|
|
def _session():
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
return Session(engine)
|
|
|
|
|
|
def test_cclasstrib_keys_unique_and_rich_fields_kept():
|
|
p = _cclass()
|
|
codes = [e["cclasstrib"] for e in p.entries]
|
|
assert len(codes) == len(set(codes)) >= 100
|
|
assert all(len(c) == 6 for c in codes)
|
|
assert p.entries[0]["indicators"] and p.entries[0]["d_ini_vig"]
|
|
assert {e["cst"] for e in p.entries} <= {e["cst"] for e in p.extra_entries}
|
|
|
|
|
|
def test_cbs_rates_2026_numeric_and_2027_pending():
|
|
p = parsers.parse_rtc_cbs_rates(CBS.read_bytes())
|
|
by_year = {e["year"]: e for e in p.entries}
|
|
assert by_year[2026]["rate"] == pytest.approx(0.009)
|
|
assert by_year[2027]["rate"] is None and by_year[2027]["rate_text"]
|
|
|
|
|
|
def test_import_idempotent_lookup_and_versions():
|
|
with _session() as s:
|
|
svc = DatasetService(s)
|
|
assert svc.dataset_versions()["rtc_cclasstrib"] is None
|
|
now = datetime.now(timezone.utc)
|
|
a = svc.import_rtc_cclasstrib(_cclass(), "file://x", now)
|
|
b = svc.import_rtc_cclasstrib(_cclass(), "file://x", now)
|
|
assert a.id == b.id and b.status == "ACTIVE" and b.record_count == len(_cclass().entries)
|
|
assert s.query(RtcCstIbsCbsEntry).count() == len(_cclass().extra_entries)
|
|
assert svc.rtc_cclasstrib(code="000001")
|
|
|
|
cbs_payload = parsers.parse_rtc_cbs_rates(CBS.read_bytes())
|
|
c = svc.import_rtc_cbs_rates(cbs_payload, "file://y", now)
|
|
assert svc.import_rtc_cbs_rates(cbs_payload, "file://y", now).id == c.id
|
|
assert svc.rtc_cbs_rate(2026).rate == pytest.approx(0.009)
|
|
v = svc.dataset_versions()
|
|
assert v["rtc_cclasstrib"] and v["rtc_cbs_rates"]
|