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
101
102
103
104
105
106
107
108
109
110
111
|
# MIT licensed
# Copyright (c) 2020, 2024 lilydjwg <lilydjwg@gmail.com>, et al.
import asyncio
import structlog
import os
from pathlib import Path
from typing import TYPE_CHECKING, Dict
if TYPE_CHECKING:
import tomli as tomllib
else:
try:
import tomllib
except ModuleNotFoundError:
import tomli as tomllib
import pytest
import pytest_asyncio
from nvchecker import core
from nvchecker import __main__ as main
from nvchecker.util import Entries, ResultData, RawResult
use_keyfile = False
async def run(
entries: Entries, max_concurrency: int = 20,
) -> Dict[str, str]:
task_sem = asyncio.Semaphore(max_concurrency)
result_q: asyncio.Queue[RawResult] = asyncio.Queue()
keyfile = os.environ.get('KEYFILE')
if use_keyfile and keyfile:
filepath = Path(keyfile)
keymanager = core.KeyManager(filepath)
else:
keymanager = core.KeyManager(None)
dispatcher = core.setup_httpclient()
entry_waiter = core.EntryWaiter()
futures = dispatcher.dispatch(
entries, task_sem, result_q,
keymanager, entry_waiter, 1, {},
)
oldvers: ResultData = {}
result_coro = core.process_result(oldvers, result_q, entry_waiter)
runner_coro = core.run_tasks(futures)
results, _has_failures = await main.run(result_coro, runner_coro)
return {k: r.version for k, r in results.items()}
@pytest_asyncio.fixture(scope="session")
async def get_version():
async def __call__(name, config):
entries = {name: config}
newvers = await run(entries)
return newvers.get(name)
return __call__
@pytest_asyncio.fixture(scope="session")
async def run_str():
async def __call__(str):
entries = tomllib.loads(str)
newvers = await run(entries)
return newvers.popitem()[1]
return __call__
@pytest_asyncio.fixture(scope="session")
async def run_str_multi():
async def __call__(str):
entries = tomllib.loads(str)
newvers = await run(entries)
return newvers
return __call__
@pytest.fixture(scope="session", autouse=True)
def raise_on_logger_msg():
def proc(logger, method_name, event_dict):
if method_name in ('warning', 'error'):
if 'exc_info' in event_dict:
exc = event_dict['exc_info']
if isinstance(exc, Exception):
raise exc
else: # exc_info=True
raise
if not event_dict['event'].startswith(('rate limited', 'no-result')):
raise RuntimeError(event_dict['event'])
return event_dict['event']
structlog.configure([proc])
def pytest_configure(config):
# register an additional marker
config.addinivalue_line(
'markers', 'needs_net: mark test to require Internet access',
)
@pytest.fixture
def keyfile():
global use_keyfile
if 'KEYFILE' not in os.environ:
pytest.skip('KEYFILE not set')
return
use_keyfile = True
yield
use_keyfile = False
|
