309 lines
13 KiB
Python
309 lines
13 KiB
Python
import json
|
|
from dataclasses import asdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.community_importer import CommunityImportError, stage_observations
|
|
from app.community_review import ExternalReviewError, map_observation, publish_observation, suggest_aliases
|
|
from app.source_lifecycle import record_scheduled_source_check, record_source_check
|
|
from app.database import Base
|
|
from app.models import CatchReport, DataSource, ExternalEntityAlias, ExternalObservation, Fish, Waterbody
|
|
from rf4_research.community_sources import parse_rf4db_catches, parse_rf4map_point, parse_rf4posts_spot
|
|
|
|
|
|
FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "rf4db_catches_sample.html"
|
|
FIXTURES = FIXTURE.parent
|
|
|
|
|
|
def record(source: str = "rf4db", external_id: str = "catch-1") -> dict[str, object]:
|
|
urls = {
|
|
"rf4db": f"https://rf4db.com/ru/catches/{external_id}",
|
|
"rf4stat-fishing": f"https://rf4-stat.ru/fishing/{external_id}",
|
|
"rf4map": f"https://rf4map.ru/points/{external_id}",
|
|
"rf4posts-spot": f"https://rf4-posts.com/ru/spots/{external_id}",
|
|
}
|
|
return {
|
|
"source_system": source,
|
|
"source_external_id": external_id,
|
|
"source_url": urls.get(source, f"https://rf4-stat.ru/fishing/{external_id}"),
|
|
"fish": "Щука",
|
|
"fish_external_id": "pike",
|
|
"waterbody": "Тестовое озеро",
|
|
"waterbody_external_id": "test-lake",
|
|
"x": 71,
|
|
"y": 92,
|
|
"weight_g": None,
|
|
"published_at": None,
|
|
"bait": "Приманка",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def db() -> Session:
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
yield session
|
|
engine.dispose()
|
|
|
|
|
|
def test_staging_is_idempotent_and_preserves_first_seen(db: Session) -> None:
|
|
first = datetime(2026, 9, 3, 10, tzinfo=timezone.utc)
|
|
second = datetime(2026, 9, 3, 11, tzinfo=timezone.utc)
|
|
|
|
assert stage_observations(db, [record()], fetched_at=first) == (1, 0)
|
|
changed = record() | {"x": 73, "weight_g": 5_000}
|
|
assert stage_observations(db, [changed], fetched_at=second) == (0, 1)
|
|
|
|
item = db.scalar(select(ExternalObservation))
|
|
assert item is not None
|
|
assert (item.x, item.weight_g, item.status) == (73, 5_000, "staged")
|
|
assert item.first_seen_at.replace(tzinfo=timezone.utc) == first
|
|
assert item.last_seen_at.replace(tzinfo=timezone.utc) == second
|
|
assert db.scalar(select(func.count()).select_from(ExternalObservation)) == 1
|
|
source = db.get(DataSource, "rf4db")
|
|
assert source is not None and source.enabled is True
|
|
|
|
|
|
def test_external_ids_are_isolated_by_source(db: Session) -> None:
|
|
created, updated = stage_observations(db, [record("rf4db"), record("rf4stat-fishing")])
|
|
|
|
assert (created, updated) == (2, 0)
|
|
|
|
|
|
def test_research_sources_enter_enabled_staging(db: Session) -> None:
|
|
created, updated = stage_observations(db, [record("rf4map"), record("rf4posts-spot")])
|
|
|
|
assert (created, updated) == (2, 0)
|
|
assert db.get(DataSource, "rf4map").enabled is True
|
|
assert db.get(DataSource, "rf4posts-spot").enabled is True
|
|
|
|
|
|
def test_complete_observation_with_reviewed_aliases_is_published(db: Session) -> None:
|
|
source = DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=True)
|
|
fish = Fish(slug="pike", name_ru="Щука")
|
|
waterbody = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
|
|
db.add_all([source, fish, waterbody])
|
|
db.flush()
|
|
now = datetime.now(timezone.utc)
|
|
db.add_all([
|
|
ExternalEntityAlias(source_system="rf4db", entity_type="fish", external_id="pike", external_name="Щука", fish=fish, updated_at=now),
|
|
ExternalEntityAlias(source_system="rf4db", entity_type="waterbody", external_id="test-lake", external_name="Тестовое озеро", waterbody=waterbody, updated_at=now),
|
|
])
|
|
db.commit()
|
|
|
|
assert stage_observations(db, [record() | {"weight_g": 5_000}]) == (1, 0)
|
|
|
|
item = db.scalar(select(ExternalObservation))
|
|
assert item is not None
|
|
assert item.status == "published"
|
|
assert item.catch_report is not None
|
|
assert item.catch_report.fish_id == fish.id
|
|
assert item.catch_report.waterbody_id == waterbody.id
|
|
assert db.scalar(select(func.count()).select_from(CatchReport)) == 1
|
|
|
|
assert stage_observations(db, [record() | {"weight_g": 5_000}]) == (0, 1)
|
|
db.refresh(item)
|
|
assert item.status == "published"
|
|
assert db.scalar(select(func.count()).select_from(CatchReport)) == 1
|
|
|
|
|
|
def test_changed_published_record_requires_review_and_reuses_report(db: Session) -> None:
|
|
fish = Fish(slug="pike", name_ru="Щука")
|
|
water = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
|
|
db.add_all([fish, water])
|
|
db.commit()
|
|
# D04: auto-publish now works with name match fallback
|
|
stage_observations(db, [record() | {"weight_g": 5000}])
|
|
item = db.scalar(select(ExternalObservation))
|
|
# Item auto-published via name match fallback (D04)
|
|
assert item.status == "published"
|
|
report = publish_observation(db, item)
|
|
report_id = report.id
|
|
stage_observations(db, [record() | {"weight_g": 6000}])
|
|
assert item.status == "staged"
|
|
assert report.moderation_status.value == "pending"
|
|
assert report.weight_g == 5000
|
|
assert item.weight_g == 6000
|
|
assert item.moderation_version == 1
|
|
assert item.reviewed_at is not None
|
|
stage_observations(db, [record() | {"weight_g": 6000}])
|
|
assert item.status == "staged"
|
|
with pytest.raises(ExternalReviewError):
|
|
publish_observation(db, item)
|
|
map_observation(db, item, fish, water)
|
|
updated = publish_observation(db, item)
|
|
assert updated.id == report_id
|
|
assert updated.weight_g == 6000
|
|
assert updated.moderation_status.value == "approved"
|
|
assert db.scalar(select(func.count()).select_from(CatchReport)) == 1
|
|
|
|
|
|
def test_missing_source_withdraws_published_record_until_manual_review(db: Session) -> None:
|
|
fish = Fish(slug="pike", name_ru="Щука")
|
|
water = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
|
|
db.add_all([fish, water])
|
|
db.commit()
|
|
seen = datetime(2026, 9, 13, 8, tzinfo=timezone.utc)
|
|
checked = datetime(2026, 9, 13, 9, tzinfo=timezone.utc)
|
|
stage_observations(db, [record() | {"weight_g": 5000}], fetched_at=seen)
|
|
item = db.scalar(select(ExternalObservation))
|
|
assert item is not None and item.catch_report is not None
|
|
|
|
record_source_check(db, item, "missing", checked_at=checked)
|
|
|
|
assert item.status == "withdrawn"
|
|
assert item.source_check_status == "missing"
|
|
assert item.source_checked_at.replace(tzinfo=timezone.utc) == checked
|
|
assert item.catch_report.moderation_status.value == "pending"
|
|
assert item.moderation_version == 1
|
|
|
|
stage_observations(db, [record() | {"weight_g": 5000}], fetched_at=checked)
|
|
assert item.status == "staged"
|
|
assert item.source_check_status == "available"
|
|
assert item.catch_report.moderation_status.value == "pending"
|
|
assert "reappeared" in (item.review_note or "")
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["temporary_error", "blocked"])
|
|
def test_non_authoritative_source_failures_do_not_withdraw(db: Session, status: str) -> None:
|
|
fish = Fish(slug="pike", name_ru="Щука")
|
|
water = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
|
|
db.add_all([fish, water])
|
|
db.commit()
|
|
stage_observations(db, [record() | {"weight_g": 5000}])
|
|
item = db.scalar(select(ExternalObservation))
|
|
assert item is not None and item.catch_report is not None
|
|
|
|
record_source_check(db, item, status) # type: ignore[arg-type]
|
|
|
|
assert item.status == "published"
|
|
assert item.catch_report.moderation_status.value == "approved"
|
|
assert item.source_check_status == status
|
|
|
|
|
|
def test_scheduled_failure_only_affects_exact_source_url(db: Session) -> None:
|
|
stage_observations(db, [
|
|
record(external_id="matching"),
|
|
record(external_id="other") | {"source_url": "https://rf4db.com/catches/other"},
|
|
])
|
|
|
|
affected = record_scheduled_source_check(
|
|
db,
|
|
source_system="rf4db",
|
|
source_url="https://rf4db.com/ru/catches/matching",
|
|
status="missing",
|
|
)
|
|
items = {item.source_external_id: item for item in db.scalars(select(ExternalObservation))}
|
|
|
|
assert affected == 1
|
|
assert items["matching"].status == "withdrawn"
|
|
assert items["other"].status != "withdrawn"
|
|
assert items["other"].source_check_status == "available"
|
|
|
|
record_scheduled_source_check(
|
|
db,
|
|
source_system="rf4db",
|
|
source_url="https://rf4db.com/ru/catches/matching",
|
|
status="available",
|
|
)
|
|
assert items["matching"].source_check_status == "available"
|
|
assert items["matching"].status == "withdrawn"
|
|
|
|
|
|
def test_auto_publication_requires_enabled_source(db: Session) -> None:
|
|
source = DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=False)
|
|
fish = Fish(slug="pike", name_ru="Щука")
|
|
waterbody = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
|
|
db.add_all([source, fish, waterbody])
|
|
db.flush()
|
|
now = datetime.now(timezone.utc)
|
|
db.add_all([
|
|
ExternalEntityAlias(source_system="rf4db", entity_type="fish", external_id="pike", external_name="Щука", fish=fish, updated_at=now),
|
|
ExternalEntityAlias(source_system="rf4db", entity_type="waterbody", external_id="test-lake", external_name="Тестовое озеро", waterbody=waterbody, updated_at=now),
|
|
])
|
|
db.commit()
|
|
|
|
stage_observations(db, [record() | {"weight_g": 5_000}])
|
|
|
|
item = db.scalar(select(ExternalObservation))
|
|
assert item is not None and item.status == "staged" and item.catch_report is None
|
|
|
|
|
|
def test_confirmed_aliases_are_suggestions_and_cannot_be_retargeted(db: Session) -> None:
|
|
source = DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=True)
|
|
pike = Fish(slug="pike", name_ru="Щука")
|
|
perch = Fish(slug="perch", name_ru="Окунь")
|
|
waterbody = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
|
|
db.add_all([source, pike, perch, waterbody])
|
|
db.commit()
|
|
stage_observations(db, [record(external_id="first")])
|
|
first = db.scalar(select(ExternalObservation).where(ExternalObservation.source_external_id == "first"))
|
|
assert first is not None
|
|
map_observation(db, first, pike, waterbody)
|
|
|
|
stage_observations(db, [record(external_id="second")])
|
|
second = db.scalar(select(ExternalObservation).where(ExternalObservation.source_external_id == "second"))
|
|
assert second is not None
|
|
suggested_fish, suggested_waterbody = suggest_aliases(db, second)
|
|
assert (suggested_fish.slug, suggested_waterbody.slug) == ("pike", "test-lake")
|
|
|
|
with pytest.raises(ExternalReviewError, match="confirmed fish alias"):
|
|
map_observation(db, second, perch, waterbody)
|
|
db.rollback()
|
|
alias = db.scalar(select(ExternalEntityAlias).where(ExternalEntityAlias.entity_type == "fish"))
|
|
assert alias is not None and alias.fish_id == pike.id
|
|
|
|
|
|
def test_parser_json_can_be_staged_without_losing_provenance(db: Session) -> None:
|
|
parsed = parse_rf4db_catches(FIXTURE.read_text(encoding="utf-8"))
|
|
payload = json.loads(json.dumps([asdict(item) for item in parsed], default=str))
|
|
|
|
assert stage_observations(db, payload) == (1, 0)
|
|
item = db.scalar(select(ExternalObservation))
|
|
assert item is not None
|
|
assert (item.source_system, item.fish_external_id, item.x, item.y) == ("rf4db", "pike", 71, 92)
|
|
|
|
|
|
@pytest.mark.parametrize(("fixture_name", "source_url", "parser"), [
|
|
("rf4map_point_sample.html", "https://rf4map.ru/points/275", parse_rf4map_point),
|
|
(
|
|
"rf4posts_spot_sample.html",
|
|
"https://rf4-posts.com/ru/spots/d0c6d9c6-4ebf-49a7-98a8-9a562553a8ee",
|
|
parse_rf4posts_spot,
|
|
),
|
|
])
|
|
def test_new_parser_json_can_be_staged(
|
|
db: Session, fixture_name: str, source_url: str, parser,
|
|
) -> None:
|
|
parsed = parser((FIXTURES / fixture_name).read_text(encoding="utf-8"), source_url=source_url)
|
|
payload = json.loads(json.dumps([asdict(item) for item in parsed], default=str))
|
|
|
|
created, updated = stage_observations(db, payload)
|
|
|
|
assert (created, updated) == (len(parsed), 0)
|
|
assert db.scalar(select(func.count()).select_from(ExternalObservation)) == len(parsed)
|
|
|
|
|
|
def test_invalid_source_rolls_back_caller_transaction(db: Session) -> None:
|
|
with pytest.raises(CommunityImportError, match="unsupported source_system"):
|
|
stage_observations(db, [record("unknown")])
|
|
db.rollback()
|
|
|
|
assert db.scalar(select(func.count()).select_from(ExternalObservation)) == 0
|
|
|
|
|
|
@pytest.mark.parametrize("change", [
|
|
{"source_url": "https://attacker.example/catch-1"},
|
|
{"x": 10_001},
|
|
{"weight_g": 3_000_001},
|
|
])
|
|
def test_invalid_provenance_and_ranges_are_rejected(db: Session, change: dict[str, object]) -> None:
|
|
with pytest.raises(CommunityImportError):
|
|
stage_observations(db, [record() | change])
|
|
db.rollback()
|