84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
import os
|
|
|
|
import pytest
|
|
|
|
from apps.competitions.models import Competition, Season
|
|
from apps.ingestion.models import IngestionError, IngestionRun
|
|
from apps.ingestion.services.sync import run_sync_job
|
|
from apps.players.models import Player
|
|
from apps.providers.exceptions import ProviderRateLimitError
|
|
from apps.stats.models import PlayerSeason, PlayerSeasonStats
|
|
from apps.teams.models import Team
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_run_full_sync_creates_domain_objects(settings):
|
|
settings.PROVIDER_DEFAULT_NAMESPACE = "mvp_demo"
|
|
|
|
run = run_sync_job(provider_namespace="mvp_demo", job_type=IngestionRun.JobType.FULL_SYNC)
|
|
|
|
assert run.status == IngestionRun.RunStatus.SUCCESS
|
|
assert Competition.objects.count() >= 1
|
|
assert Team.objects.count() >= 1
|
|
assert Season.objects.count() >= 1
|
|
assert Player.objects.count() >= 1
|
|
assert PlayerSeason.objects.count() >= 1
|
|
assert PlayerSeasonStats.objects.count() >= 1
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_full_sync_is_idempotent(settings):
|
|
settings.PROVIDER_DEFAULT_NAMESPACE = "mvp_demo"
|
|
|
|
run_sync_job(provider_namespace="mvp_demo", job_type=IngestionRun.JobType.FULL_SYNC)
|
|
counts_after_first = {
|
|
"competition": Competition.objects.count(),
|
|
"team": Team.objects.count(),
|
|
"season": Season.objects.count(),
|
|
"player": Player.objects.count(),
|
|
"player_season": PlayerSeason.objects.count(),
|
|
"player_stats": PlayerSeasonStats.objects.count(),
|
|
}
|
|
|
|
run_sync_job(provider_namespace="mvp_demo", job_type=IngestionRun.JobType.FULL_SYNC)
|
|
counts_after_second = {
|
|
"competition": Competition.objects.count(),
|
|
"team": Team.objects.count(),
|
|
"season": Season.objects.count(),
|
|
"player": Player.objects.count(),
|
|
"player_season": PlayerSeason.objects.count(),
|
|
"player_stats": PlayerSeasonStats.objects.count(),
|
|
}
|
|
|
|
assert counts_after_first == counts_after_second
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_incremental_sync_runs_successfully(settings):
|
|
settings.PROVIDER_DEFAULT_NAMESPACE = "mvp_demo"
|
|
|
|
run = run_sync_job(
|
|
provider_namespace="mvp_demo",
|
|
job_type=IngestionRun.JobType.INCREMENTAL,
|
|
cursor="demo-cursor",
|
|
)
|
|
|
|
assert run.status == IngestionRun.RunStatus.SUCCESS
|
|
assert run.records_processed > 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_run_sync_handles_rate_limit(settings):
|
|
settings.PROVIDER_DEFAULT_NAMESPACE = "mvp_demo"
|
|
os.environ["PROVIDER_MVP_FORCE_RATE_LIMIT"] = "1"
|
|
|
|
with pytest.raises(ProviderRateLimitError):
|
|
run_sync_job(provider_namespace="mvp_demo", job_type=IngestionRun.JobType.FULL_SYNC)
|
|
|
|
run = IngestionRun.objects.order_by("-id").first()
|
|
assert run is not None
|
|
assert run.status == IngestionRun.RunStatus.FAILED
|
|
assert IngestionError.objects.filter(ingestion_run=run).exists()
|
|
|
|
os.environ.pop("PROVIDER_MVP_FORCE_RATE_LIMIT", None)
|