scrapfly-scrapers/wellfound-scraper/test.py at main · scrapfly/scrapfly-scrapers · GitHub

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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
from cerberus import Validator
import pytest
import wellfound
import pprint

pp = pprint.PrettyPrinter(indent=4)

# enable scrapfly cache
wellfound.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}")


def require_min_presence(items, key, min_perc=0.1):
    """check whether dataset contains items with some amount of non-null values for a given key"""
    count = sum(1 for item in items if item.get(key))
    if count < len(items) * min_perc:
        pytest.fail(
            f'inadequate presence of "{key}" field in dataset, only {count} out of {len(items)} items have it (expected {min_perc*100}%)'
        )


company_schema = {
    "__typename": {"type": "string"},
    "id": {"type": "string"},
    "name": {"type": "string"},
    "slug": {"type": "string"},
    "jobListingCounts": {
        "type": "dict",
        "schema": {
            "__typename": {"type": "string"},
            "roles": {
                "type": "list",
                "schema": {
                    "type": "dict",
                    "schema": {
                        "__typename": {"type": "string"},
                        "optionId": {"type": "string"},
                        "name": {"type": "string"},
                        "count": {"type": "integer"},
                    }
                }
            },
            "locations": {
                "type": "list",
                "schema": {
                    "type": "dict",
                    "schema": {
                        "__typename": {"type": "string"},
                        "optionId": {"type": "string"},
                        "name": {"type": "string"},
                        "count": {"type": "integer"},
                    }
                }
            }
        }
    }
}

search_schema = {
    "__typename": {"type": "string"},
    "id": {"type": "string"},
    "companySize": {"type": "string", "nullable": True},
    "highConcept": {"type": "string", "nullable": True},
    "logoUrl": {"type": "string"},
    "name": {"type": "string"},
    "slug": {"type": "string"},
}


@pytest.mark.asyncio
@pytest.mark.flaky(reruns=3, reruns_delay=30)
async def test_company_scraping():
    companies_data = await wellfound.scrape_companies(
        urls = [
            "https://www.wellfound.com/company/moxion-power-co/jobs"
        ]
    )
    validator = Validator(company_schema, allow_unknown=True)
    for item in companies_data:
        validate_or_fail(item, validator)
        assert len(companies_data) == 1


@pytest.mark.asyncio
@pytest.mark.flaky(reruns=3, reruns_delay=30)
async def test_search_scraping():
    search_data = await wellfound.scrape_search(
        role="python-developer", max_pages=2
    )
    validator = Validator(search_schema, allow_unknown=True)
    for item in search_data:
        validate_or_fail(item, validator)
    for k in search_schema:
        require_min_presence(search_data, k, min_perc=search_schema[k].get("min_presence", 0.1))
    assert len(search_data) >= 10