104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
from pathlib import Path
|
|
|
|
from mywhoosh_garmin_sync.state import ActivityStore
|
|
|
|
|
|
def test_state_transitions_and_hash_lookup(tmp_path: Path):
|
|
store = ActivityStore(tmp_path / "state.sqlite3")
|
|
store.initialize()
|
|
|
|
store.record_downloaded(
|
|
source_ref="source-1",
|
|
title="Ride",
|
|
source_url="https://example.test/ride.fit",
|
|
raw_path=tmp_path / "raw.fit",
|
|
raw_sha256="raw-hash",
|
|
)
|
|
assert store.get_status("source-1") == "downloaded"
|
|
assert store.is_terminal_source("source-1") is False
|
|
|
|
store.mark_converted(
|
|
source_ref="source-1",
|
|
converted_path=tmp_path / "converted.fit",
|
|
converted_sha256="converted-hash",
|
|
patched_field_count=4,
|
|
)
|
|
assert store.get_status("source-1") == "converted"
|
|
assert store.is_uploaded_hash("converted-hash") is False
|
|
|
|
store.mark_uploaded("source-1", "123")
|
|
assert store.get_status("source-1") == "uploaded"
|
|
assert store.is_terminal_source("source-1") is True
|
|
assert store.is_uploaded_hash("raw-hash") is True
|
|
assert store.is_uploaded_hash("converted-hash") is True
|
|
|
|
|
|
def test_failed_activity_can_be_retried(tmp_path: Path):
|
|
store = ActivityStore(tmp_path / "state.sqlite3")
|
|
store.initialize()
|
|
store.record_downloaded("source-1", "Ride", None, tmp_path / "raw.fit", "hash")
|
|
store.mark_failed("source-1", "network failed")
|
|
|
|
assert store.get_status("source-1") == "failed"
|
|
assert store.is_terminal_source("source-1") is False
|
|
|
|
|
|
def test_sync_check_summary_round_trip(tmp_path: Path):
|
|
store = ActivityStore(tmp_path / "state.sqlite3")
|
|
store.initialize()
|
|
|
|
check_id = store.begin_sync_check()
|
|
store.finish_sync_check(
|
|
check_id=check_id,
|
|
status="problem",
|
|
message="MyWhoosh login did not complete",
|
|
error_type="mywhoosh_login",
|
|
downloaded_count=2,
|
|
processed_count=1,
|
|
failed_count=1,
|
|
)
|
|
|
|
latest = store.get_latest_sync_check()
|
|
|
|
assert latest is not None
|
|
assert latest.id == check_id
|
|
assert latest.status == "problem"
|
|
assert latest.message == "MyWhoosh login did not complete"
|
|
assert latest.error_type == "mywhoosh_login"
|
|
assert latest.downloaded_count == 2
|
|
assert latest.processed_count == 1
|
|
assert latest.failed_count == 1
|
|
assert latest.finished_at is not None
|
|
|
|
|
|
def test_recent_uploaded_activities_are_limited_and_ordered(tmp_path: Path):
|
|
store = ActivityStore(tmp_path / "state.sqlite3")
|
|
store.initialize()
|
|
|
|
for index in range(6):
|
|
source_ref = f"source-{index}"
|
|
store.record_downloaded(
|
|
source_ref=source_ref,
|
|
title=f"Ride {index}",
|
|
source_url=f"https://example.test/{index}",
|
|
raw_path=tmp_path / f"raw-{index}.fit",
|
|
raw_sha256=f"raw-hash-{index}",
|
|
)
|
|
store.mark_converted(
|
|
source_ref=source_ref,
|
|
converted_path=tmp_path / f"converted-{index}.fit",
|
|
converted_sha256=f"converted-hash-{index}",
|
|
patched_field_count=index,
|
|
)
|
|
store.mark_uploaded(source_ref, str(index))
|
|
|
|
recent = store.get_recent_uploaded_activities(5)
|
|
|
|
assert [activity.title for activity in recent] == [
|
|
"Ride 5",
|
|
"Ride 4",
|
|
"Ride 3",
|
|
"Ride 2",
|
|
"Ride 1",
|
|
]
|