52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from canonical_layer.period_snapshot import (
|
|
normalize_dt,
|
|
parse_dt,
|
|
parse_record_datetime,
|
|
window_bounds_from_key,
|
|
window_key,
|
|
)
|
|
|
|
|
|
def test_parse_dt_odata_and_iso() -> None:
|
|
from_odata = parse_dt("/Date(1583020800000)/")
|
|
assert from_odata is not None
|
|
assert from_odata.year == 2020
|
|
assert from_odata.month == 3
|
|
assert from_odata.day == 1
|
|
|
|
from_iso = parse_dt("2020-03-01T00:00:00Z")
|
|
assert from_iso is not None
|
|
assert from_iso.tzinfo is not None
|
|
assert normalize_dt(from_iso).year == 2020
|
|
|
|
|
|
def test_window_key_and_bounds_month() -> None:
|
|
dt = datetime(2020, 11, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
key = window_key(dt, granularity="month")
|
|
assert key == "2020-11"
|
|
|
|
start, end = window_bounds_from_key(key, granularity="month")
|
|
assert start.isoformat() == "2020-11-01T00:00:00+00:00"
|
|
assert end.isoformat() == "2020-12-01T00:00:00+00:00"
|
|
|
|
|
|
def test_window_key_and_bounds_week() -> None:
|
|
dt = datetime(2020, 12, 30, 7, 0, 0, tzinfo=timezone.utc)
|
|
key = window_key(dt, granularity="week")
|
|
assert key.startswith("2020-W")
|
|
|
|
start, end = window_bounds_from_key(key, granularity="week")
|
|
assert end > start
|
|
|
|
|
|
def test_parse_record_datetime_finds_common_fields() -> None:
|
|
row = {"Date": "2020-05-20T10:00:00Z"}
|
|
dt = parse_record_datetime(row)
|
|
assert dt is not None
|
|
assert dt.year == 2020
|
|
assert dt.month == 5
|