-
Notifications
You must be signed in to change notification settings - Fork 103
/
Copy pathtest.py
82 lines (71 loc) · 2.66 KB
/
test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from cerberus import Validator
import goat
import pytest
import pprint
pp = pprint.PrettyPrinter(indent=4)
# enable scrapfly cache
goat.BASE_CONFIG["cache"] = True
def validate_or_fail(item, validator):
if not validator.validate(item):
pp.pformat(item)
pytest.fail(f"Validation failed for item: {pp.pformat(item)}\nErrors: {validator.errors}")
product_schema = {
"brandName": {"type": "string"},
"color": {"type": "string"},
"designer": {"type": "string"},
"details": {"type": "string"},
"forAuction": {"type": "boolean", "nullable": True},
"id": {"type": "integer"},
"internalShot": {"type": "string"},
"maximumOfferCents": {"type": "integer"},
"midsole": {"type": "string"},
"minimumOfferCents": {"type": "integer"},
"name": {"type": "string"},
"productCategory": {"type": "string"},
"productType": {"type": "string"},
"silhouette": {"type": "string"},
"sizeBrand": {"type": "string"},
"sizeRange": {"type": "list", "schema": {"type": "float"}},
"sizeBrand": {"type": "string"},
"sku": {"type": "string"},
"slug": {"type": "string"},
"specialDisplayPriceCents": {"type": "integer"},
"specialType": {"type": "string"},
"status": {"type": "string"},
"upperMaterial": {"type": "string"},
}
search_schema = {
"id": {"type": "string"},
"sku": {"type": "string"},
"slug": {"type": "string"},
"color": {"type": "string"},
"category": {"type": "string"},
"image_url": {"type": "string"},
"product_type": {"type": "string"},
"release_date": {"type": "integer"},
"release_date_year": {"type": "integer"},
"retail_price_cents": {"type": "integer"},
"variation_id": {"type": "string"},
"box_condition": {"type": "string"},
"product_condition": {"type": "string"},
}
@pytest.mark.asyncio
async def test_product_scraping():
products_data = await goat.scrape_products(
urls=[
"https://www.goat.com/sneakers/air-jordan-3-retro-white-cement-reimagined-dn3707-100",
"https://www.goat.com/sneakers/travis-scott-x-air-jordan-1-retro-high-og-cd4487-100",
"https://www.goat.com/sneakers/travis-scott-x-wmns-air-jordan-1-low-og-olive-dz4137-106",
]
)
validator = Validator(product_schema, allow_unknown=True)
for item in products_data:
validate_or_fail(item, validator)
assert len(products_data) >= 1
@pytest.mark.asyncio
async def test_search_scraping():
search_data = await goat.scrape_search("pumar dark", max_pages=3)
validator = Validator(search_schema, allow_unknown=True)
for item in search_data:
validate_or_fail(item, validator)
assert len(search_data) >= 3