133 lines
3.7 KiB
Python
133 lines
3.7 KiB
Python
import csv
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from order_bak.csv_utils import (
|
|
CSV_COLUMNS,
|
|
entity_to_row,
|
|
row_to_entity,
|
|
write_csv,
|
|
read_csv,
|
|
)
|
|
|
|
|
|
def test_csv_columns_defined():
|
|
assert "PartitionKey" in CSV_COLUMNS
|
|
assert "RowKey" in CSV_COLUMNS
|
|
assert "OrderTime" in CSV_COLUMNS
|
|
assert "state" in CSV_COLUMNS
|
|
assert len(CSV_COLUMNS) == 13
|
|
|
|
|
|
def test_entity_to_row_converts_datetime():
|
|
entity = {
|
|
"PartitionKey": "user1",
|
|
"RowKey": "order1",
|
|
"OrderTime": datetime(2025, 6, 15, 10, 30, 0, tzinfo=timezone.utc),
|
|
"ReceiveTime": datetime(2025, 6, 15, 11, 0, 0, tzinfo=timezone.utc),
|
|
"revenue": 99.5,
|
|
"platformOrder": "plat123",
|
|
"prodCount": 2,
|
|
"pfId": "user1",
|
|
"prodId": "prod1",
|
|
"prodPrice": 49.75,
|
|
"state": 1,
|
|
"ver": 3,
|
|
"customData": '{"key":"val"}',
|
|
}
|
|
row = entity_to_row(entity)
|
|
assert row["OrderTime"] == "2025-06-15T10:30:00+00:00"
|
|
assert row["ReceiveTime"] == "2025-06-15T11:00:00+00:00"
|
|
assert row["revenue"] == "99.5"
|
|
assert row["state"] == "1"
|
|
|
|
|
|
def test_entity_to_row_handles_none():
|
|
entity = {
|
|
"PartitionKey": "user1",
|
|
"RowKey": "order1",
|
|
"OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
|
|
"ReceiveTime": None,
|
|
"revenue": None,
|
|
"platformOrder": None,
|
|
"prodCount": None,
|
|
"pfId": None,
|
|
"prodId": None,
|
|
"prodPrice": None,
|
|
"state": 0,
|
|
"ver": None,
|
|
"customData": None,
|
|
}
|
|
row = entity_to_row(entity)
|
|
assert row["ReceiveTime"] == ""
|
|
assert row["revenue"] == ""
|
|
|
|
|
|
def test_row_to_entity_roundtrip(tmp_path):
|
|
original = {
|
|
"PartitionKey": "user1",
|
|
"RowKey": "order1",
|
|
"OrderTime": "2025-06-15T10:30:00+00:00",
|
|
"ReceiveTime": "",
|
|
"revenue": "99.5",
|
|
"platformOrder": "plat123",
|
|
"prodCount": "2",
|
|
"pfId": "user1",
|
|
"prodId": "prod1",
|
|
"prodPrice": "49.75",
|
|
"state": "1",
|
|
"ver": "3",
|
|
"customData": '{"key":"val"}',
|
|
}
|
|
entity = row_to_entity(original)
|
|
assert entity["PartitionKey"] == "user1"
|
|
assert entity["state"] == 1
|
|
assert entity["revenue"] == 99.5
|
|
assert entity["prodCount"] == 2
|
|
assert entity["ReceiveTime"] is None
|
|
|
|
|
|
def test_write_and_read_csv(tmp_path):
|
|
entities = [
|
|
{
|
|
"PartitionKey": "user1",
|
|
"RowKey": "order1",
|
|
"OrderTime": datetime(2025, 6, 15, 10, 30, tzinfo=timezone.utc),
|
|
"ReceiveTime": None,
|
|
"revenue": 99.5,
|
|
"platformOrder": "p1",
|
|
"prodCount": 2,
|
|
"pfId": "user1",
|
|
"prodId": "prod1",
|
|
"prodPrice": 49.75,
|
|
"state": 1,
|
|
"ver": 3,
|
|
"customData": None,
|
|
},
|
|
{
|
|
"PartitionKey": "user2",
|
|
"RowKey": "order2",
|
|
"OrderTime": datetime(2025, 6, 16, 12, 0, tzinfo=timezone.utc),
|
|
"ReceiveTime": datetime(2025, 6, 16, 13, 0, tzinfo=timezone.utc),
|
|
"revenue": 10.0,
|
|
"platformOrder": "p2",
|
|
"prodCount": 1,
|
|
"pfId": "user2",
|
|
"prodId": "prod2",
|
|
"prodPrice": 10.0,
|
|
"state": 1,
|
|
"ver": 3,
|
|
"customData": '{"note":"test"}',
|
|
},
|
|
]
|
|
csv_path = tmp_path / "test.csv"
|
|
write_csv(csv_path, entities)
|
|
|
|
result = list(read_csv(csv_path))
|
|
assert len(result) == 2
|
|
assert result[0]["PartitionKey"] == "user1"
|
|
assert result[1]["PartitionKey"] == "user2"
|
|
assert result[1]["revenue"] == 10.0
|
|
assert result[0]["ReceiveTime"] is None
|
|
assert result[1]["ReceiveTime"] is not None
|