Spaces:
Running
Running
| """BDD step definitions for platform presence.""" | |
| from __future__ import annotations | |
| import asyncio | |
| from behave import given, then, when | |
| from app.services.isbn_catalog import classify_retailer_url | |
| from app.services.platform_presence import SCORED_PLATFORMS, scan_platform_presence | |
| def _run(coro): | |
| return asyncio.run(coro) | |
| def step_sample_book(context, key): | |
| book = context.sample_books[key] | |
| context.book = book | |
| def step_scan(context): | |
| book = context.book | |
| context.scan = _run( | |
| scan_platform_presence( | |
| title=book["title"], | |
| author=book["author"], | |
| isbn=book["isbn"], | |
| source_platform=book["platform"], | |
| source_url=book["url"], | |
| buy_url=book["url"], | |
| ) | |
| ) | |
| def step_platform_verified(context, platform_id): | |
| hit = next(p for p in context.scan.platforms if p.platform_id == platform_id) | |
| assert hit.status == "verified", f"{platform_id} status={hit.status}" | |
| assert hit.listing_url, f"{platform_id} missing URL" | |
| assert classify_retailer_url(hit.listing_url, platform_id) == "product" | |
| def step_no_search_verified(context): | |
| for p in context.scan.platforms: | |
| if p.status == "verified" and p.listing_url: | |
| kind = classify_retailer_url(p.listing_url, p.platform_id) | |
| assert kind == "product", f"{p.platform_id} verified with search URL" | |
| def step_min_listed(context, count): | |
| assert context.scan.listed_count >= count | |
| def step_load_platforms(context): | |
| context.platform_ids = [p["platform_id"] for p in SCORED_PLATFORMS] | |
| def step_platform_count(context, count): | |
| assert len(context.platform_ids) == count | |
| def step_platform_in_list(context, platform_id): | |
| assert platform_id in context.platform_ids | |